"""Unit tests for the analysis engine. Tests series completeness analysis, duplicate detection, and quality comparison. """ from datetime import datetime, timezone from pathlib import Path from vlm.analysis import analyze_series_completeness, compare_quality, detect_duplicates from vlm.models import MovieIdentity, SeriesIdentity, VideoFile def _movie(title="Movie", year=2020, **kw): return MovieIdentity( title=title, year=year, confidence=kw.pop("confidence", 0.9), needs_review=kw.pop("needs_review", False), original_filename=kw.pop("original_filename", f"{title}.{year}.mkv"), **kw, ) def _series(title="Show", season=1, episodes=None, **kw): if episodes is None: episodes = [1] return SeriesIdentity( title=title, season=season, episodes=episodes, confidence=kw.pop("confidence", 0.9), needs_review=kw.pop("needs_review", False), original_filename=kw.pop( "original_filename", f"{title.replace(' ', '.')}.S{season:02d}E01.mkv" if season is not None else f"{title.replace(' ', '.')}.E01.mkv", ), **kw, ) def _video(filename="file.mkv", size=1000, category="movie", **kw): return VideoFile( path=kw.pop("path", Path(f"/tmp/{filename}")), filename=filename, size_bytes=size, modified_timestamp=kw.pop("modified_timestamp", datetime(2023, 1, 1)), category=category, **kw, ) class TestSeriesCompletenessAnalysis: """Test series completeness analysis functionality.""" def test_detect_single_gap(self): """Test detection of a single missing episode.""" episodes = [ _series("Show Name", episodes=[1], original_filename="Show.Name.S01E01.mkv"), _series("Show Name", episodes=[2], original_filename="Show.Name.S01E02.mkv"), _series("Show Name", episodes=[4], original_filename="Show.Name.S01E04.mkv"), _series("Show Name", episodes=[5], original_filename="Show.Name.S01E05.mkv"), ] result = analyze_series_completeness(episodes) assert len(result) == 1 assert result[0].series_title == "Show Name" assert result[0].season == 1 assert result[0].episodes_found == [1, 2, 4, 5] assert result[0].episodes_missing == [3] def test_detect_multiple_gaps(self): """Test detection of multiple missing episodes.""" episodes = [ _series("Show Name", episodes=[1], original_filename="Show.Name.S01E01.mkv"), _series("Show Name", episodes=[3], original_filename="Show.Name.S01E03.mkv"), _series("Show Name", episodes=[5], original_filename="Show.Name.S01E05.mkv"), _series("Show Name", episodes=[7], original_filename="Show.Name.S01E07.mkv"), ] result = analyze_series_completeness(episodes) assert len(result) == 1 assert result[0].episodes_found == [1, 3, 5, 7] assert result[0].episodes_missing == [2, 4, 6] def test_no_gaps_returns_empty(self): """Test that complete seasons are not included in results.""" episodes = [ _series("Show Name", episodes=[1], original_filename="Show.Name.S01E01.mkv"), _series("Show Name", episodes=[2], original_filename="Show.Name.S01E02.mkv"), _series("Show Name", episodes=[3], original_filename="Show.Name.S01E03.mkv"), ] result = analyze_series_completeness(episodes) assert len(result) == 0 def test_multi_season_independence(self): """Test that gap detection for one season doesn't affect others.""" episodes = [ # Season 1 - has gap at episode 2 _series("Show Name", season=1, episodes=[1], original_filename="Show.Name.S01E01.mkv"), _series("Show Name", season=1, episodes=[3], original_filename="Show.Name.S01E03.mkv"), # Season 2 - complete _series("Show Name", season=2, episodes=[1], original_filename="Show.Name.S02E01.mkv"), _series("Show Name", season=2, episodes=[2], original_filename="Show.Name.S02E02.mkv"), # Season 3 - has gap at episode 5 _series("Show Name", season=3, episodes=[4], original_filename="Show.Name.S03E04.mkv"), _series("Show Name", season=3, episodes=[6], original_filename="Show.Name.S03E06.mkv"), ] result = analyze_series_completeness(episodes) # Should have 2 results (seasons 1 and 3 with gaps) assert len(result) == 2 # Find season 1 result season1 = next(r for r in result if r.season == 1) assert season1.episodes_found == [1, 3] assert season1.episodes_missing == [2] # Find season 3 result season3 = next(r for r in result if r.season == 3) assert season3.episodes_found == [4, 6] assert season3.episodes_missing == [5] def test_multi_episode_files(self): """Test handling of multi-episode files.""" episodes = [ _series("Show Name", episodes=[1, 2], original_filename="Show.Name.S01E01-E02.mkv"), _series("Show Name", episodes=[4], original_filename="Show.Name.S01E04.mkv"), ] result = analyze_series_completeness(episodes) assert len(result) == 1 assert result[0].episodes_found == [1, 2, 4] assert result[0].episodes_missing == [3] def test_different_series_separate_analysis(self): """Test that different series are analyzed separately.""" episodes = [ # Series A - has gap _series("Series A", episodes=[1], original_filename="Series.A.S01E01.mkv"), _series("Series A", episodes=[3], original_filename="Series.A.S01E03.mkv"), # Series B - complete _series("Series B", episodes=[1], original_filename="Series.B.S01E01.mkv"), _series("Series B", episodes=[2], original_filename="Series.B.S01E02.mkv"), ] result = analyze_series_completeness(episodes) # Only Series A should be in results assert len(result) == 1 assert result[0].series_title == "Series A" assert result[0].episodes_missing == [2] def test_skip_episodes_without_season(self): """Test that episodes with season=None are excluded from analysis.""" episodes = [ _series("Show Name", episodes=[1], original_filename="Show.Name.S01E01.mkv"), _series("Show Name", episodes=[2], original_filename="Show.Name.S01E02.mkv"), _series("Show Name", season=None, confidence=0.3, needs_review=True, original_filename="Show.Name.Episode.1.mkv"), ] result = analyze_series_completeness(episodes) # Should only analyze season 1, which is complete assert len(result) == 0 def test_skip_episodes_with_empty_episode_list(self): """Test that episodes with empty episode list are excluded from analysis.""" episodes = [ _series("Show Name", episodes=[1], original_filename="Show.Name.S01E01.mkv"), _series("Show Name", episodes=[3], original_filename="Show.Name.S01E03.mkv"), _series("Show Name", episodes=[], confidence=0.3, needs_review=True, original_filename="Show.Name.S01.mkv"), ] result = analyze_series_completeness(episodes) # Should detect gap at episode 2 assert len(result) == 1 assert result[0].episodes_missing == [2] def test_non_sequential_start(self): """Test gap detection when episodes don't start at 1.""" episodes = [ _series("Show Name", episodes=[5], original_filename="Show.Name.S01E05.mkv"), _series("Show Name", episodes=[6], original_filename="Show.Name.S01E06.mkv"), _series("Show Name", episodes=[8], original_filename="Show.Name.S01E08.mkv"), ] result = analyze_series_completeness(episodes) # Should detect gap at episode 7 in range [5, 8] assert len(result) == 1 assert result[0].episodes_found == [5, 6, 8] assert result[0].episodes_missing == [7] def test_empty_input(self): """Test handling of empty episode list.""" result = analyze_series_completeness([]) assert len(result) == 0 def test_single_episode_no_gap(self): """Test that a single episode has no gaps.""" episodes = [ _series("Show Name", episodes=[1], original_filename="Show.Name.S01E01.mkv"), ] result = analyze_series_completeness(episodes) # Single episode has no gaps assert len(result) == 0 class TestDuplicateDetection: """Test duplicate detection functionality.""" def test_detect_movie_duplicates(self): """Test detection of duplicate movies with identical title and year.""" now = datetime.now(timezone.utc) identities = [ _movie("The Matrix", 1999, original_filename="The.Matrix.1999.1080p.mkv"), _movie("The Matrix", 1999, original_filename="The.Matrix.1999.720p.mkv"), _movie("Inception", 2010), ] files = [ _video("The.Matrix.1999.1080p.mkv", 2_000_000_000, modified_timestamp=now, resolution="1920x1080", codec="h264"), _video("The.Matrix.1999.720p.mkv", 1_000_000_000, modified_timestamp=now, resolution="1280x720", codec="h264"), _video("Inception.2010.mkv", 1_500_000_000, modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) # Should find one duplicate group (The Matrix) assert len(result) == 1 assert isinstance(result[0].identity, MovieIdentity) assert result[0].identity.title == "The Matrix" assert result[0].identity.year == 1999 assert len(result[0].files) == 2 assert len(result[0].quality_comparison) == 2 def test_detect_series_duplicates(self): """Test detection of duplicate series episodes.""" now = datetime.now(timezone.utc) identities = [ _series("Breaking Bad", episodes=[1], original_filename="Breaking.Bad.S01E01.1080p.mkv"), _series("Breaking Bad", episodes=[1], original_filename="Breaking.Bad.S01E01.720p.mkv"), _series("Breaking Bad", episodes=[2], original_filename="Breaking.Bad.S01E02.mkv"), ] files = [ _video("Breaking.Bad.S01E01.1080p.mkv", 1_500_000_000, "series", modified_timestamp=now, resolution="1920x1080"), _video("Breaking.Bad.S01E01.720p.mkv", 800_000_000, "series", modified_timestamp=now, resolution="1280x720"), _video("Breaking.Bad.S01E02.mkv", 1_200_000_000, "series", modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) # Should find one duplicate group (S01E01) assert len(result) == 1 assert isinstance(result[0].identity, SeriesIdentity) assert result[0].identity.title == "Breaking Bad" assert result[0].identity.season == 1 assert 1 in result[0].identity.episodes assert len(result[0].files) == 2 def test_no_duplicates(self): """Test that unique files are not flagged as duplicates.""" now = datetime.now(timezone.utc) identities = [ _movie("Movie A", 2020), _movie("Movie B", 2021), ] files = [ _video("Movie.A.2020.mkv", 1_000_000_000, modified_timestamp=now), _video("Movie.B.2021.mkv", 1_000_000_000, modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) assert len(result) == 0 def test_skip_movies_without_year(self): """Test that movies without year are excluded from duplicate detection.""" now = datetime.now(timezone.utc) identities = [ _movie("Unknown Movie", year=None, confidence=0.3, needs_review=True, original_filename="Unknown.Movie.mkv"), _movie("Unknown Movie", year=None, confidence=0.3, needs_review=True, original_filename="Unknown.Movie.2.mkv"), ] files = [ _video("Unknown.Movie.mkv", 1_000_000_000, modified_timestamp=now), _video("Unknown.Movie.2.mkv", 1_000_000_000, modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) # Should not detect duplicates for files needing review assert len(result) == 0 def test_skip_series_without_season(self): """Test that series without season are excluded from duplicate detection.""" now = datetime.now(timezone.utc) identities = [ _series("Unknown Show", season=None, confidence=0.3, needs_review=True, original_filename="Unknown.Show.E01.mkv"), _series("Unknown Show", season=None, confidence=0.3, needs_review=True, original_filename="Unknown.Show.Episode.1.mkv"), ] files = [ _video("Unknown.Show.E01.mkv", 1_000_000_000, "series", modified_timestamp=now), _video("Unknown.Show.Episode.1.mkv", 1_000_000_000, "series", modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) assert len(result) == 0 def test_skip_series_with_empty_episodes(self): """Test that series with empty episode list are excluded.""" now = datetime.now(timezone.utc) identities = [ _series("Show Name", episodes=[], confidence=0.3, needs_review=True, original_filename="Show.Name.S01.mkv"), _series("Show Name", episodes=[], confidence=0.3, needs_review=True, original_filename="Show.Name.Season.1.mkv"), ] files = [ _video("Show.Name.S01.mkv", 1_000_000_000, "series", modified_timestamp=now), _video("Show.Name.Season.1.mkv", 1_000_000_000, "series", modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) assert len(result) == 0 def test_quality_comparison_includes_all_metadata(self): """Test that quality comparison includes all available metadata.""" now = datetime.now(timezone.utc) identities = [ _movie("Test Movie", 2020, original_filename="Test.Movie.2020.1080p.mkv"), _movie("Test Movie", 2020, original_filename="Test.Movie.2020.720p.mkv"), ] files = [ _video("Test.Movie.2020.1080p.mkv", 2_000_000_000, modified_timestamp=now, resolution="1920x1080", codec="h264", duration_seconds=7200.0, bitrate_kbps=5000), _video("Test.Movie.2020.720p.mkv", 1_000_000_000, modified_timestamp=now, resolution="1280x720", codec="h264", duration_seconds=7200.0, bitrate_kbps=2500), ] result = detect_duplicates(list(zip(identities, files))) assert len(result) == 1 comparison = result[0].quality_comparison assert len(comparison) == 2 # Check first file comparison data assert comparison[0]['filename'] == "Test.Movie.2020.1080p.mkv" assert comparison[0]['size_bytes'] == 2000000000 assert comparison[0]['resolution'] == "1920x1080" assert comparison[0]['codec'] == "h264" assert comparison[0]['duration_seconds'] == 7200.0 assert comparison[0]['bitrate_kbps'] == 5000 # Check second file comparison data assert comparison[1]['filename'] == "Test.Movie.2020.720p.mkv" assert comparison[1]['size_bytes'] == 1000000000 assert comparison[1]['resolution'] == "1280x720" def test_multi_episode_file_duplicates(self): """Test duplicate detection for multi-episode files.""" now = datetime.now(timezone.utc) identities = [ _series("Show", episodes=[1, 2], original_filename="Show.S01E01-E02.mkv"), _series("Show", episodes=[1], original_filename="Show.S01E01.mkv"), _series("Show", episodes=[2], original_filename="Show.S01E02.mkv"), ] files = [ _video("Show.S01E01-E02.mkv", 2_000_000_000, "series", modified_timestamp=now), _video("Show.S01E01.mkv", 1_000_000_000, "series", modified_timestamp=now), _video("Show.S01E02.mkv", 1_000_000_000, "series", modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) # Should find duplicates for both E01 and E02 assert len(result) == 2 def test_different_years_not_duplicates(self): """Test that same title with different years are not duplicates.""" now = datetime.now(timezone.utc) identities = [ _movie("The Thing", 1982), _movie("The Thing", 2011), ] files = [ _video("The.Thing.1982.mkv", 1_000_000_000, modified_timestamp=now), _video("The.Thing.2011.mkv", 1_000_000_000, modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) assert len(result) == 0 def test_different_seasons_not_duplicates(self): """Test that same series/episode in different seasons are not duplicates.""" now = datetime.now(timezone.utc) identities = [ _series("Show", season=1, original_filename="Show.S01E01.mkv"), _series("Show", season=2, original_filename="Show.S02E01.mkv"), ] files = [ _video("Show.S01E01.mkv", 1_000_000_000, "series", modified_timestamp=now), _video("Show.S02E01.mkv", 1_000_000_000, "series", modified_timestamp=now), ] result = detect_duplicates(list(zip(identities, files))) assert len(result) == 0 class TestQualityComparison: """Test quality comparison functionality.""" def test_compare_quality_with_all_metadata(self): """Test quality comparison with all metadata available.""" now = datetime.now(timezone.utc) files = [ _video("file1.mkv", 2_000_000_000, modified_timestamp=now, resolution="1920x1080", codec="h264", duration_seconds=7200.0, bitrate_kbps=5000), _video("file2.mkv", 1_000_000_000, modified_timestamp=now, resolution="1280x720", codec="h265", duration_seconds=7200.0, bitrate_kbps=2500), ] result = compare_quality(files) assert len(result) == 2 assert result[0]['filename'] == "file1.mkv" assert result[0]['size_bytes'] == 2000000000 assert result[0]['resolution'] == "1920x1080" assert result[0]['codec'] == "h264" assert result[0]['duration_seconds'] == 7200.0 assert result[0]['bitrate_kbps'] == 5000 assert result[1]['filename'] == "file2.mkv" assert result[1]['size_bytes'] == 1000000000 assert result[1]['resolution'] == "1280x720" assert result[1]['codec'] == "h265" def test_compare_quality_with_partial_metadata(self): """Test quality comparison when some metadata is missing.""" now = datetime.now(timezone.utc) files = [ _video("file1.mkv", 2_000_000_000, modified_timestamp=now, resolution="1920x1080"), _video("file2.mkv", 1_000_000_000, modified_timestamp=now), ] result = compare_quality(files) assert len(result) == 2 assert result[0]['filename'] == "file1.mkv" assert result[0]['size_bytes'] == 2000000000 assert result[0]['resolution'] == "1920x1080" assert 'codec' not in result[0] assert 'duration_seconds' not in result[0] assert 'bitrate_kbps' not in result[0] assert result[1]['filename'] == "file2.mkv" assert result[1]['size_bytes'] == 1000000000 assert 'resolution' not in result[1] assert 'codec' not in result[1] def test_compare_quality_empty_list(self): """Test quality comparison with empty file list.""" result = compare_quality([]) assert len(result) == 0 def test_compare_quality_single_file(self): """Test quality comparison with single file.""" now = datetime.now(timezone.utc) files = [ _video("file.mkv", 1_500_000_000, modified_timestamp=now, resolution="1920x1080", codec="h264"), ] result = compare_quality(files) assert len(result) == 1 assert result[0]['filename'] == "file.mkv" assert result[0]['size_bytes'] == 1500000000