"""The library audit: is what we fetched actually all there?
Strange New Worlds S04E04 sat in the library as 10.6 minutes of a 51-minute
episode for five days. Nothing surfaced it -- it was found by watching it. Every
number needed to catch it was already on the machine, so this page is mostly an
admission that having the numbers is not the same as showing them.
Scope is the ingest queue, not the whole library: those are the copies this
addon fetched, and the only ones the button can act on.
Run: .venv/bin/python3 -m pytest test_library_audit.py -q
"""
import json
import os
import pytest
import addon
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
"""Every store this touches lives beside addon.py in the deployed
checkout. A test that read the real ingest queue would reach TMDB for
seventeen real titles; one that wrote the real verified store would mark
production files trustworthy."""
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE",
os.path.join(str(tmp_path), "queue.json"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE",
os.path.join(str(tmp_path), "verified.json"))
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
yield
@pytest.fixture
def library(monkeypatch, tmp_path):
"""One fetched episode, 50 minutes per TMDB, whose length is settable."""
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01"},
"tv/1/season/1/episode/1": {"name": "Pilot", "runtime": 50},
}[path])
with open(addon.INGEST_QUEUE_FILE, "w") as f:
json.dump({"series:tmdb:1:1:1:1080": {
"state": "done", "args": ["series", "tmdb:1", 1, 1, 1080],
"ts": 100.0}}, f)
meta = addon._library_meta("series", "tmdb:1", 1, 1)
path = addon._library_path("series", meta["title"], meta["year"],
meta["tmdb_id"], 1, 1, meta["episode_title"])
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"x" * 1024)
seconds = {"value": 3000.0}
monkeypatch.setattr(addon, "_media_duration", lambda p: seconds["value"])
return path, seconds
@pytest.fixture
def client():
return addon.app.test_client()
# --- what the audit reports -------------------------------------------------
def test_a_complete_copy_reads_ok(library):
assert addon._audit_rows()[0]["status"] == addon.AUDIT_OK
def test_a_short_copy_is_flagged(library):
_, seconds = library
seconds["value"] = 636.7
row = addon._audit_rows()[0]
assert row["status"] == addon.AUDIT_SHORT
assert round(row["pct"]) == 21
def test_a_copy_that_is_gone_reads_missing(library):
path, _ = library
os.remove(path)
assert addon._audit_rows()[0]["status"] == addon.AUDIT_MISSING
def test_a_copy_with_nothing_to_measure_against_is_not_called_ok(library, monkeypatch):
"""Present and playable, but unchecked. Folding that into "ok" would be
claiming a verification that never happened."""
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01"},
"tv/1/season/1/episode/1": {"name": "Pilot"},
}[path])
monkeypatch.setattr(addon, "_library_meta_cache", {})
assert addon._audit_rows()[0]["status"] == addon.AUDIT_UNKNOWN
def test_a_source_verified_copy_says_so(library):
path, _ = library
assert addon._audit_rows()[0]["verified"] is False
addon._record_verified(path, 3000.0, 3000.0)
assert addon._audit_rows()[0]["verified"] is True
def test_a_queue_entry_with_no_arguments_is_skipped(library):
"""Pre-resume entries cannot be re-run, so offering a button would lie."""
with open(addon.INGEST_QUEUE_FILE, "w") as f:
json.dump({"old:key": {"state": "done", "ts": 1.0}}, f)
assert addon._audit_rows() == []
# --- the page ---------------------------------------------------------------
def test_the_page_renders_and_counts_the_bad_ones(library, client):
_, seconds = library
seconds["value"] = 636.7
body = client.get("/library").get_data(as_text=True)
assert "1 incomplete" in body and "Re-download" in body
def test_a_title_cannot_inject_markup(library, client, monkeypatch):
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "<script>x</script>", "first_air_date": "2020"},
"tv/1/season/1/episode/1": {"name": "Pilot", "runtime": 50},
}[path])
monkeypatch.setattr(addon, "_library_meta_cache", {})
body = client.get("/library").get_data(as_text=True)
assert "<script>x</script>" not in body
def test_the_json_endpoint_carries_the_same_verdicts(library, client):
_, seconds = library
seconds["value"] = 636.7
rows = client.get("/library/audit.json").get_json()["rows"]
assert rows[0]["status"] == addon.AUDIT_SHORT
# --- the button -------------------------------------------------------------
def test_re_ingest_is_not_reachable_by_a_get(client):
"""Kodi issues speculative GETs all over this service -- the parent-directory
scan on every play is already a known 404. A re-download must never be
something a stray fetch can start."""
assert client.get("/library/reingest").status_code == 405
def test_re_ingest_starts_a_forced_job(library, client, monkeypatch):
calls = []
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: (calls.append((a, k)), "running")[1])
r = client.post("/library/reingest", data={
"kind": "series", "ident": "tmdb:1", "season": "1",
"episode": "1", "height": "1080", "force": "1"})
assert r.status_code == 303
assert calls[0][0] == ("series", "tmdb:1", 1, 1, 1080)
assert calls[0][1]["force"] is True
def test_re_ingest_rejects_a_bad_kind(client):
r = client.post("/library/reingest", data={"kind": "../x", "ident": "tmdb:1"})
assert r.status_code == 400
# --- forcing ----------------------------------------------------------------
def test_a_forced_ingest_ignores_a_copy_already_held(library, monkeypatch):
"""Without force, _run_ingest returns "done" for a file already on disk --
which is right for the automatic path and useless for a human saying the
copy is wrong."""
ran = []
monkeypatch.setattr(addon, "ingest_source", lambda: object())
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: "http://cdn/master.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: url)
monkeypatch.setattr(addon, "_remux_to_mp4",
lambda url, dest: ran.append(dest))
assert addon._run_ingest("series", "tmdb:1", 1, 1, 1080) == "done"
assert ran == []
addon._run_ingest("series", "tmdb:1", 1, 1, 1080, force=True)
assert len(ran) == 1
def test_forcing_drops_the_source_verification(library, monkeypatch):
"""The record attests the copy matched the source it came from. That is not
the question a human asking for a re-download is raising, so it must not
veto them -- and must not survive to vouch for a file being replaced."""
path, _ = library
addon._record_verified(path, 3000.0, 3000.0)
monkeypatch.setattr(addon, "ingest_source", lambda: object())
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: "http://cdn/master.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: url)
monkeypatch.setattr(addon, "_remux_to_mp4", lambda url, dest: None)
addon._run_ingest("series", "tmdb:1", 1, 1, 1080, force=True)
assert addon._load_verified() == {}