Tom FluxandClaude Opus 5
155f05773d
Phases 1-4 of plan.md §13. Forked from youtube-automate as planned rather than
written from scratch: naming, NFO, auth, the admin UI, settings and the DB layer
came across largely unchanged, download.py is gone, and the pieces that only make
sense for a streaming library are new.
api.py YouTube Data API v3 client. The whole metadata path.
strm.py Materialising: a .strm, an .nfo and a thumbnail. Replaces the
330-line download.py, because the job is writing a URL to a file.
subsync.py The subscription mirror, most of which is refusals.
reap.py Retention, rewritten around the 30-day window and min_keep_videos.
discovery.py RSS polling plus an API-backed, resumable, bounded backfill.
proxy/ The verified PoC, moved in with a systemd unit.
330 tests, all passing, no network and no yt-dlp in any of them. The suite leans
towards the failure paths, because that is where this design can actually hurt
someone: a 403 that looks like an unsubscribe, a video that ages out and comes
back, a title that never arrives. tests/test_proxy.py replaces the two standalone
scripts under proxy/ and now drives the real make_handler(mgr, ...) rather than the
PoC's make_handler(path, done), so routing and video-id validation are covered too.
Ran it end to end against the live API and it found three real bugs.
The first was mine and the tests caught it: strm.remove pruned empty directories
up to the media root, so a channel directory whose tvshow.nfo happened to be
missing would be deleted along with the season. It only looked safe because
tvshow.nfo normally stops the walk. The prune boundary is now the channel
directory explicitly.
The other two only showed up against real data, and they compounded. The backfill
inserted rows with no title and left the RSS poll to fill them in — but RSS returns
15 entries, which for Pitch Side spans 23 days against a 30-day window, so five of
twenty episodes were named after their video ids. Worse, strm.materialise wrote
that fallback back to the database as the title, which made the row look titled and
permanently disabled the repair path. Both fixed: playlistItems.list now requests
snippet as well as contentDetails, which costs the same single quota unit and
carries the title alongside the exact publish date, and the fallback is used for the
filename without being persisted. A title that does arrive late now also removes the
badly-named files and re-queues, so the episode is rewritten rather than keeping its
video-id name forever. Verified against the live API: all twenty Pitch Side episodes
now carry real titles.
Measured on the real account: 119 subscriptions queued for approval and none added
on the first sync, then a two-channel run backfilled and materialised 26 episodes in
under seven seconds.
Two deliberate departures from plan.md, both recorded there:
min_keep_videos defaults to 5 rather than being left as an open question. Without
it 52 of 117 measured channels are empty Jellyfin series that flicker in and out as
their single video crosses the retention line, and the plan already recommended it.
The Jellyfin refresh is a bare /Library/Refresh with a comment explaining why it
must stay that way. A normal scan makes zero media probes; FullRefresh does probe,
and at 400 episodes that is 400 cold starts.
Not yet done: no systemd units are installed (needs root — deploy/deploy.sh), the
admin UI has no routes for sources or the approval queue yet, and nothing has been
pointed at the real media root.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
367 lines
13 KiB
Python
367 lines
13 KiB
Python
"""Subscription mirroring, and above all its refusals.
|
|
|
|
The sync is authoritative in both directions and the removal half deletes a
|
|
channel's whole tree, so most of what needs pinning down here is what it does
|
|
with *bad* data. Every one of these failure modes looks identical to "he
|
|
unsubscribed from everything" on the wire:
|
|
|
|
403 subscriptionForbidden he re-ticked the privacy box
|
|
network error susan's link dropped
|
|
200 with zero items could be true, could be a broken response
|
|
|
|
None of them may delete anything. The tests below are the reason that claim can
|
|
be made with a straight face.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from ytstream import api, subsync, videos
|
|
|
|
from conftest import CHANNEL_ID, FakeApi, add_channel, add_video, patch_api
|
|
|
|
BROTHER = "UCPcTWaLV8zwx4WP4QExHj4Q"
|
|
|
|
|
|
@pytest.fixture()
|
|
def source(conn):
|
|
key = subsync.add_source(conn, channel_id=BROTHER, label="C Flux")
|
|
with conn:
|
|
conn.execute("UPDATE source SET imported = 1 WHERE key = ?", (key,))
|
|
return subsync.get_source(conn, key)
|
|
|
|
|
|
@pytest.fixture()
|
|
def fresh_source(conn):
|
|
"""A source that has never imported — the first-sync path."""
|
|
key = subsync.add_source(conn, channel_id=BROTHER, label="C Flux")
|
|
return subsync.get_source(conn, key)
|
|
|
|
|
|
def sub(channel_id, title):
|
|
return {"channel_id": channel_id, "title": title}
|
|
|
|
|
|
def _fake(monkeypatch, **kwargs):
|
|
return patch_api(monkeypatch, subsync, FakeApi(**kwargs))
|
|
|
|
|
|
# ------------------------------------------------------------------- additions
|
|
|
|
|
|
def test_first_sync_queues_everything_and_adds_nothing(
|
|
conn, settings, fresh_source, monkeypatch
|
|
):
|
|
"""119 subscriptions would trip any cap, so day one is approval-only."""
|
|
_fake(monkeypatch, subs=[sub(f"UC{i:022d}", f"Chan {i}") for i in range(40)])
|
|
|
|
stats = subsync.sync_source(conn, settings, fresh_source)
|
|
|
|
assert stats["added"] == 0
|
|
assert stats["queued"] == 40
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0
|
|
assert len(subsync.pending(conn)) == 40
|
|
# And it does not queue them again on the next pass.
|
|
assert subsync.get_source(conn, fresh_source["key"])["imported"] == 1
|
|
|
|
|
|
def test_second_sync_adds_within_the_cap(conn, settings, source, monkeypatch):
|
|
settings.set("subsync_max_new", "25")
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")],
|
|
channel={"channel_id": "UCaaaaaaaaaaaaaaaaaaaaaa", "title": "Alpha",
|
|
"description": "", "handle": None, "avatar_url": None})
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["added"] == 1
|
|
row = conn.execute("SELECT * FROM channel").fetchone()
|
|
assert row["title"] == "Alpha"
|
|
assert row["source"] == "youtube"
|
|
|
|
|
|
def test_burst_over_the_cap_adds_nothing_and_queues_all(
|
|
conn, settings, source, monkeypatch
|
|
):
|
|
settings.set("subsync_max_new", "3")
|
|
_fake(monkeypatch, subs=[sub(f"UC{i:022d}", f"Chan {i}") for i in range(10)])
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["added"] == 0
|
|
assert stats["queued"] == 10
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0
|
|
|
|
|
|
def test_rejected_channels_are_never_queued_again(
|
|
conn, settings, fresh_source, monkeypatch
|
|
):
|
|
entries = [sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]
|
|
_fake(monkeypatch, subs=entries)
|
|
subsync.sync_source(conn, settings, fresh_source)
|
|
|
|
queued = subsync.pending(conn)
|
|
assert len(queued) == 1
|
|
subsync.resolve(conn, [queued[0]["id"]], "rejected")
|
|
|
|
source = subsync.get_source(conn, fresh_source["key"])
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["queued"] == 0
|
|
assert stats["added"] == 0
|
|
assert subsync.pending(conn) == []
|
|
|
|
|
|
def test_approving_subscribes(conn, settings, fresh_source, monkeypatch):
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")],
|
|
channel={"channel_id": "UCaaaaaaaaaaaaaaaaaaaaaa", "title": "Alpha",
|
|
"description": "Desc", "handle": "@alpha", "avatar_url": None})
|
|
subsync.sync_source(conn, settings, fresh_source)
|
|
queued = subsync.pending(conn)
|
|
|
|
stats = subsync.approve(conn, settings, [row["id"] for row in queued])
|
|
|
|
assert stats == {"added": 1, "failed": 0}
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
assert subsync.pending(conn) == []
|
|
|
|
|
|
# -------------------------------------------------------------------- refusals
|
|
|
|
|
|
def test_private_subscriptions_change_nothing(conn, settings, source, monkeypatch):
|
|
add_channel(conn, CHANNEL_ID, "Existing", "Existing")
|
|
_fake(monkeypatch, raises=api.SubscriptionsPrivate(
|
|
403, "subscriptionForbidden", "not allowed"))
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["refused"]
|
|
assert "private" in stats["refused"]
|
|
assert stats["removed"] == 0
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
# The channel's miss counter must not move either, or three consecutive
|
|
# outages would delete the library without a single healthy response.
|
|
assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 0
|
|
|
|
|
|
def test_empty_response_is_treated_as_suspect(conn, settings, source, monkeypatch):
|
|
"""A genuinely empty list and a broken one are indistinguishable, so assume
|
|
the harmless reading."""
|
|
add_channel(conn, CHANNEL_ID, "Existing", "Existing")
|
|
_fake(monkeypatch, subs=[])
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["refused"]
|
|
assert stats["removed"] == 0
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
|
|
|
|
def test_api_not_configured_changes_nothing(conn, settings, source, monkeypatch):
|
|
add_channel(conn, CHANNEL_ID, "Existing", "Existing")
|
|
_fake(monkeypatch, raises=api.NotConfigured(403, "forbidden", "blocked"))
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["refused"]
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
|
|
|
|
def test_network_error_changes_nothing(conn, settings, source, monkeypatch):
|
|
add_channel(conn, CHANNEL_ID, "Existing", "Existing")
|
|
_fake(monkeypatch, raises=api.ApiError(0, "network", "connection reset"))
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["refused"]
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
|
|
|
|
def test_refusal_is_recorded_on_the_source(conn, settings, source, monkeypatch):
|
|
_fake(monkeypatch, raises=api.ApiError(0, "network", "boom"))
|
|
subsync.sync_source(conn, settings, source)
|
|
|
|
row = subsync.get_source(conn, source["key"])
|
|
assert row["last_sync_ok"] == 0
|
|
assert row["consecutive_failures"] == 1
|
|
assert "boom" in row["last_error"]
|
|
|
|
|
|
def test_three_outages_in_a_row_still_delete_nothing(
|
|
conn, settings, source, monkeypatch
|
|
):
|
|
"""The threshold counts absences from healthy responses, not failures."""
|
|
add_channel(conn, CHANNEL_ID, "Existing", "Existing")
|
|
_fake(monkeypatch, raises=api.ApiError(0, "network", "down"))
|
|
|
|
for _ in range(5):
|
|
subsync.sync_source(conn, settings, subsync.get_source(conn, source["key"]))
|
|
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
|
|
|
|
# -------------------------------------------------------------------- removals
|
|
|
|
|
|
def test_absence_counts_up_but_does_not_delete_below_the_threshold(
|
|
conn, settings, source, monkeypatch
|
|
):
|
|
settings.set("subsync_missing_threshold", "3")
|
|
add_channel(conn, CHANNEL_ID, "Doomed", "Doomed")
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
settings.set("subsync_max_new", "0") # keep the addition path out of this
|
|
|
|
for expected in (1, 2):
|
|
stats = subsync.sync_source(
|
|
conn, settings, subsync.get_source(conn, source["key"])
|
|
)
|
|
assert stats["removed"] == 0
|
|
assert stats["pending_removal"] == 1
|
|
assert conn.execute(
|
|
"SELECT missing_syncs FROM channel WHERE dir_name = 'Doomed'"
|
|
).fetchone()[0] == expected
|
|
|
|
|
|
def test_deletion_happens_on_the_threshold_sync(
|
|
conn, settings, source, media_root, monkeypatch
|
|
):
|
|
settings.set("subsync_missing_threshold", "3")
|
|
settings.set("subsync_max_new", "0")
|
|
doomed = add_channel(conn, CHANNEL_ID, "Doomed", "Doomed")
|
|
tree = media_root / "Doomed"
|
|
(tree / "Season 2026").mkdir(parents=True)
|
|
(tree / "tvshow.nfo").write_text("<tvshow/>")
|
|
add_video(conn, doomed["id"], "vid00000001")
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
|
|
for _ in range(3):
|
|
stats = subsync.sync_source(
|
|
conn, settings, subsync.get_source(conn, source["key"])
|
|
)
|
|
|
|
assert stats["removed"] == 1
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0
|
|
assert not tree.exists()
|
|
# The videos went with it, via ON DELETE CASCADE.
|
|
assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 0
|
|
|
|
|
|
def test_reappearing_resets_the_counter(conn, settings, source, monkeypatch):
|
|
settings.set("subsync_missing_threshold", "3")
|
|
settings.set("subsync_max_new", "0")
|
|
add_channel(conn, CHANNEL_ID, "Flaky", "Flaky")
|
|
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
subsync.sync_source(conn, settings, subsync.get_source(conn, source["key"]))
|
|
assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 1
|
|
|
|
_fake(monkeypatch, subs=[sub(CHANNEL_ID, "Flaky")])
|
|
subsync.sync_source(conn, settings, subsync.get_source(conn, source["key"]))
|
|
|
|
assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 0
|
|
|
|
|
|
def test_manual_channels_are_never_removed(
|
|
conn, settings, source, media_root, monkeypatch
|
|
):
|
|
settings.set("subsync_missing_threshold", "1")
|
|
settings.set("subsync_max_new", "0")
|
|
add_channel(conn, CHANNEL_ID, "Pinned", "Pinned", source="manual")
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
|
|
for _ in range(3):
|
|
stats = subsync.sync_source(
|
|
conn, settings, subsync.get_source(conn, source["key"])
|
|
)
|
|
|
|
assert stats["removed"] == 0
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 0
|
|
|
|
|
|
def test_threshold_of_one_deletes_on_the_first_absence(
|
|
conn, settings, source, monkeypatch
|
|
):
|
|
"""Configurable so tests need not loop; the default stays at 3."""
|
|
settings.set("subsync_missing_threshold", "1")
|
|
settings.set("subsync_max_new", "0")
|
|
add_channel(conn, CHANNEL_ID, "Doomed", "Doomed")
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["removed"] == 1
|
|
|
|
|
|
def test_zero_threshold_is_clamped_to_one(conn, settings, source, monkeypatch):
|
|
"""A stored 0 must not mean "delete before any absence is confirmed"."""
|
|
settings.set("subsync_missing_threshold", "0")
|
|
settings.set("subsync_max_new", "0")
|
|
add_channel(conn, CHANNEL_ID, "Doomed", "Doomed")
|
|
_fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
# Clamped to 1, so the first absence is enough — but it took one absence,
|
|
# not zero, and the row was actually observed missing.
|
|
assert stats["removed"] == 1
|
|
assert stats["seen"] == 1
|
|
|
|
|
|
# --------------------------------------------------------------------- general
|
|
|
|
|
|
def test_unsafe_channel_dir_is_not_deleted(conn, settings, media_root):
|
|
"""A blank dir_name must never resolve the delete to the media root."""
|
|
row = add_channel(conn, CHANNEL_ID, "Bad", " ")
|
|
(media_root / "keepme").mkdir()
|
|
|
|
from ytstream import strm
|
|
|
|
assert strm.remove_channel_tree(row) is False
|
|
assert (media_root / "keepme").exists()
|
|
|
|
|
|
def test_one_unresolvable_channel_does_not_abort_the_sync(
|
|
conn, settings, source, monkeypatch
|
|
):
|
|
class Exploding(FakeApi):
|
|
def channel(self, channel_id):
|
|
if channel_id.endswith("bad"):
|
|
raise api.ApiError(500, "backendError", "boom")
|
|
return {"channel_id": channel_id, "title": "Fine", "description": "",
|
|
"handle": None, "avatar_url": None}
|
|
|
|
monkeypatch.setattr(
|
|
subsync.channels, "subscribe_from_sync",
|
|
lambda conn, settings, cid, title: (_ for _ in ()).throw(RuntimeError("no"))
|
|
if cid.endswith("bad") else add_channel(conn, cid, title, title),
|
|
)
|
|
patch_api(monkeypatch, subsync,
|
|
FakeApi(subs=[sub("UC" + "a" * 19 + "bad", "Bad"),
|
|
sub("UC" + "b" * 22, "Good")]))
|
|
|
|
stats = subsync.sync_source(conn, settings, source)
|
|
|
|
assert stats["added"] == 1
|
|
assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1
|
|
|
|
|
|
def test_sync_all_aggregates_and_counts_refusals(conn, settings, source, monkeypatch):
|
|
_fake(monkeypatch, raises=api.ApiError(0, "network", "boom"))
|
|
totals = subsync.sync_all(conn, settings)
|
|
assert totals["sources"] == 1
|
|
assert totals["refused"] == 1
|
|
|
|
|
|
def test_disabled_source_is_skipped(conn, settings, source, monkeypatch):
|
|
with conn:
|
|
conn.execute("UPDATE source SET enabled = 0")
|
|
fake = _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")])
|
|
|
|
totals = subsync.sync_all(conn, settings)
|
|
|
|
assert totals["sources"] == 0
|
|
assert fake.subscription_calls == 0
|