backend/tests/test_dedup_detection.py
Rama master · réplica de lectura del repositorio original.
"""Duplicate detection tests: SHA-256 hash-based dedup at download time.
Tests:
- TestComputeContentHash: normalization, CRLF/LF, BOM, empty content
- TestDuplicateSubtitleError: error attributes and hierarchy
- TestSaveSubtitleDedup: first save, duplicate skip, stale hash cleanup,
cross-directory (not a duplicate), disabled via config
"""
import os
import sys
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from dedup_engine import compute_content_hash_from_bytes
from error_handler import DuplicateSubtitleError, SublarrError
# ── TestComputeContentHash ────────────────────────────────────────────────────
class TestComputeContentHash:
def test_returns_sha256_hex(self):
result = compute_content_hash_from_bytes(b"hello")
assert len(result) == 64
assert all(c in "0123456789abcdef" for c in result)
def test_deterministic(self):
assert compute_content_hash_from_bytes(b"abc") == compute_content_hash_from_bytes(b"abc")
def test_crlf_equals_lf(self):
crlf = compute_content_hash_from_bytes(b"line1\r\nline2\r\n")
lf = compute_content_hash_from_bytes(b"line1\nline2\n")
assert crlf == lf
def test_leading_trailing_whitespace_stripped(self):
stripped = compute_content_hash_from_bytes(b"content")
padded = compute_content_hash_from_bytes(b" content ")
assert stripped == padded
def test_different_content_different_hash(self):
assert compute_content_hash_from_bytes(b"aaa") != compute_content_hash_from_bytes(b"bbb")
def test_utf8_bom_handled(self):
# BOM is decoded as part of the string — normalization still applies
with_bom = compute_content_hash_from_bytes(b"\xef\xbb\xbfcontent")
assert len(with_bom) == 64 # just ensure it doesn't crash
def test_empty_bytes(self):
result = compute_content_hash_from_bytes(b"")
assert len(result) == 64
# ── TestDuplicateSubtitleError ────────────────────────────────────────────────
class TestDuplicateSubtitleError:
def test_is_sublarr_error(self):
err = DuplicateSubtitleError("abc123", "/existing.srt", "/new.srt")
assert isinstance(err, SublarrError)
def test_http_status_409(self):
err = DuplicateSubtitleError("abc123", "/existing.srt", "/new.srt")
assert err.http_status == 409
def test_attributes(self):
err = DuplicateSubtitleError("deadbeef", "/old/path.srt", "/new/path.srt")
assert err.content_hash == "deadbeef"
assert err.existing_path == "/old/path.srt"
assert err.attempted_path == "/new/path.srt"
def test_context_populated(self):
err = DuplicateSubtitleError("hash1", "/a.srt", "/b.srt")
assert err.context["content_hash"] == "hash1"
assert err.context["existing_path"] == "/a.srt"
# ── TestSaveSubtitleDedup ─────────────────────────────────────────────────────
def _make_result(content: bytes = b"1\n00:00:01,000 --> 00:00:02,000\nHello\n"):
from providers.base import SubtitleFormat, SubtitleResult
r = SubtitleResult.__new__(SubtitleResult)
r.content = content
r.format = SubtitleFormat.SRT
r.provider_name = "test_provider"
r.language = "de"
r.score = 100
r.subtitle_id = "sub123"
return r
def _make_manager():
from providers import ProviderManager
mgr = ProviderManager.__new__(ProviderManager)
mgr._providers = {}
return mgr
class TestSaveSubtitleDedup:
"""Tests for dedup logic inside ProviderManager.save_subtitle()."""
def _patch_save(self, tmp_path, dedup_enabled=True, repo_matches=None):
"""Return a context manager that patches all save_subtitle dependencies."""
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=dedup_enabled)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = repo_matches or []
return (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
mock_repo,
)
def test_first_save_writes_file(self, tmp_path):
"""First download should write the file normally."""
mgr = _make_manager()
result = _make_result()
output = str(tmp_path / "episode.de.srt")
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=True)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = []
with (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
):
saved = mgr.save_subtitle(result, output)
assert os.path.isfile(saved)
assert open(saved, "rb").read() == result.content
def test_duplicate_raises_error(self, tmp_path):
"""Re-downloading what is already on disk raises DuplicateSubtitleError.
This used to save to ``episode2.de.srt`` — a *different* episode in the
same folder — and assert a duplicate, which pinned the 1.14.2 bug in
place as expected behaviour. Identity is the media file now, so the
case the check actually exists for is the one covered here.
"""
mgr = _make_manager()
result = _make_result()
existing = str(tmp_path / "episode.de.srt")
open(existing, "wb").write(result.content)
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=True)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = [
{"file_path": existing, "format": "srt", "language": "de"}
]
with (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
pytest.raises(DuplicateSubtitleError) as exc_info,
):
mgr.save_subtitle(result, str(tmp_path / "episode.de.srt"))
assert exc_info.value.existing_path == existing
def test_cross_directory_not_duplicate(self, tmp_path):
"""Same hash in a different directory is not a duplicate."""
mgr = _make_manager()
result = _make_result()
other_dir = tmp_path / "other_series"
other_dir.mkdir()
existing = str(other_dir / "episode.de.srt")
open(existing, "wb").write(result.content)
output = str(tmp_path / "episode.de.srt")
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=True)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = [
{"file_path": existing, "format": "srt", "language": "de"}
]
with (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
):
saved = mgr.save_subtitle(result, output)
assert os.path.isfile(saved)
def test_stale_hash_cleaned_and_write_proceeds(self, tmp_path):
"""Stale hash entries (file deleted) are cleaned up and write proceeds."""
mgr = _make_manager()
result = _make_result()
stale_path = str(tmp_path / "deleted.de.srt") # does NOT exist on disk
output = str(tmp_path / "episode.de.srt")
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=True)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = [
{"file_path": stale_path, "format": "srt", "language": "de"}
]
with (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
):
saved = mgr.save_subtitle(result, output)
mock_repo.delete_hashes_by_paths.assert_called_once_with([stale_path])
assert os.path.isfile(saved)
def test_dedup_disabled_always_writes(self, tmp_path):
"""When dedup_on_download=False, no duplicate check is performed."""
mgr = _make_manager()
result = _make_result()
existing = str(tmp_path / "episode.de.srt")
open(existing, "wb").write(result.content)
output = str(tmp_path / "episode2.de.srt")
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=False)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = [
{"file_path": existing, "format": "srt", "language": "de"}
]
with (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
):
saved = mgr.save_subtitle(result, output)
assert os.path.isfile(saved)
mock_repo.find_by_content_hash.assert_not_called()
class TestDedupIdentityIsTheMediaFile:
"""A duplicate means "this exact subtitle is already here for THIS episode".
Regression cover for the 1.14.2 field report: a subtitle downloaded for
S01E04 that happened to be byte-identical to the sidecar already sitting
next to S01E02 was classified as a duplicate of it, because the only
identity test was "same directory". A season folder holds every episode of
a season, so that test is satisfied by any other episode.
The caller (``wanted_search.post_processor``) treats DuplicateSubtitleError
as a successful download: it deletes the wanted item and returns the OTHER
file's path. So a false duplicate silently drops the record that S01E04 is
still missing a subtitle.
"""
def _save(self, tmp_path, existing_name, attempt_name, content=None):
mgr = _make_manager()
result = _make_result()
if content is not None:
result.content = content
existing = str(tmp_path / existing_name)
with open(existing, "wb") as fh:
fh.write(result.content)
mock_settings = MagicMock(media_path=str(tmp_path), dedup_on_download=True)
mock_repo = MagicMock()
mock_repo.find_by_content_hash.return_value = [
{"file_path": existing, "format": "srt", "language": "de"}
]
with (
patch("config.get_settings", return_value=mock_settings),
patch("security_utils.is_safe_path", return_value=True),
patch("db.repositories.cleanup.CleanupRepository", return_value=mock_repo),
):
return mgr.save_subtitle(result, str(tmp_path / attempt_name)), existing
def test_other_episode_in_same_folder_is_not_a_duplicate(self, tmp_path):
saved, existing = self._save(tmp_path, "Show.S01E02.de.srt", "Show.S01E04.de.srt")
assert os.path.isfile(saved), "S01E04 must get its own sidecar written"
assert os.path.abspath(saved) != os.path.abspath(existing)
def test_same_episode_is_still_a_duplicate(self, tmp_path):
"""The case dedup exists for: re-downloading what is already there."""
with pytest.raises(DuplicateSubtitleError):
self._save(tmp_path, "Show.S01E02.de.srt", "Show.S01E02.de.srt")
def test_unparseable_filename_is_written_not_skipped(self, tmp_path):
"""Fail towards a redundant write, never towards a dropped wanted item."""
saved, _ = self._save(tmp_path, "weird-name.srt", "Show.S01E04.de.srt")
assert os.path.isfile(saved)