ffx/tests/unit/test_shifted_season_controller.py

from __future__ import annotations

import logging
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch


SRC_ROOT = Path(__file__).resolve().parents[2] / "src"

if str(SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(SRC_ROOT))


from ffx.database import databaseContext  # noqa: E402
from ffx.model.pattern import Pattern  # noqa: E402
from ffx.model.track import Track  # noqa: E402
from ffx.show_controller import ShowController  # noqa: E402
from ffx.show_descriptor import ShowDescriptor  # noqa: E402
from ffx.shifted_season_controller import ShiftedSeasonController  # noqa: E402
from ffx.track_type import TrackType  # noqa: E402


class StaticConfig:
    def __init__(self, data: dict | None = None):
        self._data = data or {}

    def getData(self):
        return self._data


def make_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.handlers = []
    logger.setLevel(logging.DEBUG)
    logger.propagate = False
    logger.addHandler(logging.NullHandler())
    return logger


def make_context(database_path: Path) -> dict:
    return {
        "logger": make_logger(f"ffx-test-shifted-{database_path.stem}"),
        "config": StaticConfig(),
        "database": databaseContext(str(database_path)),
    }


class ShiftedSeasonControllerTests(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory()
        self.database_path = Path(self.tempdir.name) / "shifted-season-test.db"
        self.context = make_context(self.database_path)
        self.show_controller = ShowController(self.context)
        self.shifted_season_controller = ShiftedSeasonController(self.context)

    def tearDown(self):
        self.context["database"]["engine"].dispose()
        self.tempdir.cleanup()

    def add_show(self, show_id: int, name: str = "Demo Show"):
        self.show_controller.updateShow(
            ShowDescriptor(id=show_id, name=name, year=2000 + show_id)
        )

    def add_pattern(self, show_id: int, expression: str) -> int:
        self.add_show(show_id)
        Session = self.context["database"]["session"]
        session = Session()
        try:
            pattern = Pattern(show_id=show_id, pattern=expression)
            session.add(pattern)
            session.flush()
            session.add(
                Track(
                    pattern_id=pattern.getId(),
                    track_type=TrackType.VIDEO.index(),
                    codec_name="h264",
                    index=0,
                    source_index=0,
                    disposition_flags=0,
                    audio_layout=0,
                )
            )
            session.commit()
            return pattern.getId()
        finally:
            session.close()

    def test_shift_season_uses_show_mapping_when_no_pattern_mapping_exists(self):
        pattern_id = self.add_pattern(1, r"^demo_(s[0-9]+e[0-9]+)\.mkv$")
        self.shifted_season_controller.addShiftedSeason(
            showId=1,
            shiftedSeasonObj={
                "original_season": 1,
                "first_episode": 1,
                "last_episode": 10,
                "season_offset": 2,
                "episode_offset": 5,
            },
        )

        with patch.object(self.context["logger"], "info") as mocked_info:
            shifted_season, shifted_episode = self.shifted_season_controller.shiftSeason(
                showId=1,
                patternId=pattern_id,
                season=1,
                episode=3,
            )

        self.assertEqual((3, 8), (shifted_season, shifted_episode))
        mocked_info.assert_called_once_with(
            "Setting season shift 1/3 -> 3/8 from show"
        )

    def test_shift_season_prefers_pattern_mapping_over_show_mapping(self):
        pattern_id = self.add_pattern(1, r"^demo_(s[0-9]+e[0-9]+)\.mkv$")
        self.shifted_season_controller.addShiftedSeason(
            showId=1,
            shiftedSeasonObj={
                "original_season": 1,
                "first_episode": 1,
                "last_episode": 10,
                "season_offset": 2,
                "episode_offset": 5,
            },
        )
        self.shifted_season_controller.addShiftedSeason(
            patternId=pattern_id,
            shiftedSeasonObj={
                "original_season": 1,
                "first_episode": 1,
                "last_episode": 10,
                "season_offset": 1,
                "episode_offset": -2,
            },
        )

        with patch.object(self.context["logger"], "info") as mocked_info:
            shifted_season, shifted_episode = self.shifted_season_controller.shiftSeason(
                showId=1,
                patternId=pattern_id,
                season=1,
                episode=3,
            )

        self.assertEqual((2, 1), (shifted_season, shifted_episode))
        mocked_info.assert_called_once_with(
            "Setting season shift 1/3 -> 2/1 from pattern"
        )

    def test_shift_season_pattern_zero_offsets_override_show_mapping_to_identity(self):
        pattern_id = self.add_pattern(1, r"^demo_(s[0-9]+e[0-9]+)\.mkv$")
        self.shifted_season_controller.addShiftedSeason(
            showId=1,
            shiftedSeasonObj={
                "original_season": 1,
                "first_episode": 1,
                "last_episode": 10,
                "season_offset": 2,
                "episode_offset": 5,
            },
        )
        self.shifted_season_controller.addShiftedSeason(
            patternId=pattern_id,
            shiftedSeasonObj={
                "original_season": 1,
                "first_episode": 1,
                "last_episode": 10,
                "season_offset": 0,
                "episode_offset": 0,
            },
        )

        with patch.object(self.context["logger"], "info") as mocked_info:
            shifted_season, shifted_episode = self.shifted_season_controller.shiftSeason(
                showId=1,
                patternId=pattern_id,
                season=1,
                episode=3,
            )

        self.assertEqual((1, 3), (shifted_season, shifted_episode))
        mocked_info.assert_not_called()

    def test_shift_season_falls_back_to_identity_when_no_rule_matches(self):
        pattern_id = self.add_pattern(1, r"^demo_(s[0-9]+e[0-9]+)\.mkv$")

        with patch.object(self.context["logger"], "info") as mocked_info:
            shifted_season, shifted_episode = self.shifted_season_controller.shiftSeason(
                showId=1,
                patternId=pattern_id,
                season=4,
                episode=20,
            )

        self.assertEqual((4, 20), (shifted_season, shifted_episode))
        mocked_info.assert_not_called()


if __name__ == "__main__":
    unittest.main()