Aniworld/tests/unit/test_serie_scanner_scan_dedup.py

"""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}"
    )