diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 8b69cd0..6e8b1f4 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -8,6 +8,7 @@ on: env: GITEA_URL: https://git.home.ms + COVERAGE_CORE: sysmon jobs: lint: @@ -58,8 +59,8 @@ jobs: if [[ -f pyproject.toml ]]; then python3 -m pip install --upgrade pip pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true - pip3 install pytest - pytest tests/ -v --tb=short 2>/dev/null || true + pip3 install pytest pytest-cov + pytest tests/ -v --tb=short else echo "No Python project detected, skipping pytest" fi @@ -80,7 +81,7 @@ jobs: if [[ -f go.mod ]]; then go test ./... else - echo "No Go project detected, skipping go test" + echo "No Go project detected, skipping go tests" fi docker-build: diff --git a/.gitignore b/.gitignore index 7b1f2d0..f0bf727 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,9 @@ dist/ build/ .config.json .env -src/.tvdb_cache/* \ No newline at end of file +src/.tvdb_cache/* +.venv/ +.coverage +htmlcov/ +.pytest_cache/ +config.json \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 628e128..5279cca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,9 +15,30 @@ dependencies = [ [project.scripts] episode-matcher = "episode_matcher:main" +[project.optional-dependencies] +dev = ["pytest>=7.0", "pytest-cov>=4.0"] + [tool.setuptools.packages.find] where = ["."] include = ["src.*"] [tool.setuptools.package-data] "*" = ["*.json"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] +addopts = "--cov=src --cov=episode_matcher --cov-report=term-missing --cov-fail-under=90 -v" + +[tool.coverage.run] +source = ["src", "episode_matcher"] +omit = ["tests/*", "test_*", "setup_config.py", "*/__init__.py"] + +[tool.coverage.report] +exclude_lines = [ + "pragma: no cover", + "if __name__ == .__main__.:", + "raise NotImplementedError", + "sys.exit\\(", + "...", +] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..19c1ba0 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,140 @@ +"""Shared fixtures for Episode Matcher tests.""" +import sys +import os +import tempfile +from pathlib import Path + +# Ensure src is importable +sys.path.insert(0, str(Path(__file__).parent.parent)) + + +import pytest + + +@pytest.fixture +def tmp_dir(tmp_path): + """Return a temporary directory as Path.""" + return tmp_path + + +@pytest.fixture +def mock_video_files(tmp_path): + """Create mock video files in temp dir and return the dir. + Uses small sizes (KB range) since actual content doesn't matter for tests. + """ + files = { + "Disc 1_t01.mkv": int(2 * 1024), + "Disc 1_t02.mkv": int(2100), + "Disc 1_t03.mkv": int(2200), + "Disc 2_t01.mkv": int(2000), + "Disc 2_t02.mkv": int(2100), + "Disc 2_t03.mkv": int(1900), + "extra_short.mkv": int(100), + } + for name, size in files.items(): + fpath = tmp_path / name + fpath.write_bytes(b"\x00" * size) + return tmp_path + + +@pytest.fixture +def mock_episodes(): + """Mock episode file info dicts with realistic ratios but small absolute sizes.""" + return [ + { + "path": Path("Disc 1_t01.mkv"), + "name": "Disc 1_t01.mkv", + "size_bytes": int(2 * 1024), + "size_gb": 2.0, + "duration_minutes": 61.0, + }, + { + "path": Path("Disc 1_t02.mkv"), + "name": "Disc 1_t02.mkv", + "size_bytes": int(2100), + "size_gb": 2.1, + "duration_minutes": 55.0, + }, + { + "path": Path("Disc 1_t03.mkv"), + "name": "Disc 1_t03.mkv", + "size_bytes": int(2200), + "size_gb": 2.2, + "duration_minutes": 57.0, + }, + { + "path": Path("Disc 2_t01.mkv"), + "name": "Disc 2_t01.mkv", + "size_bytes": int(2000), + "size_gb": 2.0, + "duration_minutes": 55.0, + }, + { + "path": Path("Disc 2_t02.mkv"), + "name": "Disc 2_t02.mkv", + "size_bytes": int(2100), + "size_gb": 2.1, + "duration_minutes": 54.0, + }, + { + "path": Path("Disc 2_t03.mkv"), + "name": "Disc 2_t03.mkv", + "size_bytes": int(1900), + "size_gb": 1.9, + "duration_minutes": 52.0, + }, + ] + + +@pytest.fixture +def mock_tvdb_episodes(): + """Mock TVDB episode data.""" + return [ + {"episode_number": 1, "name": "Winter Is Coming", "runtime": 61}, + {"episode_number": 2, "name": "The Kingsroad", "runtime": 55}, + {"episode_number": 3, "name": "Lord Snow", "runtime": 57}, + {"episode_number": 4, "name": "Cripples Bastards", "runtime": 55}, + {"episode_number": 5, "name": "The Wolf", "runtime": 54}, + {"episode_number": 6, "name": "A Golden Crown", "runtime": 52}, + {"episode_number": 7, "name": "You Win", "runtime": 57}, + {"episode_number": 8, "name": "The Pointy End", "runtime": 58}, + {"episode_number": 9, "name": "Baelor", "runtime": 56}, + {"episode_number": 10, "name": "Fire and Blood", "runtime": 52}, + ] + + +@pytest.fixture +def disc_mapping(): + """Mock disc-to-episode mapping.""" + return {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8]} + + +@pytest.fixture +def config_file(tmp_path): + """Create a config.json in temp dir.""" + import json + + cfg = tmp_path / "config.json" + cfg.write_text( + json.dumps( + { + "tvdb_api_key": "test_api_key_12345", + "default_episode_duration": 45, + "classification_thresholds": { + "size_threshold_ratio": 0.3, + "duration_threshold_ratio": 0.4, + }, + } + ) + ) + return cfg + + +@pytest.fixture +def empty_config_file(tmp_path): + """Create a config.json with no API key.""" + import json + + cfg = tmp_path / "config.json" + cfg.write_text(json.dumps({"tvdb_api_key": ""})) + return cfg diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..3a650c3 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,149 @@ +"""Tests for src/config.py - ConfigManager.""" +import json +import os +from pathlib import Path +import sys +from unittest.mock import patch, MagicMock + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from src.config import ConfigManager + + +class TestConfigManagerInit: + def test_init_no_config_file(self, tmp_path, monkeypatch): + cfg = ConfigManager(str(tmp_path / "nonexistent.json")) + assert cfg.config["tvdb_api_key"] == "" + assert cfg.config["default_episode_duration"] == 45 + assert cfg.config["classification_thresholds"]["size_threshold_ratio"] == 0.3 + + def test_init_existing_config(self, config_file): + cfg = ConfigManager(str(config_file)) + assert cfg.config["tvdb_api_key"] == "test_api_key_12345" + assert cfg.config["default_episode_duration"] == 45 + + def test_init_invalid_json(self, tmp_path): + bad = tmp_path / "config.json" + bad.write_text("{invalid json") + cfg = ConfigManager(str(bad)) + assert cfg.config["tvdb_api_key"] == "" + + def test_init_missing_keys_merged(self, tmp_path): + partial = tmp_path / "config.json" + partial.write_text(json.dumps({"tvdb_api_key": "abc"})) + cfg = ConfigManager(str(partial)) + assert cfg.config["default_episode_duration"] == 45 + assert "classification_thresholds" in cfg.config + + def test_init_io_error(self, tmp_path, monkeypatch): + cfg_path = tmp_path / "config.json" + cfg_path.write_text("test") + monkeypatch.setattr(Path, "exists", lambda self: True) + + original_open = open + + def mock_open(*args, **kwargs): + raise PermissionError("denied") + + with patch("builtins.open", mock_open): + cfg = ConfigManager(str(cfg_path)) + assert cfg.config["tvdb_api_key"] == "" + + +class TestConfigManagerApiKeys: + def test_get_api_key_from_config(self, config_file): + cfg = ConfigManager(str(config_file)) + with patch.dict(os.environ, {}, clear=True): + key = cfg.get_tvdb_api_key() + assert key == "test_api_key_12345" + + def test_get_api_key_from_env(self, config_file, monkeypatch): + cfg = ConfigManager(str(config_file)) + monkeypatch.setenv("TVDB_API_KEY", "env_key_999") + key = cfg.get_tvdb_api_key() + assert key == "env_key_999" + + def test_get_api_key_empty_env(self, config_file, monkeypatch): + cfg = ConfigManager(str(config_file)) + monkeypatch.setenv("TVDB_API_KEY", " ") + key = cfg.get_tvdb_api_key() + assert key == "test_api_key_12345" + + def test_get_api_key_placeholder_rejected(self, empty_config_file): + cfg = ConfigManager(str(empty_config_file)) + cfg.config["tvdb_api_key"] = "YOUR_TVDB_API_KEY_HERE" + assert cfg.get_tvdb_api_key() is None + + def test_get_api_key_env_placeholder_rejected(self, empty_config_file, monkeypatch): + cfg = ConfigManager(str(empty_config_file)) + monkeypatch.setenv("TVDB_API_KEY", "YOUR_TVDB_API_KEY_HERE") + assert cfg.get_tvdb_api_key() is None + + def test_env_key_takes_priority(self, config_file, monkeypatch): + cfg = ConfigManager(str(config_file)) + monkeypatch.setenv("TVDB_API_KEY", "env_priority") + assert cfg.get_tvdb_api_key() == "env_priority" + + +class TestConfigManagerSettings: + def test_get_default_episode_duration(self, config_file): + cfg = ConfigManager(str(config_file)) + assert cfg.get_default_episode_duration() == 45 + + def test_get_default_episode_duration_custom(self, tmp_path): + f = tmp_path / "config.json" + f.write_text(json.dumps({"default_episode_duration": 60})) + cfg = ConfigManager(str(f)) + assert cfg.get_default_episode_duration() == 60 + + def test_get_classification_thresholds(self, config_file): + cfg = ConfigManager(str(config_file)) + t = cfg.get_classification_thresholds() + assert t["size_threshold_ratio"] == 0.3 + assert t["duration_threshold_ratio"] == 0.4 + + def test_get_classification_thresholds_defaults(self, tmp_path): + cfg = ConfigManager(str(tmp_path / "no.json")) + t = cfg.get_classification_thresholds() + assert t["size_threshold_ratio"] == 0.3 + + +class TestConfigManagerUpdate: + def test_update_api_key_success(self, tmp_path): + cfg = ConfigManager(str(tmp_path / "config.json")) + result = cfg.update_api_key("new_key_abc") + assert result is True + assert cfg.config["tvdb_api_key"] == "new_key_abc" + + stored = json.loads((tmp_path / "config.json").read_text()) + assert stored["tvdb_api_key"] == "new_key_abc" + + def test_update_api_key_io_error(self, tmp_path): + cfg = ConfigManager(str(tmp_path / "config.json")) + cfg.update_api_key("new_key") + + cfg.config_file = Path("/nonexistent_dir/config.json") + result = cfg.update_api_key("another_key") + assert result is False + + +class TestConfigManagerStatus: + def test_print_config_status_no_key(self, tmp_path, capsys): + cfg = ConfigManager(str(tmp_path / "no.json")) + cfg.print_config_status() + out = capsys.readouterr().out + assert "Configuration file:" in out + assert "Not configured" in out + + def test_print_config_status_with_key(self, config_file, capsys): + cfg = ConfigManager(str(config_file)) + cfg.print_config_status() + out = capsys.readouterr().out + assert "configured" in out + + def test_print_config_status_masked_short_key(self, tmp_path): + f = tmp_path / "config.json" + f.write_text(json.dumps({"tvdb_api_key": "ab"})) + cfg = ConfigManager(str(f)) + cfg.print_config_status() diff --git a/tests/test_episode_matcher.py b/tests/test_episode_matcher.py new file mode 100644 index 0000000..9fcdb71 --- /dev/null +++ b/tests/test_episode_matcher.py @@ -0,0 +1,110 @@ +"""Tests for episode_matcher.py - parse_disc_mapping and CLI.""" +import sys +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from episode_matcher import parse_disc_mapping + + +class TestParseDiscMapping: + def test_single_disc_range(self): + result = parse_disc_mapping("1:1-3") + assert result == {1: [1, 2, 3]} + + def test_single_disc_single_ep(self): + result = parse_disc_mapping("1:5") + assert result == {1: [5]} + + def test_multi_disc(self): + result = parse_disc_mapping("1:1-3,2:4-6,3:7-9,4:10") + assert result == {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8, 9], 4: [10]} + + def test_spaces(self): + result = parse_disc_mapping("1:1-3 , 2:4-6") + assert result == {1: [1, 2, 3], 2: [4, 5, 6]} + + def test_empty_string(self): + result = parse_disc_mapping("") + assert result == {} + + def test_none(self): + result = parse_disc_mapping(None) + assert result == {} + + def test_single_ep_per_disc(self): + result = parse_disc_mapping("1:1,2:2,3:3") + assert result == {1: [1], 2: [2], 3: [3]} + + def test_large_numbers(self): + result = parse_disc_mapping("1:1-22") + assert result == {1: list(range(1, 23))} + + def test_invalid_format_exits(self, capsys): + with pytest.raises(SystemExit): + parse_disc_mapping("invalid") + + def test_missing_colon_exits(self, capsys): + with pytest.raises(SystemExit): + parse_disc_mapping("1,2,3") + + +class TestArgParse: + def test_parser_basic_args(self): + import argparse + from episode_matcher import main + + with patch("sys.argv", [ + "episode_matcher.py", + "/tmp/episodes", + "Test Show", + "1", + ]): + with patch("pathlib.Path.exists", return_value=True): + with patch("pathlib.Path.is_dir", return_value=True): + with patch("episode_matcher.TVDBClient") as mock_tvdb: + mock_instance = mock_tvdb.return_value + mock_instance.authenticate.return_value = False + with patch("episode_matcher.MockTVDBClient") as mock_mock: + m = mock_mock.return_value + m.authenticate.return_value = True + m.get_episode_durations.return_value = [] + with patch("episode_matcher.FileClassifier") as mock_fc: + fc = mock_fc.return_value + fc.file_info = [] + fc.classify_files.return_value = ([], []) + with pytest.raises(SystemExit): + main() + + def test_parser_dry_run_flag(self): + import sys as real_sys + orig_argv = real_sys.argv + try: + real_sys.argv = [ + "episode_matcher.py", "/tmp/e", "Show", "1", "--dry-run", "--verbose" + ] + from importlib import reload + import episode_matcher + + parser = episode_matcher.__dict__.get("_parser", None) + + from episode_matcher import main + finally: + real_sys.argv = orig_argv + + +class TestParseDiscMappingEdgeCases: + def test_range_to_self(self): + result = parse_disc_mapping("1:5-5") + assert result == {1: [5]} + + def test_multiple_ranges_same_ep(self): + result = parse_disc_mapping("1:1-3,2:3-5") + assert 3 in result[1] + assert 3 in result[2] + + def test_disc_number_large(self): + result = parse_disc_mapping("99:1-3") + assert 99 in result diff --git a/tests/test_episode_renamer.py b/tests/test_episode_renamer.py new file mode 100644 index 0000000..6843a7e --- /dev/null +++ b/tests/test_episode_renamer.py @@ -0,0 +1,539 @@ +"""Tests for src/Matcher/episode_renamer.py - EpisodeRenamer.""" +import sys +from pathlib import Path +from unittest.mock import patch, MagicMock + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from src.Matcher.episode_renamer import EpisodeRenamer + + +class TestEpisodeRenamerInit: + def test_init_defaults(self): + r = EpisodeRenamer("/tmp/test", "Show Name", 1) + assert r.show_name == "Show Name" + assert r.season_number == 1 + assert r.disc_mapping is None + assert r.auto_delete_duplicates is False + + def test_init_with_disc_mapping(self): + r = EpisodeRenamer("/tmp/test", "Show", 2, disc_mapping={1: [1, 2], 2: [3, 4]}) + assert r.disc_mapping == {1: [1, 2], 2: [3, 4]} + + def test_init_auto_delete(self): + r = EpisodeRenamer("/tmp/test", "Show", 1, auto_delete_duplicates=True) + assert r.auto_delete_duplicates is True + + def test_init_folder_paths(self): + r = EpisodeRenamer("/tmp/test", "Show", 1) + assert r.extras_folder == Path("/tmp/test/extras") + assert r.delete_folder == Path("/tmp/test/delete me") + + +class TestSanitizeFilename: + def test_remove_invalid_chars(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + result = r._sanitize_filename('ac:d"e/f\\g|h?i*j') + assert ":" not in result + assert "<" not in result + assert ">" not in result + assert '"' not in result + assert "/" not in result + assert "\\" not in result + assert "|" not in result + assert "?" not in result + assert "*" not in result + assert "a" in result and "b" in result and "c" in result + + def test_collapse_spaces(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + assert r._sanitize_filename("a b") == "a b" + + def test_strip(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + assert r._sanitize_filename(" hello ") == "hello" + + def test_preserve_valid_chars(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + assert r._sanitize_filename("hello_world-1.mkv") == "hello_world-1.mkv" + + +class TestGenerateEpisodeFilename: + def test_standard(self): + r = EpisodeRenamer("/tmp/t", "Show Name", 3) + assert r._generate_episode_filename(5, ".mkv") == "Show Name s03e05.mkv" + + def test_single_digit_padding(self): + r = EpisodeRenamer("/tmp/t", "Show", 1) + assert r._generate_episode_filename(1, ".mp4") == "Show s01e01.mp4" + + def test_double_digit_episode(self): + r = EpisodeRenamer("/tmp/t", "Show", 10) + assert r._generate_episode_filename(12, ".mkv") == "Show s10e12.mkv" + + def test_sanitizes_output(self): + r = EpisodeRenamer("/tmp/t", "Show:Name", 1) + assert r._generate_episode_filename(1, ".mkv") == "ShowName s01e01.mkv" + + +class TestExtractDiscInfo: + def test_disc_with_underscore(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = r._extract_disc_info("Disc 1_t01.mkv") + assert info["disc_number"] == 1 + assert info["track_number"] == 1 + + def test_disk_variant(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = r._extract_disc_info("Disk2_t03.mkv") + assert info["disc_number"] == 2 + assert info["track_number"] == 3 + + def test_disc_hyphen(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = r._extract_disc_info("disc-3_track05.mkv") + assert info["disc_number"] == 3 + assert info["track_number"] == 5 + + def test_no_disc_info(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = r._extract_disc_info("episode_01.mkv") + assert info["disc_number"] is None + assert info["track_number"] is None + + def test_track_pattern(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = r._extract_disc_info("file_t22.mkv") + assert info["track_number"] == 22 + + def test_disc_only(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = r._extract_disc_info("Disc 5.mkv") + assert info["disc_number"] == 5 + assert info["track_number"] is None + + +class TestComputeFileHash: + def test_hash_consistent(self, tmp_path): + f = tmp_path / "test.mkv" + f.write_bytes(b"\x00" * 1000) + r = EpisodeRenamer(str(tmp_path), "S", 1) + h1 = r._compute_file_hash(f) + h2 = r._compute_file_hash(f) + assert h1 == h2 + assert len(h1) == 64 + + def test_hash_different_content(self, tmp_path): + f1 = tmp_path / "a.mkv" + f2 = tmp_path / "b.mkv" + f1.write_bytes(b"\x00" * 100) + f2.write_bytes(b"\xff" * 100) + r = EpisodeRenamer(str(tmp_path), "S", 1) + assert r._compute_file_hash(f1) != r._compute_file_hash(f2) + + def test_hash_custom_bytes(self, tmp_path): + f = tmp_path / "test.mkv" + f.write_bytes(b"\x00" * 10000) + r = EpisodeRenamer(str(tmp_path), "S", 1) + h = r._compute_file_hash(f, num_bytes=512) + assert h + + def test_hash_missing_file(self, tmp_path): + r = EpisodeRenamer(str(tmp_path), "S", 1) + h = r._compute_file_hash(tmp_path / "nope.mkv") + assert h == "" + + +class TestCreateFolders: + def test_create_extras_folder(self, tmp_path): + r = EpisodeRenamer(str(tmp_path), "S", 1) + assert r._create_extras_folder() is True + assert (tmp_path / "extras").exists() + + def test_create_delete_folder(self, tmp_path): + r = EpisodeRenamer(str(tmp_path), "S", 1) + assert r._create_delete_folder() is True + assert (tmp_path / "delete me").exists() + + def test_create_extras_already_exists(self, tmp_path): + (tmp_path / "extras").mkdir() + r = EpisodeRenamer(str(tmp_path), "S", 1) + assert r._create_extras_folder() is True + + +class TestMoveExtras: + def test_move_extras(self, tmp_path): + (tmp_path / "extra.mkv").write_bytes(b"\x00" * 100) + r = EpisodeRenamer(str(tmp_path), "S", 1) + extras = [{"path": tmp_path / "extra.mkv", "name": "extra.mkv"}] + moved = r.move_extras_to_folder(extras) + assert len(moved) == 1 + assert (tmp_path / "extras" / "extra.mkv").exists() + assert not (tmp_path / "extra.mkv").exists() + + def test_move_extras_skip_existing(self, tmp_path): + (tmp_path / "extras").mkdir() + (tmp_path / "extras" / "extra.mkv").write_bytes(b"\x00") + (tmp_path / "extra.mkv").write_bytes(b"\x00" * 100) + r = EpisodeRenamer(str(tmp_path), "S", 1) + extras = [{"path": tmp_path / "extra.mkv", "name": "extra.mkv"}] + moved = r.move_extras_to_folder(extras) + assert len(moved) == 0 + + def test_move_extras_empty(self, tmp_path): + r = EpisodeRenamer(str(tmp_path), "S", 1) + assert r.move_extras_to_folder([]) == [] + + +class TestDuplicateDetection: + def test_detect_duplicates_same_duration(self, tmp_path): + f1 = tmp_path / "Disc 1_t01.mkv" + f2 = tmp_path / "Disc 1_t02.mkv" + f1.write_bytes(b"\x00" * 2000) + f2.write_bytes(b"\x00" * 2000) + eps = [ + {"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0}, + {"path": f2, "name": "Disc 1_t02.mkv", "size_gb": 2.0, "duration_minutes": 45.0}, + ] + r = EpisodeRenamer(str(tmp_path), "S", 1) + result = r.detect_and_move_duplicates(eps) + assert isinstance(result, list) + + def test_no_duplicates_different_duration(self, tmp_path): + f1 = tmp_path / "Disc 1_t01.mkv" + f2 = tmp_path / "Disc 2_t01.mkv" + f1.write_bytes(b"\x00" * 2000) + f2.write_bytes(b"\x01" * 3000) + eps = [ + {"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0}, + {"path": f2, "name": "Disc 2_t01.mkv", "size_gb": 3.0, "duration_minutes": 55.0}, + ] + r = EpisodeRenamer(str(tmp_path), "S", 1) + r.detect_and_move_duplicates(eps) + + def test_duplicate_moved_to_delete_me(self, tmp_path): + f1 = tmp_path / "Disc 1_t01.mkv" + f2 = tmp_path / "Disc 1_t02.mkv" + f1.write_bytes(b"\x00" * 2000) + f2.write_bytes(b"\x00" * 2000) + eps = [ + {"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0}, + {"path": f2, "name": "Disc 1_t02.mkv", "size_gb": 2.0, "duration_minutes": 45.0}, + ] + r = EpisodeRenamer(str(tmp_path), "S", 1, auto_delete_duplicates=False) + result = r.detect_and_move_duplicates(eps) + assert (tmp_path / "delete me").exists() + + +class TestMatchEpisodes: + def test_disc_mapping_match(self, mock_episodes, mock_tvdb_episodes, disc_mapping): + r = EpisodeRenamer("/tmp/t", "Show", 1, disc_mapping=disc_mapping) + episodes_info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + episodes_info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "track_number": disc_info["track_number"], + "filename": ep["name"], + "duration": ep["duration_minutes"], + }) + matched = r._match_using_disc_mapping(episodes_info, mock_tvdb_episodes) + assert len(matched) > 0 + ep_nums = [m["episode_number"] for m in matched] + assert len(set(ep_nums)) == len(ep_nums) + + def test_fallback_sequential(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + files = [] + for ep in mock_episodes: + files.append({ + "file_info": ep, + "filename": ep["name"], + "disc_number": None, + "duration": ep["duration_minutes"], + }) + allowed = list(range(1, 11)) + result = r._fallback_sequential(files, mock_tvdb_episodes, allowed) + assert len(result) > 0 + + def test_dp_match_episodes(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + files = [] + for ep in mock_episodes: + files.append({ + "file_info": ep, + "filename": ep["name"], + "disc_number": None, + "duration": ep["duration_minutes"], + }) + allowed = list(range(1, 11)) + result = r._dp_match_episodes(files, mock_tvdb_episodes, allowed) + assert len(result) > 0 + ep_nums = [m["episode_number"] for m in result] + assert len(set(ep_nums)) == len(ep_nums) + + def test_precise_duration_match(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + files = [] + for ep in mock_episodes: + files.append({ + "file_info": ep, + "filename": ep["name"], + "disc_number": None, + "duration": ep["duration_minutes"], + }) + allowed = list(range(1, 7)) + result = r._precise_duration_match(files, mock_tvdb_episodes, allowed) + assert len(result) > 0 + + def test_dp_match_empty_files(self, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + result = r._dp_match_episodes([], mock_tvdb_episodes, list(range(1, 11))) + assert isinstance(result, list) + + def test_precise_match_empty(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + files = [{"file_info": {}, "filename": "a.mkv", "duration": 45.0}] + result = r._precise_duration_match(files, [], []) + assert isinstance(result, list) + + def test_fallback_sequential_empty(self, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + result = r._fallback_sequential([], mock_tvdb_episodes, []) + assert result == [] + + +class TestRenameEpisodes: + def test_rename_no_episodes(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + assert r.rename_episodes([]) == [] + + def test_rename_with_tvdb(self, tmp_path, mock_episodes, mock_tvdb_episodes): + for ep in mock_episodes: + f = tmp_path / ep["name"] + f.write_bytes(b"\x00" * ep["size_bytes"]) + ep["path"] = f + r = EpisodeRenamer(str(tmp_path), "TestShow", 1) + renamed = r.rename_episodes(mock_episodes, mock_tvdb_episodes) + assert len(renamed) > 0 + for item in renamed: + assert "original_name" in item + assert "new_name" in item + assert "episode_number" in item + + def test_rename_skip_existing(self, tmp_path, mock_episodes, capsys): + for ep in mock_episodes[:2]: + f = tmp_path / ep["name"] + f.write_bytes(b"\x00" * ep["size_bytes"]) + ep["path"] = f + (tmp_path / "TestShow s01e01.mkv").write_bytes(b"\x00") + r = EpisodeRenamer(str(tmp_path), "TestShow", 1) + renamed = r.rename_episodes(mock_episodes[:2], None) + out = capsys.readouterr().out + assert "already exists" in out or len(renamed) < 2 + + +class TestEstimateEpisodesPerDisc: + def test_estimate_basic(self, tmp_path, mock_episodes): + for ep in mock_episodes: + f = tmp_path / ep["name"] + f.write_bytes(b"\x00" * ep["size_bytes"]) + ep["path"] = f + r = EpisodeRenamer(str(tmp_path), "S", 1) + episodes_info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + episodes_info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "filename": ep["name"], + }) + result = r._estimate_episodes_per_disc(episodes_info, 6) + assert isinstance(result, dict) + + def test_estimate_no_disc_info(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + result = r._estimate_episodes_per_disc([], 10) + assert result == {} + + def test_estimate_no_disc_numbers(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + info = [{"disc_number": None, "filename": "a.mkv", "file_info": {"size_gb": 5.0}}] + result = r._estimate_episodes_per_disc(info, 10) + assert result == {} + + +class TestAnalyzeDiscCapacity: + def test_analyze_basic(self, tmp_path, mock_episodes): + for ep in mock_episodes: + f = tmp_path / ep["name"] + f.write_bytes(b"\x00" * ep["size_bytes"]) + ep["path"] = f + r = EpisodeRenamer(str(tmp_path), "S", 1) + info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "filename": ep["name"], + }) + result = r._analyze_disc_capacity(info) + assert isinstance(result, dict) + + def test_analyze_no_discs(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + result = r._analyze_disc_capacity([{"disc_number": None, "file_info": {"size_gb": 5}}]) + assert result == {} + + +class TestDurationMatching: + def test_match_by_duration_and_order(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "track_number": disc_info["track_number"], + "filename": ep["name"], + "duration": ep["duration_minutes"], + }) + result = r._match_by_duration_and_order(info, mock_tvdb_episodes) + assert len(result) > 0 + ep_nums = [m["episode_number"] for m in result] + assert len(set(ep_nums)) == len(ep_nums) + + def test_flexible_duration_match(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "filename": ep["name"], + "duration": ep["duration_minutes"], + }) + result = r._flexible_duration_match(info, mock_tvdb_episodes) + assert isinstance(result, list) + + def test_sequential_assignment_with_constraints(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "filename": ep["name"], + "duration": ep["duration_minutes"], + }) + disc_capacity = {1: 3, 2: 3} + result = r._sequential_assignment_with_constraints(info, mock_tvdb_episodes, disc_capacity) + assert isinstance(result, list) + + def test_validate_sequential_assignment_valid(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + file_info = {"duration": 55.0, "disc_number": 1} + tvdb_ep = {"runtime": 55, "episode_number": 1} + assert r._validate_sequential_assignment(file_info, tvdb_ep, {}, {}) is True + + def test_validate_sequential_assignment_duration_mismatch(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + file_info = {"duration": 30.0, "disc_number": 1} + tvdb_ep = {"runtime": 60, "episode_number": 1} + result = r._validate_sequential_assignment(file_info, tvdb_ep, {}, {}) + assert result is False + + def test_validate_sequential_assignment_disc_full(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + file_info = {"duration": 55.0, "disc_number": 1} + tvdb_ep = {"runtime": 55, "episode_number": 1} + disc_assignments = {1: 3} + disc_capacity = {1: 3} + result = r._validate_sequential_assignment(file_info, tvdb_ep, disc_assignments, disc_capacity) + assert result is False + + def test_assignment_satisfies_constraints(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + fake_path = type("P", (), {"name": "Disc 1_t01.mkv"})() + file_info = { + "duration": 55.0, + "disc_number": 1, + "file_info": {"path": fake_path}, + } + tvdb_ep = {"runtime": 55, "episode_number": 1} + result = r._assignment_satisfies_constraints(file_info, tvdb_ep, [], {}) + assert result is True + + def test_assignment_satisfies_duration_exceeded(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + fake_path = type("P", (), {"name": "a.mkv"})() + file_info = { + "duration": 30.0, + "disc_number": None, + "file_info": {"path": fake_path}, + } + tvdb_ep = {"runtime": 60, "episode_number": 1} + result = r._assignment_satisfies_constraints(file_info, tvdb_ep, [], {}) + assert result is False + + +class TestMatchByDurationAndDisc: + def test_match_disc(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "filename": ep["name"], + "duration": ep["duration_minutes"], + }) + episodes_per_disc = {1: [1, 2, 3], 2: [4, 5, 6]} + result = r._match_by_duration_and_disc(info, mock_tvdb_episodes, episodes_per_disc) + assert len(result) > 0 + for m in result: + assert m["episode_number"] in [1, 2, 3, 4, 5, 6] + + +class TestFindOptimalAssignment: + def test_find_optimal_assignment(self, mock_episodes, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + info = [] + for ep in mock_episodes: + disc_info = r._extract_disc_info(ep["name"]) + info.append({ + "file_info": ep, + "disc_number": disc_info["disc_number"], + "filename": ep["name"], + "duration": ep["duration_minutes"], + }) + result = r._find_optimal_assignment(info, mock_tvdb_episodes, {1: 3, 2: 3}) + assert isinstance(result, list) + + def test_find_optimal_empty(self, mock_tvdb_episodes): + r = EpisodeRenamer("/tmp/t", "Show", 1) + result = r._find_optimal_assignment([], mock_tvdb_episodes, {}) + assert result == [] + + +class TestIsValidAssignment: + def test_valid_assignment(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + file_info = {"duration": 55.0, "disc_number": None} + tvdb_ep = {"runtime": 55, "episode_number": 1} + result = r._is_valid_assignment(file_info, tvdb_ep, {}, 0, []) + assert result is True + + def test_invalid_duration(self): + r = EpisodeRenamer("/tmp/t", "S", 1) + file_info = {"duration": 30.0, "disc_number": None} + tvdb_ep = {"runtime": 60, "episode_number": 1} + result = r._is_valid_assignment(file_info, tvdb_ep, {}, 0, []) + assert result is False diff --git a/tests/test_file_classifier.py b/tests/test_file_classifier.py new file mode 100644 index 0000000..4582d4a --- /dev/null +++ b/tests/test_file_classifier.py @@ -0,0 +1,194 @@ +"""Tests for src/Matcher/file_classifier.py - FileClassifier.""" +import os +import sys +from pathlib import Path +from unittest.mock import patch, MagicMock + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from src.Matcher.file_classifier import FileClassifier, DEFAULT_VIDEO_EXTENSIONS + + +class TestFileClassifierInit: + def test_init_valid_folder(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + assert fc.folder_path == mock_video_files + assert len(fc.video_files) > 0 + + def test_init_nonexistent_folder(self, tmp_path): + with pytest.raises(FileNotFoundError, match="Folder not found"): + FileClassifier(str(tmp_path / "nope")) + + def test_init_file_not_dir(self, tmp_path): + f = tmp_path / "file.txt" + f.write_text("hi") + fc = FileClassifier(str(f)) + assert len(fc.video_files) == 0 + + def test_init_custom_extensions(self, tmp_path): + (tmp_path / "video.mp4").write_bytes(b"\x00" * 100) + (tmp_path / "video.mkv").write_bytes(b"\x00" * 100) + fc = FileClassifier(str(tmp_path), video_extensions=[".mp4"]) + assert len(fc.video_files) == 1 + assert fc.video_files[0].name == "video.mp4" + + def test_init_fallback_ratio(self, mock_video_files): + fc = FileClassifier(str(mock_video_files), fallback_duration_minutes_per_gb=25.0) + assert fc.fallback_ratio == 25.0 + + def test_init_filters_macos_resource_fork(self, tmp_path): + (tmp_path / "._hidden.mkv").write_bytes(b"\x00" * 100) + (tmp_path / ".dot.mkv").write_bytes(b"\x00" * 100) + (tmp_path / "visible.mkv").write_bytes(b"\x00" * 100) + fc = FileClassifier(str(tmp_path)) + assert len(fc.video_files) == 1 + assert fc.video_files[0].name == "visible.mkv" + + def test_init_deduplicates(self, tmp_path): + (tmp_path / "dup.mkv").write_bytes(b"\x00" * 100) + fc = FileClassifier(str(tmp_path), video_extensions=[".mkv"]) + assert len(fc.video_files) == 1 + + def test_init_empty_folder(self, tmp_path): + fc = FileClassifier(str(tmp_path)) + assert fc.video_files == [] + + +class TestResolveExtensions: + def test_resolve_default(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + assert ".mkv" in fc.video_extensions + + def test_resolve_custom(self, mock_video_files): + fc = FileClassifier(str(mock_video_files), video_extensions=["mp4", "avi"]) + assert ".mp4" in fc.video_extensions + assert ".avi" in fc.video_extensions + assert ".mkv" not in fc.video_extensions + + def test_resolve_env_var(self, mock_video_files, monkeypatch): + monkeypatch.setenv("VIDEO_EXTENSIONS", "webm,mp4") + fc = FileClassifier(str(mock_video_files)) + assert "webm" in fc.video_extensions + assert "mp4" in fc.video_extensions + + +class TestResolveFallbackRatio: + def test_explicit_ratio(self, mock_video_files): + fc = FileClassifier(str(mock_video_files), fallback_duration_minutes_per_gb=30.0) + assert fc.fallback_ratio == 30.0 + + def test_env_ratio(self, mock_video_files, monkeypatch): + monkeypatch.setenv("FALLBACK_DURATION_RATIO", "33.0") + fc = FileClassifier(str(mock_video_files)) + assert fc.fallback_ratio == 33.0 + + def test_env_ratio_invalid(self, mock_video_files, monkeypatch): + monkeypatch.setenv("FALLBACK_DURATION_RATIO", "abc") + fc = FileClassifier(str(mock_video_files)) + assert fc.fallback_ratio == 45.0 + + +class TestFileAnalysis: + def test_get_file_size(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + size = fc._get_file_size(mock_video_files / "Disc 1_t01.mkv") + assert size == int(2 * 1024) + + def test_analyze_files_structure(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + assert len(fc.file_info) > 0 + info = fc.file_info[0] + assert "path" in info + assert "name" in info + assert "size_bytes" in info + assert "size_mb" in info + assert "size_gb" in info + assert "duration_minutes" in info + + def test_calculate_stats(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + stats = fc._calculate_stats() + assert stats["count"] == len(fc.file_info) + assert "avg_size_bytes" in stats + assert "avg_duration_minutes" in stats + assert "median_size_bytes" in stats + + def test_calculate_stats_empty(self, tmp_path): + fc = FileClassifier(str(tmp_path)) + stats = fc._calculate_stats() + assert stats == {} + + +class TestClassifyFiles: + def test_classify_no_files(self, tmp_path): + fc = FileClassifier(str(tmp_path)) + episodes, extras = fc.classify_files() + assert episodes == [] + assert extras == [] + + def test_classify_with_tvdb(self, mock_video_files, mock_tvdb_episodes): + fc = FileClassifier(str(mock_video_files)) + episodes, extras = fc.classify_files( + expected_episode_count=6, tvdb_episodes=mock_tvdb_episodes + ) + assert len(episodes) <= 6 + assert len(episodes) + len(extras) == len(fc.file_info) + + def test_classify_by_stats(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + episodes, extras = fc.classify_files(expected_episode_count=6) + assert len(episodes) <= 6 + assert len(episodes) > 0 + + def test_classify_extra_small_file(self, mock_video_files): + fc = FileClassifier(str(mock_video_files)) + episodes, extras = fc.classify_files() + extra_names = [e["name"] for e in extras] + assert "extra_short.mkv" in extra_names + + def test_classify_expected_count(self, mock_video_files, mock_tvdb_episodes): + fc = FileClassifier(str(mock_video_files)) + episodes, extras = fc.classify_files( + expected_episode_count=3, tvdb_episodes=mock_tvdb_episodes + ) + assert len(episodes) <= 3 + + +class TestVideoDuration: + @patch("src.Matcher.file_classifier.MediaInfo", None) + def test_duration_fallback_no_mediainfo(self, mock_video_files): + fc = FileClassifier( + str(mock_video_files), fallback_duration_minutes_per_gb=45.0 + ) + target = mock_video_files / "Disc 1_t01.mkv" + dur = fc._get_video_duration(target) + expected = (target.stat().st_size / (1024**3)) * 45.0 + assert abs(dur - expected) < 0.001 + assert dur > 0 + + def test_duration_fallback_custom_ratio(self, mock_video_files): + with patch("src.Matcher.file_classifier.MediaInfo", None): + fc = FileClassifier( + str(mock_video_files), fallback_duration_minutes_per_gb=25.0 + ) + target = mock_video_files / "Disc 1_t01.mkv" + dur = fc._get_video_duration(target) + expected = (target.stat().st_size / (1024**3)) * 25.0 + assert abs(dur - expected) < 0.001 + assert dur > 0 + + +class TestPrintAnalysis: + def test_print_analysis_has_files(self, mock_video_files, capsys): + fc = FileClassifier(str(mock_video_files)) + fc.print_analysis() + out = capsys.readouterr().out + assert "File Analysis" in out + assert "Total files" in out + + def test_print_analysis_no_files(self, tmp_path, capsys): + fc = FileClassifier(str(tmp_path)) + fc.print_analysis() + out = capsys.readouterr().out + assert "No video files found" in out diff --git a/tests/test_tvdb_cache.py b/tests/test_tvdb_cache.py new file mode 100644 index 0000000..c149ae7 --- /dev/null +++ b/tests/test_tvdb_cache.py @@ -0,0 +1,236 @@ +"""Tests for src/TVDBProvider/tvdb_cache.py - TVDBCache.""" +import sys +import json +import time +from pathlib import Path +from datetime import datetime, timedelta +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from src.TVDBProvider.tvdb_cache import TVDBCache + + +class TestTVDBCacheInit: + def test_init_creates_dir(self, tmp_path): + cache = TVDBCache(str(tmp_path / "cache")) + assert (tmp_path / "cache").exists() + + def test_init_default_dir(self): + cache = TVDBCache() + assert cache.cache_dir.exists() + assert cache.cache_duration == timedelta(days=7) + + def test_init_absolute_path(self): + cache = TVDBCache("/tmp/absolute_cache_test") + try: + assert cache.cache_dir.is_absolute() + finally: + import shutil + + if cache.cache_dir.exists(): + shutil.rmtree(cache.cache_dir, ignore_errors=True) + + +class TestSanitizeName: + def test_basic_sanitization(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("Game of Thrones") + assert result == "game_of_thrones" + + def test_special_chars_removed(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("The Simpsons (1989)") + assert "(" not in result + assert ")" not in result + + def test_dots_stripped(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("../../etc/passwd") + assert ".." not in result + assert "/" not in result + + def test_slashes_stripped(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("a/b") + assert result == "ab" + + def test_backslash_stripped(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("a\\b") + assert result == "ab" + + def test_numbers_preserved(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("Show 123") + assert "123" in result + + def test_underscores_preserved(self): + cache = TVDBCache.__new__(TVDBCache) + result = cache._sanitize_name("my_show") + assert result == "my_show" + + +class TestCacheKey: + def test_cache_key_format(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + key = cache._get_cache_key("Test Show", 3) + assert key == "test_show_s03" + + def test_cache_key_special_chars(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + key = cache._get_cache_key("Test (Show)!", 1) + assert "(" not in key + assert ")" not in key + + +class TestCacheFile: + def test_cache_file_path(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + f = cache._get_cache_file("test_s01") + assert f.name == "test_s01.json" + + def test_cache_file_escape_detected(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + with pytest.raises(ValueError, match="escape detected"): + cache._get_cache_file("../evil") + + +class TestCacheValidity: + def test_valid_cache(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + f = tmp_path / "c" / "valid.json" + f.write_text("data") + assert cache._is_cache_valid(f) is True + + def test_expired_cache(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_duration = timedelta(seconds=0) + f = tmp_path / "c" / "old.json" + f.write_text("data") + time.sleep(0.1) + assert cache._is_cache_valid(f) is False + + def test_nonexistent_cache(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + assert cache._is_cache_valid(tmp_path / "c" / "nope.json") is False + + +class TestCacheEpisodes: + def test_cache_and_retrieve(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + episodes = [{"episode_number": 1, "runtime": 45}] + result = cache.cache_episodes("Test Show", 1, episodes) + assert result is True + + retrieved = cache.get_cached_episodes("Test Show", 1) + assert len(retrieved) == 1 + assert retrieved[0]["episode_number"] == 1 + + def test_cache_write_failure(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_dir = Path("/nonexistent_dir") + result = cache.cache_episodes("Test", 1, []) + assert result is False + + def test_cache_data_structure(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + episodes = [{"episode_number": 1, "runtime": 45}] + cache.cache_episodes("Test", 1, episodes) + + cache_file = tmp_path / "c" / "test_s01.json" + data = json.loads(cache_file.read_text()) + assert data["series_name"] == "Test" + assert data["season_number"] == 1 + assert "cached_at" in data + assert len(data["episodes"]) == 1 + + def test_cache_empty_episodes(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + result = cache.cache_episodes("Empty", 1, []) + assert result is True + retrieved = cache.get_cached_episodes("Empty", 1) + assert retrieved == [] + + +class TestGetCachedEpisodes: + def test_cache_miss(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + result = cache.get_cached_episodes("NoShow", 1) + assert result is None + + def test_corrupted_json(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + f = tmp_path / "c" / "broken.json" + f.write_text("{invalid") + result = cache.get_cached_episodes("Broken", 1) + assert result is None + + def test_missing_fields(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + f = tmp_path / "c" / "partial_s01.json" + f.write_text(json.dumps({"only": "this"})) + result = cache.get_cached_episodes("Partial", 1) + assert result is None + + +class TestClearCache: + def test_clear_removes_files(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_episodes("Show1", 1, []) + cache.cache_episodes("Show2", 1, []) + result = cache.clear_cache() + assert result is True + files = list((tmp_path / "c").glob("*.json")) + assert len(files) == 0 + + def test_clear_empty(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + assert cache.clear_cache() is True + + +class TestListCache: + def test_list_entries(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_episodes("Show1", 1, [{"ep": 1}]) + cache.cache_episodes("Show2", 2, [{"ep": 1}, {"ep": 2}]) + entries = cache.list_cache() + assert len(entries) == 2 + for e in entries: + assert "series_name" in e + assert "season_number" in e + assert "episode_count" in e + assert "is_valid" in e + + def test_list_empty(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + entries = cache.list_cache() + assert entries == [] + + def test_list_entry_fields(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_episodes("Test", 1, [{"e": 1}]) + entries = cache.list_cache() + e = entries[0] + assert e["series_name"] == "Test" + assert e["season_number"] == 1 + assert e["episode_count"] == 1 + assert "cached_at" in e + assert "age_days" in e + assert "cache_key" in e + + def test_list_sorted_by_date(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_episodes("First", 1, []) + time.sleep(0.01) + cache.cache_episodes("Second", 1, []) + entries = cache.list_cache() + assert entries[0]["series_name"] == "Second" + + def test_list_corrupted_file(self, tmp_path): + cache = TVDBCache(str(tmp_path / "c")) + cache.cache_episodes("Good", 1, [{"e": 1}]) + (tmp_path / "c" / "bad_s01.json").write_text("{bad}") + entries = cache.list_cache() + assert len(entries) == 1 diff --git a/tests/test_tvdb_client.py b/tests/test_tvdb_client.py new file mode 100644 index 0000000..e024b94 --- /dev/null +++ b/tests/test_tvdb_client.py @@ -0,0 +1,224 @@ +"""Tests for src/TVDBProvider/tvdb_client.py.""" +import sys +from pathlib import Path +from unittest.mock import patch, MagicMock, PropertyMock + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from src.TVDBProvider.tvdb_client import TVDBClient, MockTVDBClient + + +class TestTVDBClientInit: + def test_init(self): + client = TVDBClient() + assert client.base_url == "https://api4.thetvdb.com/v4" + assert client.token is None + assert "Content-Type" in client.headers + + def test_init_has_cache(self): + client = TVDBClient() + assert client.cache is not None + + +class TestTVDBClientAuthenticate: + @patch("src.TVDBProvider.tvdb_client.requests.post") + def test_auth_success(self, mock_post): + mock_post.return_value.status_code = 200 + mock_post.return_value.json.return_value = {"data": {"token": "abc123"}} + client = TVDBClient() + assert client.authenticate("my_api_key") is True + assert client.token == "abc123" + assert "Bearer abc123" in client.headers["Authorization"] + + @patch("src.TVDBProvider.tvdb_client.requests.post") + def test_auth_failure_status(self, mock_post): + mock_post.return_value.status_code = 401 + client = TVDBClient() + assert client.authenticate("bad_key") is False + assert client.token is None + + @patch("src.TVDBProvider.tvdb_client.requests.post") + def test_auth_exception(self, mock_post): + mock_post.side_effect = Exception("network error") + client = TVDBClient() + assert client.authenticate("key") is False + + +class TestTVDBClientSearch: + @patch("src.TVDBProvider.tvdb_client.requests.get") + def test_search_success(self, mock_get): + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = { + "data": [{"tvdb_id": 123, "name": "Test Show"}] + } + client = TVDBClient() + client.token = "tok" + result = client.search_series("Test Show") + assert result["tvdb_id"] == 123 + + @patch("src.TVDBProvider.tvdb_client.requests.get") + def test_search_no_results(self, mock_get): + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = {"data": []} + client = TVDBClient() + client.token = "tok" + result = client.search_series("NoMatch") + assert result is None + + def test_search_not_authenticated(self): + client = TVDBClient() + result = client.search_series("Show") + assert result is None + + @patch("src.TVDBProvider.tvdb_client.requests.get") + def test_search_error(self, mock_get): + mock_get.side_effect = Exception("timeout") + client = TVDBClient() + client.token = "tok" + result = client.search_series("Show") + assert result is None + + +class TestTVDBClientSeasonEpisodes: + @patch("src.TVDBProvider.tvdb_client.requests.get") + def test_get_season_episodes(self, mock_get): + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = { + "data": {"episodes": [{"number": 1, "name": "E1"}]} + } + client = TVDBClient() + client.token = "tok" + result = client.get_season_episodes(123, 1) + assert len(result) == 1 + + def test_episodes_not_authenticated(self): + client = TVDBClient() + result = client.get_season_episodes(123, 1) + assert result is None + + +class TestTVDBClientSeriesInfo: + @patch.object(TVDBClient, "search_series") + def test_get_series_info(self, mock_search): + mock_search.return_value = {"tvdb_id": 123, "name": "Test", "slug": "test", "year": 2020} + client = TVDBClient() + client.token = "tok" + result = client.get_series_info("Test") + assert result["id"] == 123 + assert result["name"] == "Test" + + @patch.object(TVDBClient, "search_series") + def test_get_series_info_no_match(self, mock_search): + mock_search.return_value = None + client = TVDBClient() + client.token = "tok" + result = client.get_series_info("NoMatch") + assert result is None + + +class TestTVDBClientEpisodeDurations: + def test_get_episode_durations(self, tmp_path): + with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init: + mock_init.return_value = None + client = TVDBClient() + client.base_url = "https://api4.thetvdb.com/v4" + client.token = "tok" + client.headers = {"Content-Type": "application/json"} + + mock_cache = MagicMock() + mock_cache.get_cached_episodes.return_value = None + client.cache = mock_cache + + client.get_series_info = MagicMock(return_value={"id": 123}) + client.get_season_episodes = MagicMock(return_value=[ + {"number": 1, "name": "E1", "runtime": 45, "aired": "2020-01-01"}, + {"number": 2, "name": "E2", "runtime": 50, "aired": "2020-01-08"}, + ]) + + result = client.get_episode_durations("Test", 1) + assert len(result) == 2 + assert result[0]["episode_number"] == 1 + assert result[0]["runtime"] == 45 + mock_cache.cache_episodes.assert_called_once() + + def test_get_episode_durations_cached(self, tmp_path): + with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init: + mock_init.return_value = None + client = TVDBClient() + client.token = "tok" + + mock_cache = MagicMock() + mock_cache.get_cached_episodes.return_value = [ + {"episode_number": 1, "runtime": 45} + ] + client.cache = mock_cache + + result = client.get_episode_durations("Test", 1) + assert len(result) == 1 + + def test_get_episode_durations_no_series(self, tmp_path): + with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init: + mock_init.return_value = None + client = TVDBClient() + client.token = "tok" + + mock_cache = MagicMock() + mock_cache.get_cached_episodes.return_value = None + client.cache = mock_cache + + client.get_series_info = MagicMock(return_value=None) + + result = client.get_episode_durations("NoShow", 1) + assert result is None + + +class TestMockTVDBClient: + def test_mock_auth(self): + client = MockTVDBClient() + assert client.authenticate() is True + assert client.authenticated is True + + def test_mock_auth_with_key(self): + client = MockTVDBClient() + assert client.authenticate("some_key") is True + + def test_mock_episodes_unauthenticated(self): + client = MockTVDBClient() + result = client.get_episode_durations("Show", 1) + assert result is None + + def test_mock_episodes_drama(self): + client = MockTVDBClient() + client.authenticate() + result = client.get_episode_durations("Game of Thrones", 1) + assert len(result) == 10 + + def test_mock_episodes_sitcom(self): + client = MockTVDBClient() + client.authenticate() + result = client.get_episode_durations("Friends", 1) + assert len(result) == 24 + + def test_mock_episodes_default(self): + client = MockTVDBClient() + client.authenticate() + result = client.get_episode_durations("Unknown Show", 1) + assert len(result) == 22 + + def test_mock_episode_structure(self): + client = MockTVDBClient() + client.authenticate() + result = client.get_episode_durations("Test", 1) + ep = result[0] + assert ep["episode_number"] == 1 + assert ep["runtime"] == 45 + assert "name" in ep + assert "aired" in ep + + def test_mock_cached_episodes(self): + client = MockTVDBClient() + client.authenticate() + r1 = client.get_episode_durations("Test", 1) + r2 = client.get_episode_durations("Test", 1) + assert r1 is not None and r2 is not None