"""The two rows that DO something, in the list Jacktook shows before playing.
The completeness gate catches a copy that is short. It cannot catch the wrong
episode, the wrong season, or unwatchable audio -- and the person who can is in
front of the TV holding Kore, which cannot drive Kodi's own UI. The source list
is the one surface they can reach, so the escape hatch lives there.
These routes are GET, unlike /library/reingest, because a Stremio row is a URL
the player opens. Kodi opens a stream URL with GET + HEAD + GET, so each action
fires up to three times and the job starters' de-duplication is what stops one
click becoming three remuxes.
Run: .venv/bin/python3 -m pytest test_kodi_action_rows.py -q
"""
import os
import pytest
import addon
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
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, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_season_jobs", {})
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01"},
"tv/1/season/4/episode/4": {"name": "Four", "runtime": 50},
"movie/9": {"title": "Film", "release_date": "2020-01-01", "runtime": 100},
}[path])
yield
@pytest.fixture
def has_copy(monkeypatch):
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: "/some/file.mp4")
@pytest.fixture
def no_copy(monkeypatch):
monkeypatch.setattr(addon, "_playable_path", lambda *a, **k: None)
def _rows(kind="series", ident="tmdb:1", season=4, episode=4):
out = []
with addon.app.test_request_context("/"):
addon._append_action_rows(out, kind, ident, season, episode)
return out
def _titles(rows):
return " | ".join(r["title"] for r in rows)
# --- which rows are offered -------------------------------------------------
def test_a_held_copy_gets_a_re_download_row(has_copy):
assert "re-download" in _titles(_rows())
def test_no_copy_means_no_re_download_row(no_copy):
"""_append_mp4_row's "play & keep" already fetches one. Offering to redo
nothing would be a dead row."""
assert "re-download" not in _titles(_rows())
def test_a_series_always_gets_a_season_row(no_copy):
assert "get season 4" in _titles(_rows())
def test_a_movie_gets_no_season_row(has_copy):
titles = _titles(_rows("movie", "tmdb:9", 0, 0))
assert "re-download" in titles and "season" not in titles
def test_an_unidentifiable_title_gets_no_rows(has_copy):
assert _rows("series", "tmdb:404", 1, 1) == []
def test_the_rows_point_at_this_host(has_copy):
urls = [r["url"] for r in _rows()]
assert any("/refetch/series/tmdb:1:4:4.m3u8" in u for u in urls)
assert any("/season/series/tmdb:1:4:4.m3u8" in u for u in urls)
# --- the routes -------------------------------------------------------------
@pytest.fixture
def route(monkeypatch):
"""Routes with the work stubbed, recording what each would have started."""
jobs, seasons = [], []
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: (jobs.append((a, k)), "running")[1])
monkeypatch.setattr(addon, "_start_season_job",
lambda *a, **k: (seasons.append(a), "running")[1])
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: "http://cdn/master.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: f"{url}?h={h}")
return type("R", (), {"jobs": jobs, "seasons": seasons,
"client": addon.app.test_client()})()
def test_refetch_forces_a_re_ingest_and_plays_live(route):
r = route.client.get("/refetch/series/tmdb:1:4:4.m3u8?h=1080")
assert r.status_code == 302 and "cdn/master.m3u8" in r.headers["Location"]
assert route.jobs[0][0] == ("series", "tmdb:1", 4, 4, 1080)
assert route.jobs[0][1]["force"] is True
def test_the_season_row_queues_the_season_and_plays_the_episode(route):
r = route.client.get("/season/series/tmdb:1:4:4.m3u8?h=1080")
assert r.status_code == 302
assert route.seasons[0] == ("tmdb:1", 4, 1080)
assert route.jobs == [] # the episode is not forced too
def test_a_movie_refetch_works(route):
r = route.client.get("/refetch/movie/tmdb:9.m3u8?h=1080")
assert r.status_code == 302
assert route.jobs[0][0] == ("movie", "tmdb:9", 0, 0, 1080)
def test_a_malformed_ident_is_refused(route):
for url in ("/refetch/series/nope.m3u8", "/season/series/tmdb:1.m3u8",
"/refetch/movie/../etc.m3u8"):
assert route.client.get(url).status_code == 404
def test_an_unresolvable_title_does_not_pretend_to_play(route, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
assert route.client.get("/refetch/series/tmdb:1:4:4.m3u8").status_code == 404
# --- the de-duplication these routes lean on --------------------------------
def test_kodis_three_requests_start_one_job(monkeypatch):
"""Kodi opens a stream URL with GET + HEAD + GET. Without this guard one
click would be three remuxes of the same episode."""
started = []
monkeypatch.setattr(addon.threading, "Thread",
lambda *a, **k: type("T", (), {
"start": lambda self: started.append(k.get("args")),
"daemon": True})())
for _ in range(3):
addon._start_mp4_job("series", "tmdb:1", 4, 4, 1080, force=True)
assert len(started) == 1
def test_a_second_season_request_is_refused_while_one_runs(monkeypatch):
monkeypatch.setattr(addon, "_season_jobs", {"season:tmdb:1:4:1080": "running"})
started = []
monkeypatch.setattr(addon.threading, "Thread",
lambda *a, **k: type("T", (), {
"start": lambda self: started.append(1),
"daemon": True})())
assert addon._start_season_job("tmdb:1", 4, 1080) == "running"
assert started == []