"""The web UI: find a title in a browser and add it to the library (labs#12).
Until this existed a download could only start from Kodi -- a play, an action
row, or a title already on /library. This page starts from the TITLE, and feeds
the same pipeline: _ensure_ingest / _start_season_job, ident `tmdb:<id>` (what
Kodi's fxl rows use, so both share one job key).
Every action is POST. Kodi fires speculative GETs at this service, and a GET
must never be able to start a download.
Run: .venv/bin/python3 -m pytest test_web_ui.py -q
"""
import os
import pytest
import addon
TMDB = {
"trending/tv/week": {"results": [{"id": 1, "name": "Show",
"first_air_date": "2020-01-01"}]},
"trending/movie/week": {"results": [{"id": 2, "title": "Film",
"release_date": "2021-05-05"}]},
"tv/popular": {"results": []},
"movie/popular": {"results": []},
"search/tv": {"results": [{"id": 1, "name": "Show <b>bold</b>",
"first_air_date": "2020-01-01"}]},
"search/movie": {"results": [{"id": 2, "title": "Film",
"release_date": "2021-05-05"}]},
"movie/2": {"title": "Film", "release_date": "2021-05-05",
"overview": "A film.", "runtime": 100},
"tv/1": {"name": "Show", "first_air_date": "2020-01-01", "overview": "A show.",
"seasons": [{"season_number": 0}, {"season_number": 1},
{"season_number": 2}]},
"tv/1/season/1": {"episodes": [
{"episode_number": 1, "name": "Pilot", "air_date": "2020-01-01"},
{"episode_number": 2, "name": "Future", "air_date": "2999-01-01"},
{"episode_number": 3, "name": "Undated"}]},
"tv/1/season/2": {"episodes": [
{"episode_number": 1, "name": "Two", "air_date": "2021-01-01"}]},
"tv/1/season/1/episode/1": {"name": "Pilot", "runtime": 50},
"tv/1/season/1/episode/2": {"name": "Future", "runtime": 50},
"tv/1/season/1/episode/3": {"name": "Undated", "runtime": 50},
"tv/1/season/2/episode/1": {"name": "Two", "runtime": 50},
}
@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, "_duration_cache", {})
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_season_jobs", {})
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_tmdb_get",
lambda path, params=None: TMDB[path])
yield
@pytest.fixture
def started(monkeypatch):
"""Record what would have been started, without starting a thread -- a
real job here would reach vidcore."""
calls = {"mp4": [], "season": []}
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: calls["mp4"].append(a) or "running")
monkeypatch.setattr(addon, "_start_season_job",
lambda *a, **k: calls["season"].append(a) or "running")
return calls
@pytest.fixture
def client():
return addon.app.test_client()
def _hold(kind, season=0, episode=0, tmdb="1"):
"""Put a complete copy in the library."""
meta = addon._library_meta(kind, f"tmdb:{tmdb}", season, episode)
path = addon._library_path(kind, meta["title"], meta["year"],
meta["tmdb_id"], season, episode,
meta["episode_title"])
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"x" * 1024)
return path
# --- discovery --------------------------------------------------------------
def test_home_has_a_search_form_and_trending_titles(client):
body = client.get("/").get_data(as_text=True)
assert "action='/web/search'" in body
assert "/web/title/series/1" in body
assert "/web/title/movie/2" in body
def test_home_keeps_the_install_url_and_the_library_link(client):
body = client.get("/").get_data(as_text=True)
assert "/manifest.json" in body
assert "href='/library'" in body
def test_search_lists_matches_linking_to_their_title_page(client):
body = client.get("/web/search?q=show&type=series").get_data(as_text=True)
assert "/web/title/series/1" in body
assert "/web/title/movie/2" not in body
def test_search_without_a_type_searches_both(client):
body = client.get("/web/search?q=x").get_data(as_text=True)
assert "/web/title/series/1" in body and "/web/title/movie/2" in body
def test_tmdb_text_is_escaped(client):
body = client.get("/web/search?q=show&type=series").get_data(as_text=True)
assert "<b>bold</b>" not in body
assert "<b>bold</b>" in body
def test_a_tmdb_failure_is_a_message_not_a_crash(client, monkeypatch):
def down(path, params=None):
raise RuntimeError("tmdb down")
monkeypatch.setattr(addon, "_tmdb_get", down)
assert client.get("/").status_code == 200
r = client.get("/web/title/movie/2")
assert r.status_code == 502
assert "TMDB" in r.get_data(as_text=True)
# --- title pages ------------------------------------------------------------
def test_a_movie_not_held_offers_download(client):
body = client.get("/web/title/movie/2").get_data(as_text=True)
assert "action='/web/download'" in body
assert "name='tmdb' value='2'" in body
def test_a_held_movie_says_so_and_offers_no_download(client):
_hold("movie", tmdb="2")
body = client.get("/web/title/movie/2").get_data(as_text=True)
assert "in library" in body
assert "action='/web/download'" not in body
def test_an_aired_episode_offers_download_and_unaired_ones_do_not(client):
body = client.get("/web/title/series/1?s=1").get_data(as_text=True)
assert "name='episode' value='1'" in body
assert "name='episode' value='2'" not in body # airs 2999
assert "name='episode' value='3'" not in body # no air date = unreleased
assert body.count("not aired") == 2
def test_series_lists_real_seasons_not_specials(client):
body = client.get("/web/title/series/1").get_data(as_text=True)
assert "?s=1" in body and "?s=2" in body
assert "?s=0" not in body
def test_season_parameter_selects_the_season(client):
body = client.get("/web/title/series/1?s=2").get_data(as_text=True)
assert "Two" in body and "Pilot" not in body
def test_a_held_episode_says_in_library(client):
_hold("series", 1, 1)
body = client.get("/web/title/series/1?s=1").get_data(as_text=True)
assert "in library" in body
assert "name='episode' value='1'" not in body
def test_a_running_ingest_shows_progress_and_refreshes(client):
addon._mp4_jobs["series:tmdb:1:1:1:1080"] = "running"
path = _hold("series", 1, 1)
os.rename(path, addon._part_path(path)) # in flight, not finished
body = client.get("/web/title/series/1?s=1").get_data(as_text=True)
assert "downloading" in body
assert "MB" in body
assert "http-equiv=refresh" in body
def test_a_page_with_nothing_running_does_not_refresh(client):
body = client.get("/web/title/series/1?s=1").get_data(as_text=True)
assert "http-equiv=refresh" not in body
def test_a_failed_ingest_says_why_and_can_be_retried(client):
addon._mp4_jobs["movie:tmdb:2:0:0:1080"] = "failed: could not resolve title"
body = client.get("/web/title/movie/2").get_data(as_text=True)
assert "could not resolve title" in body
assert "action='/web/download'" in body
def test_an_unknown_kind_is_404(client):
assert client.get("/web/title/anime/1").status_code == 404
# --- the action -------------------------------------------------------------
def test_download_an_episode_starts_one_job_with_the_tmdb_ident(client, started):
r = client.post("/web/download", data={
"kind": "series", "tmdb": "1", "season": "1", "episode": "1",
"height": "720"})
assert r.status_code == 303
assert r.headers["Location"].endswith("/web/title/series/1?s=1")
assert started["mp4"] == [("series", "tmdb:1", 1, 1, 720)]
def test_download_a_movie(client, started):
r = client.post("/web/download", data={"kind": "movie", "tmdb": "2"})
assert r.status_code == 303
assert r.headers["Location"].endswith("/web/title/movie/2")
assert started["mp4"] == [("movie", "tmdb:2", 0, 0, 1080)]
def test_download_of_a_held_title_starts_nothing(client, started):
_hold("movie", tmdb="2")
client.post("/web/download", data={"kind": "movie", "tmdb": "2"})
assert started["mp4"] == []
def test_download_a_season(client, started):
client.post("/web/download", data={
"kind": "series", "tmdb": "1", "season": "1", "scope": "season"})
assert started["season"] == [("tmdb:1", 1, 1080)]
assert started["mp4"] == []
def test_a_get_can_never_start_a_download(client, started):
r = client.get("/web/download?kind=movie&tmdb=2")
assert r.status_code == 405
assert started["mp4"] == [] and started["season"] == []
@pytest.mark.parametrize("data", [
{"kind": "movie", "tmdb": "2x"},
{"kind": "movie", "tmdb": "../2"},
{"kind": "anime", "tmdb": "2"},
{"kind": "movie", "tmdb": "2", "height": "2160"},
{"kind": "series", "tmdb": "1", "season": "one", "episode": "1"},
{"kind": "series", "tmdb": "1", "season": "0", "episode": "0"},
])
def test_bad_input_is_400_and_starts_nothing(client, started, data):
assert client.post("/web/download", data=data).status_code == 400
assert started["mp4"] == [] and started["season"] == []
def test_a_bad_season_parameter_is_400_not_a_tmdb_error(client):
"""abort() inside the page must not be swallowed into the 502 path."""
assert client.get("/web/title/series/1?s=abc").status_code == 400