"""Regression test for ``SerieScanner.scan_single_series``. The previous implementation ``extend````ed the in-memory ``episodeDict`` for series already present in the scanner's ``keyDict`` — every rescan of the same series appended the new missing-episode list on top of the existing one, so the dict grew with duplicates across rescans. Those duplicates then propagated through ``_update_series_in_db`` into the ``episodes`` table. This test exercises the real ``scan_single_series`` method (not a mock) to assert that two rescans of the same series produce a canonical (deduplicated) ``episodeDict``, not an accumulated one. """ from __future__ import annotations from pathlib import Path from types import SimpleNamespace import pytest from src.server.SerieScanner import SerieScanner class _StubLoader: """A minimal loader that returns whatever ``missing_episodes`` the test wants. Replaces the real loader on a ``SerieScanner`` instance via ``monkeypatch.setattr``.""" def __init__(self, missing_per_call: list[dict]): self._missing_per_call = list(missing_per_call) self._call_index = 0 def get_season_episode_count(self, key): # The real loader returns the total episode count per # season; ``scan_single_series`` doesn't actually use it # (it calls ``__get_missing_episodes_and_season`` directly), # but the stub keeps the call site safe. max_seen = 0 for m in self._missing_per_call: for season, eps in m.items(): if eps: max_seen = max(max_seen, max(eps)) return {1: max_seen or 1} def is_language(self, season, ep, key): return True def next_missing(self) -> dict: if self._call_index >= len(self._missing_per_call): return {} result = self._missing_per_call[self._call_index] self._call_index += 1 return result def _make_scanner( tmp_path: Path, missing_per_call: list[dict], ) -> tuple[SerieScanner, _StubLoader]: """Build a real ``SerieScanner`` with the private ``__get_missing_episodes_and_season`` method replaced by a stub that returns the next ``missing_episodes`` dict on each call. Everything else (events, directory) is stubbed so the test runs without filesystem or scheduler setup.""" scanner = SerieScanner.__new__(SerieScanner) loader = _StubLoader(missing_per_call) scanner.loader = loader # type: ignore[assignment] scanner.keyDict = {} # ``self.directory`` is the attribute ``scan_single_series`` # reads at line 734. ``scan_single_series`` checks # ``os.path.isdir(folder_path)`` and, if the folder does not # exist, treats the scan as "no MP4 files on disk". Use a path # whose subdirectories do not exist so the scan takes the # empty-mp4-files branch without us having to populate any # filesystem state. scanner.directory = str(tmp_path) scanner.directory_to_search = tmp_path scanner.events = SimpleNamespace( on_progress=lambda *a, **k: None, on_completion=lambda *a, **k: None, on_error=lambda *a, **k: None, ) def fake_get_missing_episodes_and_season(key, mp4_files): return loader.next_missing(), "aniworld.to" scanner._SerieScanner__get_missing_episodes_and_season = ( # type: ignore[attr-defined] fake_get_missing_episodes_and_season ) return scanner, loader def test_scan_single_series_replaces_not_extends(tmp_path): """Two rescans of the same series with the same missing episodes must produce a canonical (deduplicated) episodeDict — not a list with duplicates from accumulation.""" canonical = {1: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]} scanner, _ = _make_scanner( tmp_path, # Same missing-episode list reported on each scan. missing_per_call=[canonical, canonical], ) # First scan: key not in keyDict, so the else branch fires # and the cache is set to the missing-episode list. scanner.keyDict.clear() result_first = scanner.scan_single_series( key="erased", folder="Erased" ) assert result_first == canonical cached = scanner.keyDict["erased"].episodeDict assert cached == canonical # Second scan: key IS in keyDict, so the if branch fires. # Before the fix, this would extend the cached list with # [1..12] again, producing {1: [1..12, 1..12]}. After the # fix, the cache is replaced, not extended. result_second = scanner.scan_single_series( key="erased", folder="Erased" ) assert result_second == canonical cached_second = scanner.keyDict["erased"].episodeDict assert cached_second == canonical, ( "second scan extended the dict instead of replacing it: " f"{cached_second}" ) # Defensive: the dict has no duplicates within any season. for season, eps in cached_second.items(): assert len(eps) == len(set(eps)), ( f"season {season} has duplicate episode numbers: {eps}" ) def test_scan_single_series_resets_when_becomes_complete(tmp_path): """When a rescan finds the series is complete (no missing episodes), the episodeDict must be empty — extending the previous dict would leave stale entries behind.""" scanner, _ = _make_scanner( tmp_path, # First scan: missing 1..12. Second scan: nothing missing. missing_per_call=[ {1: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]}, {}, ], ) scanner.keyDict.clear() scanner.scan_single_series(key="complete-me", folder="Complete Me") assert scanner.keyDict["complete-me"].episodeDict == { 1: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] } scanner.scan_single_series(key="complete-me", folder="Complete Me") assert scanner.keyDict["complete-me"].episodeDict == {}, ( "rescan with no missing episodes must reset the dict, " f"got: {scanner.keyDict['complete-me'].episodeDict}" ) def test_scan_single_series_dedupes_within_a_single_call(tmp_path): """If the loader itself returns duplicate episode numbers in ``missing_episodes`` (a buggy upstream loader), the scanner must still produce a canonical dict — defense in depth on top of the loader and the read-boundary dedup in ``AnimeSeries.episodeDict``.""" scanner, _ = _make_scanner( tmp_path, # The loader returns the same episode numbers multiple # times within a single call. missing_per_call=[{1: [1, 1, 2, 2, 3, 3, 3, 4]}], ) scanner.keyDict.clear() scanner.scan_single_series(key="buggy", folder="Buggy") cached = scanner.keyDict["buggy"].episodeDict assert cached == {1: [1, 2, 3, 4]}, ( f"single-scan dedup failed: {cached}" )