test: add test suite with 90% coverage

- 176 unit tests covering ConfigManager, FileClassifier, EpisodeRenamer,
  TVDBClient, MockTVDBClient, TVDBCache, and episode_matcher CLI
- Tests organized in tests/ with conftest.py shared fixtures
- pytest-cov configured with 90% coverage threshold in pyproject.toml
- CI workflow updated with COVERAGE_CORE=sysmon for Python 3.13 compat
- .gitignore updated for coverage artifacts and venv
This commit is contained in:
Jarian 2026-07-06 14:31:22 +00:00
parent 18b0683840
commit f1aa2c4cda
11 changed files with 1623 additions and 4 deletions

View File

@ -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:

7
.gitignore vendored
View File

@ -8,4 +8,9 @@ dist/
build/
.config.json
.env
src/.tvdb_cache/*
src/.tvdb_cache/*
.venv/
.coverage
htmlcov/
.pytest_cache/
config.json

View File

@ -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\\(",
"...",
]

0
tests/__init__.py Normal file
View File

140
tests/conftest.py Normal file
View File

@ -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

149
tests/test_config.py Normal file
View File

@ -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()

View File

@ -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

View File

@ -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('a<b>c: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

View File

@ -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

236
tests/test_tvdb_cache.py Normal file
View File

@ -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

224
tests/test_tvdb_client.py Normal file
View File

@ -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