"""A source can be rejected for one title, and the page says where a copy
comes from.
2026-10-14, VisionQuest S01E01: vidcore's copy has ONE audio track, tagged
`eng`, and it is French. E02 from the same source is English. Nothing in the
stream can tell them apart -- the tag lies -- so the remux took it, the user
heard it, deleted it, pressed Download again ... and got the same French copy,
because vidcore is always tried first and a re-fetch had no way to skip it.
HdHub had a correct file the whole time (1080p x264, separate eng track).
So: a human verdict, recorded per FILE (a Kodi `tt…` job and a web `tmdb:…`
job name the same destination), that every ingest path honours -- web
download, season fetch, Kodi play-and-keep, restart resume all go through
_run_ingest. And the user asked for the other half the same day: the page
said "Download" and never which source it would download FROM.
Run: .venv/bin/python3 -m pytest test_source_reject.py -q
"""
import json
import os
import pytest
import addon
from test_hdhub import hdhub, ingest, isolated, probe, row, tmdb # noqa: F401
TONY = ("movie", "tmdb:1329016", 0, 0)
@pytest.fixture(autouse=True)
def rejects_file(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "SOURCE_REJECTS_FILE", str(tmp_path / "rejects.json"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "queue.json"))
monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "mp4-cache"))
yield
def _dest(kind="movie", ident="tmdb:1329016", season=0, episode=0):
meta = addon._library_meta(kind, ident, season, episode)
return addon._library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
# --- the ingest honours it --------------------------------------------------
def test_a_rejected_vidcore_is_never_resolved(ingest, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: pytest.fail("vidcore was rejected"))
addon._reject_source(_dest(), "vidcore")
assert addon._run_ingest(*TONY, 1080) == "done"
assert ingest, "HdHub must supply the copy"
def test_every_source_rejected_fetches_nothing_and_says_why(ingest, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: pytest.fail("vidcore was rejected"))
addon._reject_source(_dest(), "vidcore")
addon._reject_source(_dest(), "hdhub")
state = addon._run_ingest(*TONY, 1080)
assert state.startswith("failed") and "rejected" in state
assert ingest == []
def test_a_rejected_hdhub_is_not_the_fallback(ingest, hdhub):
addon._reject_source(_dest(), "hdhub")
state = addon._run_ingest(*TONY, 1080)
assert state.startswith("failed: ffmpeg exit 8") and ingest == []
assert hdhub.calls == []
def test_a_rejection_is_per_file_not_per_show(ingest, monkeypatch):
"""Rejecting vidcore for one copy must not touch any other title."""
addon._reject_source("/somewhere/else.mp4", "vidcore")
tried = []
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: tried.append(a) or None)
addon._run_ingest(*TONY, 1080)
assert tried
def test_a_rejection_survives_a_restart(ingest, tmp_path):
addon._reject_source(_dest(), "vidcore")
data = json.loads((tmp_path / "rejects.json").read_text())
assert data == {os.path.abspath(_dest()): ["vidcore"]}
assert addon._rejected_sources(_dest()) == {"vidcore"}
def test_allowing_again_undoes_it(ingest):
addon._reject_source(_dest(), "vidcore")
addon._allow_source(_dest(), "vidcore")
assert addon._rejected_sources(_dest()) == set()
def test_an_unreadable_rejects_file_is_never_overwritten(ingest, tmp_path):
"""Same shape as the 2026-08-31 queue loss: a writer that reads garbage as
'nothing' would persist nothing over every rejection on record."""
p = tmp_path / "rejects.json"
p.write_text("{not json")
addon._reject_source(_dest(), "vidcore")
assert p.read_text() == "{not json"
def test_an_unreadable_rejects_file_rejects_nothing(ingest):
"""A reader that cannot read must not stop a working source."""
open(addon.SOURCE_REJECTS_FILE, "w").write("{not json")
assert addon._rejected_sources(_dest()) == set()
def test_the_ingest_order_skips_rejected_sources(ingest):
assert addon._ingest_order(_dest()) == ["vidcore", "hdhub"]
addon._reject_source(_dest(), "vidcore")
assert addon._ingest_order(_dest()) == ["hdhub"]
# --- the web page ------------------------------------------------------------
@pytest.fixture
def web(ingest, monkeypatch):
started = []
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: started.append((a, k)) or "running")
monkeypatch.setattr(addon, "_tmdb_get", _web_tmdb(addon._tmdb_get))
c = addon.app.test_client()
c.started = started
return c
def _web_tmdb(inner):
def fake(path, params=None):
if path == "movie/1329016/credits":
return {}
return inner(path, params)
return fake
def _hold_fetched(source="vidcore"):
"""A copy this addon fetched, from `source`, as history records it."""
dest = _dest()
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x" * 1000)
addon._set_job_state("movie:tmdb:1329016:0:0:1080", "done",
args=("movie", "tmdb:1329016", 0, 0, 1080))
addon._append_history({"outcome": "done", "dest": dest, "source": source,
"kind": "movie", "ident": "tmdb:1329016"})
return dest
def test_the_download_button_names_its_source(web):
page = web.get("/web/title/movie/1329016").get_data(as_text=True)
assert "from vidcore (flixlab), else HdHub" in page
def test_after_a_rejection_the_button_names_the_other_source(web):
addon._reject_source(_dest(), "vidcore")
page = web.get("/web/title/movie/1329016").get_data(as_text=True)
assert "from HdHub" in page and "vidcore (flixlab) rejected" in page
assert "/web/source/allow" in page
def test_a_held_copy_says_where_it_came_from(web, monkeypatch):
_hold_fetched("hdhub")
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
page = web.get("/web/title/movie/1329016").get_data(as_text=True)
assert "in library · HdHub" in page
def test_a_running_download_says_where_from():
html = addon._web_state_html({"status": "running", "key": "k", "bytes": 5e8,
"source": "vidcore"})
assert "from vidcore (flixlab)" in html
def test_replace_rejects_the_copys_source_and_refetches(web, monkeypatch):
dest = _hold_fetched("vidcore")
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
r = web.post("/web/replace", data={"kind": "movie", "tmdb": "1329016",
"height": "1080"})
assert r.status_code == 303
assert addon._rejected_sources(dest) == {"vidcore"}
(args, kwargs), = web.started
assert args == ("movie", "tmdb:1329016", 0, 0, 1080) and kwargs == {"force": True}
assert os.path.exists(dest), "the old copy stays until the new one is whole"
def test_replace_refuses_a_copy_whose_source_is_unknown(web, monkeypatch):
dest = _hold_fetched(None)
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
r = web.post("/web/replace", data={"kind": "movie", "tmdb": "1329016"})
assert r.status_code == 409 and web.started == []
assert addon._rejected_sources(dest) == set()
def test_replace_refuses_a_copy_this_addon_did_not_fetch(web, monkeypatch):
"""The forced re-fetch overwrites the file -- same scope rule as Delete."""
dest = _dest()
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x")
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
r = web.post("/web/replace", data={"kind": "movie", "tmdb": "1329016"})
assert r.status_code == 404 and web.started == []
def test_replace_refuses_when_no_other_source_is_left(web, monkeypatch):
dest = _hold_fetched("vidcore")
addon._reject_source(dest, "hdhub")
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
r = web.post("/web/replace", data={"kind": "movie", "tmdb": "1329016"})
assert r.status_code == 409 and web.started == []
def test_replace_is_post_only(web):
assert web.get("/web/replace").status_code == 405
def test_reject_and_allow_routes(web):
form = {"kind": "movie", "tmdb": "1329016", "source": "vidcore"}
assert web.post("/web/source/reject", data=form).status_code == 303
assert addon._rejected_sources(_dest()) == {"vidcore"}
assert web.post("/web/source/allow", data=form).status_code == 303
assert addon._rejected_sources(_dest()) == set()
def test_an_unknown_source_cannot_be_rejected(web):
r = web.post("/web/source/reject",
data={"kind": "movie", "tmdb": "1329016", "source": "../x"})
assert r.status_code == 400
def test_the_watch_page_offers_a_replace_from_the_other_source(web, monkeypatch):
_hold_fetched("vidcore")
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
monkeypatch.setattr(addon, "_ffprobe_file", lambda p: None)
page = web.get("/web/watch/movie/1329016").get_data(as_text=True)
assert "Bad copy — replace from HdHub" in page and "/web/replace" in page