"""The download manager page (labs#13): what is running, and what happened.
Reads only what test_job_control's machinery already maintains: _job_info and
_season_info for live jobs, ingest_history.jsonl for the past. Cancel is POST
only, like every other action on this service.
Run: .venv/bin/python3 -m pytest test_web_downloads.py -q
"""
import json
import os
import time
import pytest
import addon
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "INGEST_HISTORY_FILE", str(tmp_path / "history.jsonl"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "queue.json"))
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "_job_info", {})
monkeypatch.setattr(addon, "_season_info", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_enough_space", lambda: True)
yield
@pytest.fixture
def client():
return addon.app.test_client()
def _live(tmp_path, phase="downloading", part_bytes=50_000_000, since_ago=10):
dest = str(tmp_path / "lib" / "Show - S02E06.mp4")
os.makedirs(os.path.dirname(dest), exist_ok=True)
if part_bytes:
with open(addon._part_path(dest), "wb") as f:
f.truncate(part_bytes)
now = time.time()
addon._job_info["series:tmdb:1:2:6:1080"] = {
"phase": phase, "source": "vidcore", "dest": dest, "started": now - 30,
"downloading_since": now - since_ago, "kind": "series",
"ident": "tmdb:1", "season": 2, "episode": 6, "height": 1080,
"label": "S02E06", "title": "Show <i>x</i>"}
def _history(*entries):
with open(addon.INGEST_HISTORY_FILE, "w") as f:
for e in entries:
f.write(json.dumps(e) + "\n")
def test_an_empty_manager_says_so(client):
body = client.get("/web/downloads").get_data(as_text=True)
assert "Nothing downloading" in body
assert "No downloads yet" in body
assert "http-equiv=refresh" not in body
def test_a_live_job_shows_title_phase_source_progress_and_speed(client, tmp_path):
_live(tmp_path)
body = client.get("/web/downloads").get_data(as_text=True)
assert "Show <i>x</i>" in body and "S02E06" in body
assert "downloading" in body and "vidcore" in body and "1080p" in body
assert "50 MB" in body
assert "MB/s" in body
assert "http-equiv=refresh" in body
def test_a_live_job_has_a_cancel_form(client, tmp_path):
_live(tmp_path)
body = client.get("/web/downloads").get_data(as_text=True)
assert "action='/web/cancel'" in body
assert "value='series:tmdb:1:2:6:1080'" in body
def test_a_resolving_job_shows_no_speed(client, tmp_path):
_live(tmp_path, phase="resolving", part_bytes=0)
body = client.get("/web/downloads").get_data(as_text=True)
assert "resolving" in body and "MB/s" not in body
def test_a_season_shows_its_position_and_its_own_cancel(client):
addon._season_info["season:tmdb:1:2:1080"] = {
"ident": "tmdb:1", "season": 2, "height": 1080, "title": "Show",
"episodes": [1, 2, 3], "index": 2, "current": "series:tmdb:1:2:2:1080",
"started": time.time()}
body = client.get("/web/downloads").get_data(as_text=True)
assert "episode 2 of 3" in body
assert "value='season:tmdb:1:2:1080'" in body
def test_history_lists_newest_first_with_outcome_source_and_size(client):
_history(
{"ts": 1000, "title": "Old", "label": "movie", "outcome": "done",
"source": "hdhub", "bytes": 2_500_000_000, "took": 300, "kind": "movie",
"ident": "tmdb:2", "season": 0, "episode": 0},
{"ts": 2000, "title": "New", "label": "S01E01",
"outcome": "failed: could not resolve title", "source": "vidcore",
"bytes": None, "took": 30, "kind": "series", "ident": "tmdb:1",
"season": 1, "episode": 1})
body = client.get("/web/downloads").get_data(as_text=True)
assert body.index("New") < body.index("Old")
assert "could not resolve title" in body
assert "hdhub" in body and "2.5 GB" in body
def test_history_links_back_to_the_title(client):
_history({"ts": 1, "title": "Show", "label": "S01E01", "outcome": "done",
"source": "vidcore", "bytes": 1, "took": 1, "kind": "series",
"ident": "tmdb:1", "season": 1, "episode": 1})
body = client.get("/web/downloads").get_data(as_text=True)
assert "/web/title/series/1?s=1" in body
def test_cancel_posts_to_the_job(client, monkeypatch):
asked = []
monkeypatch.setattr(addon, "_cancel_job", lambda k: asked.append(k) or True)
r = client.post("/web/cancel", data={"key": "series:tmdb:1:2:6:1080"})
assert r.status_code == 303
assert r.headers["Location"].endswith("/web/downloads")
assert asked == ["series:tmdb:1:2:6:1080"]
def test_cancel_routes_a_season_key_to_the_season(client, monkeypatch):
asked = []
monkeypatch.setattr(addon, "_cancel_season", lambda k: asked.append(k) or True)
client.post("/web/cancel", data={"key": "season:tmdb:1:2:1080"})
assert asked == ["season:tmdb:1:2:1080"]
def test_cancel_of_a_job_that_already_finished_is_harmless(client):
r = client.post("/web/cancel", data={"key": "series:tmdb:9:9:9:1080"})
assert r.status_code == 303
def test_a_get_can_never_cancel(client, monkeypatch):
asked = []
monkeypatch.setattr(addon, "_cancel_job", lambda k: asked.append(k))
assert client.get("/web/cancel?key=x").status_code == 405
assert asked == []
def test_every_page_links_to_the_manager(client):
assert "href='/web/downloads'" in client.get("/").get_data(as_text=True)
def test_a_cancelled_title_offers_download_again(client, monkeypatch):
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"movie/2": {"title": "Film", "release_date": "2021-01-01", "runtime": 100},
"trending/tv/week": {"results": []}}[path])
addon._mp4_jobs["movie:tmdb:2:0:0:1080"] = "cancelled"
body = client.get("/web/title/movie/2").get_data(as_text=True)
assert "cancelled" in body
assert "action='/web/download'" in body