backend/tests/test_ai_quality.py
Rama master · réplica de lectura del repositorio original.
"""Tests for the advisory AI subtitle-quality verdict (services.ai_quality)."""
import json
from pathlib import Path
import pytest
import services.ai_quality as aq
# ---- Cue sampling ---------------------------------------------------------------
def _write_srt(path: Path, lines: list[str]) -> str:
blocks = []
for i, text in enumerate(lines, start=1):
start = f"00:00:{i:02d},000"
end = f"00:00:{i:02d},900"
blocks.append(f"{i}\n{start} --> {end}\n{text}\n")
path.write_text("\n".join(blocks), encoding="utf-8")
return str(path)
def test_sample_cues_dedupes_and_skips_short(tmp_path):
path = _write_srt(
tmp_path / "a.de.srt",
["Hallo Welt!", "Hallo Welt!", "OK", "Wie geht es dir heute?"],
)
cues = aq.sample_cues(path, 30)
# duplicate removed, "OK" (<3 chars) removed
assert cues == ["Hallo Welt!", "Wie geht es dir heute?"]
def test_sample_cues_caps_and_spreads(tmp_path):
path = _write_srt(tmp_path / "a.de.srt", [f"Zeile Nummer {i} im Test" for i in range(50)])
cues = aq.sample_cues(path, 10)
assert len(cues) == 10
# even spread: first sample is from the start, samples strictly ordered
assert cues[0] == "Zeile Nummer 0 im Test"
indexes = [int(c.split()[2]) for c in cues]
assert indexes == sorted(indexes)
def test_sample_cues_truncates_long_lines(tmp_path):
path = _write_srt(tmp_path / "a.de.srt", ["x" * 500, "kurze Zeile hier"])
cues = aq.sample_cues(path, 30)
assert max(len(c) for c in cues) <= aq._MAX_CUE_CHARS
# ---- Deterministic scoring pieces -----------------------------------------------
def test_encoding_damage_thresholds():
clean = ["Alles gut hier"] * 10
assert aq._measure_encoding_damage(clean) == 0
one_bad = clean[:9] + ["kaputt � hier"]
assert aq._measure_encoding_damage(one_bad) == 1
two_bad = clean[:8] + ["kaputt � hier", "ärgerlich"]
assert aq._measure_encoding_damage(two_bad) == 2
mostly_bad = ["“Zitatâ€\x9d"] * 4 + clean[:6]
assert aq._measure_encoding_damage(mostly_bad) == 3
assert aq._measure_encoding_damage([]) == 0
def test_clamp_scores_tolerates_garbage():
scores = aq._clamp_scores(
{"machine_translation": "7", "ocr_artifacts": None, "grammar": -2}, encoding_damage=1
)
assert scores == {
"machine_translation": 3,
"ocr_artifacts": 0,
"grammar": 0,
"encoding_damage": 1,
}
@pytest.mark.parametrize(
"scores,expected",
[
(
{"machine_translation": 0, "ocr_artifacts": 0, "grammar": 0, "encoding_damage": 0},
"green",
),
(
{"machine_translation": 1, "ocr_artifacts": 1, "grammar": 0, "encoding_damage": 0},
"green",
),
(
{"machine_translation": 2, "ocr_artifacts": 0, "grammar": 0, "encoding_damage": 0},
"yellow",
),
(
{"machine_translation": 1, "ocr_artifacts": 1, "grammar": 1, "encoding_damage": 0},
"yellow",
),
({"machine_translation": 3, "ocr_artifacts": 0, "grammar": 0, "encoding_damage": 0}, "red"),
({"machine_translation": 2, "ocr_artifacts": 2, "grammar": 2, "encoding_damage": 0}, "red"),
],
)
def test_derive_verdict(scores, expected):
assert aq._derive_verdict(scores) == expected
def test_clean_reasons_caps_and_filters():
raw = [" gut ", 42, None, "x" * 500, "a", "b", "c", "d"]
reasons = aq._clean_reasons(raw)
assert len(reasons) == aq._MAX_REASONS
assert reasons[0] == "gut"
assert all(len(r) <= aq._MAX_REASON_CHARS for r in reasons)
assert aq._clean_reasons("not a list") == []
def test_parse_verdict_json_tolerates_prose():
parsed = aq._parse_verdict_json('Sure! {"grammar": 2, "reasons": ["x"]} hope that helps')
assert parsed["grammar"] == 2
with pytest.raises(ValueError):
aq._parse_verdict_json("no json here")
# ---- analyze_file with mocked LLM ------------------------------------------------
def test_analyze_file_success(tmp_path, monkeypatch):
path = _write_srt(tmp_path / "a.de.srt", [f"Ein ganz normaler Satz {i}" for i in range(10)])
monkeypatch.setattr(
aq,
"_call_ollama",
lambda cues, language, settings: (
{
"machine_translation": 2,
"ocr_artifacts": 0,
"grammar": 1,
"reasons": ["klingt wörtlich"],
},
"test-model",
),
)
result = aq.analyze_file(path, "de")
assert result is not None
assert result["verdict"] == "yellow"
assert result["scores"]["machine_translation"] == 2
assert result["reasons"] == ["klingt wörtlich"]
assert result["model"] == "test-model"
assert result["sampled_cues"] == 10
def test_analyze_file_too_few_cues_returns_none(tmp_path, monkeypatch):
path = _write_srt(tmp_path / "a.de.srt", ["Nur ein Satz hier"])
called = []
monkeypatch.setattr(aq, "_call_ollama", lambda *a: called.append(1))
assert aq.analyze_file(path, "de") is None
assert not called
def test_analyze_file_llm_error_returns_none(tmp_path, monkeypatch):
path = _write_srt(tmp_path / "a.de.srt", [f"Ein ganz normaler Satz {i}" for i in range(10)])
def _boom(cues, language, settings):
raise RuntimeError("ollama down")
monkeypatch.setattr(aq, "_call_ollama", _boom)
assert aq.analyze_file(path, "de") is None
def test_analyze_file_missing_file_returns_none():
assert aq.analyze_file("/nonexistent/x.de.srt", "de") is None
# ---- Persistence roundtrip -------------------------------------------------------
def test_save_get_batch_delete_roundtrip(app_ctx):
from db.quality import (
delete_ai_quality_result,
get_ai_quality_result,
get_ai_quality_results_for_paths,
save_ai_quality_result,
)
saved = save_ai_quality_result(
"/media/a/ep1.de.ass", "de", "yellow", '{"grammar": 2}', '["x"]', "m", 30
)
assert saved["verdict"] == "yellow"
# replace-on-save: second save for the same path leaves exactly one row
save_ai_quality_result("/media/a/ep1.de.ass", "de", "green", "{}", "[]", "m", 30)
row = get_ai_quality_result("/media/a/ep1.de.ass")
assert row["verdict"] == "green"
save_ai_quality_result("/media/a/ep2.de.ass", "de", "red", "{}", "[]", "m", 12)
batch = get_ai_quality_results_for_paths(
["/media/a/ep1.de.ass", "/media/a/ep2.de.ass", "/media/a/missing.de.ass"]
)
assert set(batch.keys()) == {"/media/a/ep1.de.ass", "/media/a/ep2.de.ass"}
assert delete_ai_quality_result("/media/a/ep1.de.ass") == 1
assert get_ai_quality_result("/media/a/ep1.de.ass") is None
assert get_ai_quality_results_for_paths([]) == {}
def test_attach_ai_quality(app_ctx):
from db.quality import save_ai_quality_result
save_ai_quality_result(
"/media/a/ep1.de.ass",
"de",
"red",
json.dumps({"machine_translation": 3}),
json.dumps(["MTL"]),
"m",
30,
)
entries = [
{"file_path": "/media/a/ep1.mkv", "language": "de", "format": "ass"},
{"file_path": "/media/a/ep2.mkv", "language": "de", "format": "srt"},
{"file_path": "", "language": "", "format": ""},
]
aq.attach_ai_quality(entries)
assert entries[0]["ai_quality"]["verdict"] == "red"
assert entries[0]["ai_quality"]["scores"] == {"machine_translation": 3}
assert entries[0]["ai_quality"]["reasons"] == ["MTL"]
assert entries[1]["ai_quality"] is None
assert entries[2]["ai_quality"] is None
# ---- maybe_queue_analysis guardrails ---------------------------------------------
def test_maybe_queue_analysis_disabled_is_noop(app_ctx, monkeypatch):
submitted = []
monkeypatch.setattr("services.background_tasks.submit_background", submitted.append)
aq.maybe_queue_analysis("/media/a/ep1.mkv", "de", "ass")
assert not submitted
def test_maybe_queue_analysis_never_raises_without_context():
# No app context, no settings guarantees — must still be silent.
aq.maybe_queue_analysis("", "", "")
# ---- Routes ---------------------------------------------------------------------
def test_route_get_requires_path(client):
resp = client.get("/api/v1/quality/ai")
assert resp.status_code == 400
def test_route_get_rejects_outside_media(client):
resp = client.get("/api/v1/quality/ai?path=/etc/passwd")
assert resp.status_code == 403
def test_route_get_returns_null_then_result(client, tmp_path):
target = str(tmp_path / "ep1.de.ass")
resp = client.get(f"/api/v1/quality/ai?path={target}")
assert resp.status_code == 200
assert resp.get_json()["result"] is None
from db.quality import save_ai_quality_result
save_ai_quality_result(target, "de", "yellow", '{"grammar": 2}', '["x"]', "m", 30)
resp = client.get(f"/api/v1/quality/ai?path={target}")
body = resp.get_json()["result"]
assert body["verdict"] == "yellow"
assert body["scores"] == {"grammar": 2}
def test_route_analyze_disabled_returns_503(client, tmp_path):
resp = client.post(
"/api/v1/quality/ai/analyze", json={"path": str(tmp_path / "a.de.srt"), "language": "de"}
)
assert resp.status_code == 503
def test_route_analyze_enabled_flow(client, tmp_path, monkeypatch):
from config import get_settings
settings = get_settings()
monkeypatch.setattr(settings, "ai_quality_enabled", True, raising=False)
resp = client.post("/api/v1/quality/ai/analyze", json={})
assert resp.status_code == 400
resp = client.post("/api/v1/quality/ai/analyze", json={"path": "/etc/passwd"})
assert resp.status_code == 403
missing = str(tmp_path / "missing.de.srt")
resp = client.post("/api/v1/quality/ai/analyze", json={"path": missing, "language": "de"})
assert resp.status_code == 404
real = _write_srt(tmp_path / "real.de.srt", [f"Ein Satz {i} zum Testen" for i in range(6)])
ran = []
monkeypatch.setattr(
"services.background_tasks.submit_background", lambda fn, *a, **kw: ran.append(fn)
)
resp = client.post("/api/v1/quality/ai/analyze", json={"path": real, "language": "de"})
assert resp.status_code == 202
assert len(ran) == 1
def test_a_verdict_for_the_bare_sidecar_is_found_from_history(app_ctx):
"""#61 A29: analysis fell back to Film.srt, the history looked up Film.de.srt."""
from db.quality import save_ai_quality_result
from services.ai_quality import attach_ai_quality
save_ai_quality_result("/media/m/Film.srt", "de", "yellow", "{}", "[]", "m", 9)
entries = [{"file_path": "/media/m/Film.mkv", "language": "de", "format": "srt"}]
attach_ai_quality(entries)
assert entries[0]["ai_quality"]["verdict"] == "yellow"
def test_the_language_sidecar_wins_over_the_bare_one(app_ctx):
from db.quality import save_ai_quality_result
from services.ai_quality import attach_ai_quality
save_ai_quality_result("/media/m/Show.srt", "de", "red", "{}", "[]", "m", 9)
save_ai_quality_result("/media/m/Show.de.srt", "de", "green", "{}", "[]", "m", 9)
entries = [{"file_path": "/media/m/Show.mkv", "language": "de", "format": "srt"}]
attach_ai_quality(entries)
assert entries[0]["ai_quality"]["verdict"] == "green"