"""Delete a copy found to be bad (labs#13). Immediate -- the user chose that
over a trash folder -- so the guard rails are about SCOPE, not undo:
- only copies this addon fetched (a queue entry names the same file). The
library is ~745 GB, the overwhelming majority not ours; this button must
not be able to touch any of it.
- never while a job holds the file (its dest lock).
- everything the copy brought: the mp4, its .srt sidecars, its verification
record, and any legacy mp4-cache copy of the same title.
- folders removed only if that leaves them EMPTY, and never above the library.
- recorded in history. No re-fetch block (the user's choice).
POST only.
Run: .venv/bin/python3 -m pytest test_web_delete.py -q
"""
import json
import os
import pytest
import addon
TMDB = {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01",
"seasons": [{"season_number": 2}]},
"tv/1/season/2": {"episodes": [
{"episode_number": 6, "name": "Six", "air_date": "2020-01-01"},
{"episode_number": 7, "name": "Seven", "air_date": "2020-01-01"}]},
"tv/1/season/2/episode/6": {"name": "Six", "runtime": 30},
"tv/1/season/2/episode/7": {"name": "Seven", "runtime": 30},
"movie/2": {"title": "Film", "release_date": "2021-05-05", "runtime": 30},
}
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "queue.json"))
monkeypatch.setattr(addon, "INGEST_HISTORY_FILE", str(tmp_path / "history.jsonl"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE", str(tmp_path / "verified.json"))
monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "mp4-cache"))
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_quality_cache", {})
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_dest_locks", {})
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: TMDB[path])
monkeypatch.setattr(addon, "_media_duration", lambda p: 1800.0)
monkeypatch.setattr(addon, "_ffprobe_file", lambda p: {"streams": [], "format": {}})
notified = []
monkeypatch.setattr(addon, "_notify_jellyfin",
lambda path, kind: notified.append((path, kind)),
raising=False)
yield notified
@pytest.fixture
def client():
return addon.app.test_client()
def _hold(kind, season=0, episode=0, tmdb="1", ours=True, ident=None):
ident = ident or f"tmdb:{tmdb}"
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" * 4096)
if ours:
addon._set_job_state(addon._mp4_key(kind, ident, season, episode, 1080),
"done", args=(kind, ident, season, episode, 1080))
return path
def _sidecars(path, *langs):
stem = os.path.splitext(path)[0]
out = []
for lang in langs:
p = f"{stem}.{lang}.srt"
with open(p, "w") as f:
f.write("1\n00:00:01,000 --> 00:00:02,000\nhi\n")
out.append(p)
return out
def _post(client, kind="series", tmdb="1", season="2", episode="6"):
return client.post("/web/delete", data={"kind": kind, "tmdb": tmdb,
"season": season, "episode": episode})
def test_delete_removes_the_copy_and_its_sidecars(client):
path = _hold("series", 2, 6)
subs = _sidecars(path, "en", "pt-br")
r = _post(client)
assert r.status_code == 303
assert r.headers["Location"].endswith("/web/title/series/1?s=2")
assert not os.path.exists(path)
assert not any(os.path.exists(s) for s in subs)
def test_delete_leaves_other_episodes_and_their_folder(client):
six = _hold("series", 2, 6)
seven = _hold("series", 2, 7)
_post(client)
assert not os.path.exists(six)
assert os.path.exists(seven)
def test_delete_does_not_touch_a_similarly_named_sidecar_of_another_episode(client):
six = _hold("series", 2, 6)
seven = _hold("series", 2, 7)
(other,) = _sidecars(seven, "en")
_post(client)
assert os.path.exists(other)
def test_emptied_folders_are_removed_but_never_the_library_root(client):
path = _hold("series", 2, 6)
season_dir = os.path.dirname(path)
show_dir = os.path.dirname(season_dir)
_post(client)
assert not os.path.exists(season_dir)
assert not os.path.exists(show_dir)
assert os.path.isdir(os.path.join(addon.LIBRARY_ROOT, "Shows"))
def test_delete_forgets_the_verification_record(client):
path = _hold("series", 2, 6)
addon._record_verified(path, 1800.0, 1800.0)
_post(client)
assert os.path.abspath(path) not in addon._load_verified()
def test_delete_removes_a_legacy_cache_copy_of_the_same_title(client):
path = _hold("series", 2, 6)
legacy = addon._mp4_path("series", "tmdb:1", 2, 6, 720)
os.makedirs(os.path.dirname(legacy), exist_ok=True)
with open(legacy, "wb") as f:
f.write(b"y")
_post(client)
assert not os.path.exists(legacy)
def test_delete_is_recorded_in_history(client):
path = _hold("series", 2, 6)
_post(client)
with open(addon.INGEST_HISTORY_FILE) as f:
(h,) = [json.loads(line) for line in f]
assert h["outcome"] == "deleted"
assert h["bytes"] == 4096
assert h["dest"] == path
assert h["label"] == "S02E06" and h["title"] == "Show"
def test_delete_marks_the_queue_entry_and_it_is_not_resumed(client, monkeypatch):
_hold("series", 2, 6)
_post(client)
with open(addon.INGEST_QUEUE_FILE) as f:
assert json.load(f)["series:tmdb:1:2:6:1080"]["state"] == "deleted"
started = []
monkeypatch.setattr(addon, "_start_mp4_job", lambda *a, **k: started.append(a))
addon._resume_ingests()
assert started == []
def test_a_copy_fetched_under_a_tt_ident_is_still_ours(client):
"""Ownership is by FILE, like the dest lock: a Kodi cinemeta job (tt…)
names the same destination as the web's tmdb: one."""
path = _hold("series", 2, 6, ours=False)
addon._set_job_state("series:tt9:2:6:1080", "done",
args=("series", "tt9", 2, 6, 1080))
real = addon._library_meta
addon._library_meta_cache["series:tt9:2:6"] = real("series", "tmdb:1", 2, 6)
assert _post(client).status_code == 303
assert not os.path.exists(path)
def test_a_copy_this_addon_did_not_fetch_cannot_be_deleted(client):
path = _hold("series", 2, 6, ours=False)
assert _post(client).status_code == 404
assert os.path.exists(path)
def test_a_copy_being_written_cannot_be_deleted(client):
path = _hold("series", 2, 6)
lock = addon._dest_lock(path)
lock.acquire()
try:
assert _post(client).status_code == 409
assert os.path.exists(path)
finally:
lock.release()
def test_deleting_what_is_not_there_is_404(client):
assert _post(client).status_code == 404
def test_a_movie_delete_returns_to_the_movie(client):
path = _hold("movie", tmdb="2")
r = _post(client, kind="movie", tmdb="2", season="0", episode="0")
assert r.headers["Location"].endswith("/web/title/movie/2")
assert not os.path.exists(path)
def test_delete_notifies_jellyfin(client, isolated):
import threading
path = _hold("series", 2, 6)
_post(client)
for t in threading.enumerate():
if t.name == "jellyfin-notify":
t.join(5)
assert isolated == [(path, "Deleted")]
def test_delete_does_not_wait_for_jellyfin(client, monkeypatch):
"""Measured live: Jellyfin took ~5 s to answer and the click waited."""
import threading
import time
gate = threading.Event()
monkeypatch.setattr(addon, "_notify_jellyfin", lambda p, t: gate.wait(5))
_hold("series", 2, 6)
t0 = time.time()
assert _post(client).status_code == 303
assert time.time() - t0 < 1
gate.set()
@pytest.mark.parametrize("data", [
{"kind": "series", "tmdb": "../1", "season": "2", "episode": "6"},
{"kind": "anime", "tmdb": "1", "season": "2", "episode": "6"},
{"kind": "series", "tmdb": "1", "season": "x", "episode": "6"},
])
def test_bad_input_is_400(client, data):
path = _hold("series", 2, 6)
assert client.post("/web/delete", data=data).status_code == 400
assert os.path.exists(path)
def test_a_get_can_never_delete(client):
path = _hold("series", 2, 6)
assert client.get("/web/delete?kind=series&tmdb=1&season=2&episode=6").status_code == 405
assert os.path.exists(path)
# --- where the button appears ------------------------------------------------
def test_watch_page_has_a_confirmed_delete(client):
_hold("series", 2, 6)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "action='/web/delete'" in body
assert "confirm(" in body
def test_title_page_offers_delete_for_our_copy_only(client):
_hold("series", 2, 6)
_hold("series", 2, 7, ours=False)
body = client.get("/web/title/series/1?s=2").get_data(as_text=True)
assert body.count("action='/web/delete'") == 1
assert "name='episode' value='6'" in body
def test_watch_page_has_no_delete_for_a_copy_not_ours(client):
_hold("series", 2, 6, ours=False)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "action='/web/delete'" not in body
def test_after_delete_the_title_offers_download_again(client):
_hold("series", 2, 6)
_post(client)
body = client.get("/web/title/series/1?s=2").get_data(as_text=True)
assert "action='/web/download'" in body