"""Tell Jellyfin when a library file is added or removed (labs#13).
Measured 2026-10-04: Jellyfin's real-time monitor picked up S02E06 by itself
three seconds before the ingest finished logging, so for ADDS this is a
backstop. It exists for the cases the monitor is not proven on -- deletes and
subtitle sidecars -- and for a monitor that misses an event (inotify on a USB
exFAT volume is not something to bet on silently).
It must never fail or delay the thing it reports: no config -> nothing; any
error -> logged and dropped.
Run: .venv/bin/python3 -m pytest test_jellyfin_notify.py -q
"""
import json
import os
import threading
import pytest
import addon
@pytest.fixture
def config(tmp_path, monkeypatch):
path = tmp_path / "jellyfin.json"
path.write_text(json.dumps({"url": "http://jf:8096", "api_key": "k123"}))
monkeypatch.setattr(addon, "JELLYFIN_CONFIG", str(path))
monkeypatch.setattr(addon, "_jellyfin_cfg", None)
return path
@pytest.fixture
def posts(monkeypatch):
sent = []
class Resp:
status_code = 204
def raise_for_status(self):
pass
def post(url, json=None, headers=None, timeout=None):
sent.append({"url": url, "json": json, "headers": headers, "timeout": timeout})
return Resp()
monkeypatch.setattr(addon.requests, "post", post)
return sent
def test_it_posts_a_targeted_media_update(config, posts):
addon._notify_jellyfin("/mnt/usb/Shows/A/a.mp4", "Created")
(p,) = posts
assert p["url"] == "http://jf:8096/Library/Media/Updated"
assert p["headers"]["X-Emby-Token"] == "k123"
assert p["json"] == {"Updates": [{"Path": "/mnt/usb/Shows/A/a.mp4",
"UpdateType": "Created"}]}
assert p["timeout"] and p["timeout"] <= 10
def test_no_config_means_no_call(tmp_path, monkeypatch, posts):
monkeypatch.setattr(addon, "JELLYFIN_CONFIG", str(tmp_path / "absent.json"))
monkeypatch.setattr(addon, "_jellyfin_cfg", None)
addon._notify_jellyfin("/x.mp4", "Created")
assert posts == []
def test_a_jellyfin_failure_never_raises(config, monkeypatch):
def down(*a, **k):
raise addon.requests.ConnectionError("refused")
monkeypatch.setattr(addon.requests, "post", down)
addon._notify_jellyfin("/x.mp4", "Deleted") # must not raise
def test_a_bad_config_never_raises(tmp_path, monkeypatch, posts):
bad = tmp_path / "jellyfin.json"
bad.write_text("{not json")
monkeypatch.setattr(addon, "JELLYFIN_CONFIG", str(bad))
monkeypatch.setattr(addon, "_jellyfin_cfg", None)
addon._notify_jellyfin("/x.mp4", "Created")
assert posts == []
def test_the_suite_never_reaches_the_real_jellyfin():
"""conftest points JELLYFIN_CONFIG at nothing for every test. On stortron
the real jellyfin.json sits beside addon.py; without the guard any test
reaching an ingest would report temp paths to the production server."""
assert not os.path.exists(addon.JELLYFIN_CONFIG)
# --- wiring -----------------------------------------------------------------
META = {"tmdb_id": "1", "title": "Show", "year": "2020",
"episode_title": "Six", "runtime": 30}
def test_a_finished_ingest_reports_created(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "q.json"))
monkeypatch.setattr(addon, "INGEST_HISTORY_FILE", str(tmp_path / "h.jsonl"))
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: dict(META))
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_library_has", lambda *a, **k: False)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://cdn/m")
monkeypatch.setattr(addon, "_cap_variant", lambda u, h=None: u)
monkeypatch.setattr(addon, "_fetch_subtitles_for", lambda *a, **k: None)
def remux(url, dest):
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x")
monkeypatch.setattr(addon, "_remux_to_mp4", remux)
seen = []
monkeypatch.setattr(addon, "_notify_jellyfin", lambda p, t: seen.append((p, t)))
assert addon._run_ingest("series", "tmdb:1", 2, 6, 1080) == "done"
assert len(seen) == 1 and seen[0][1] == "Created"
assert seen[0][0].endswith("Show (2020) - S02E06 - Six.mp4")
def test_a_failed_ingest_reports_nothing(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: dict(META))
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_library_has", lambda *a, **k: False)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_ingest_from_hdhub", lambda *a, **k: "no usable file")
seen = []
monkeypatch.setattr(addon, "_notify_jellyfin", lambda p, t: seen.append((p, t)))
assert addon._run_ingest("series", "tmdb:1", 2, 6, 1080).startswith("failed")
assert seen == []