"""Discovery: feed parsing, the fallback path, and the two repair mechanisms.""" from datetime import date, timedelta import pytest from conftest import add_video, feed_bytes from youtube_automate import discovery, util, videos class TestParseEntries: def test_parses_all_entries(self): entries = discovery.parse_entries(feed_bytes("uulf_feed.xml")) assert len(entries) == 3 assert entries[0]["video_id"] == "08Ajr5fP52I" assert entries[0]["published"] == date(2026, 8, 7) def test_unescapes_description_entities(self): entries = discovery.parse_entries(feed_bytes("uulf_feed.xml")) assert "&" in entries[0]["description"] assert "" in entries[0]["description"] def test_keeps_characters_the_filename_would_strip(self): entries = discovery.parse_entries(feed_bytes("uulf_feed.xml")) assert entries[1]["title"] == "Trying to use a Nortel PBX: part two" def test_empty_feed_yields_nothing(self): empty = b'' assert discovery.parse_entries(empty) == [] def test_unparseable_feed_raises(self): with pytest.raises(discovery.FeedUnavailable): discovery.parse_entries(b"' b"no id" ) assert discovery.parse_entries(payload) == [] class TestFeedUrls: def test_uulf_strips_the_uc_prefix(self): url = discovery.uulf_feed_url("UCW7jUEpYT_t0Gsf632d6_wQ") assert "playlist_id=UULFW7jUEpYT_t0Gsf632d6_wQ" in url def test_uc_feed_uses_channel_id(self): url = discovery.uc_feed_url("UCW7jUEpYT_t0Gsf632d6_wQ") assert "channel_id=UCW7jUEpYT_t0Gsf632d6_wQ" in url class TestPollChannel: def test_queues_recent_and_skips_old(self, conn, settings, channel, monkeypatch): recent = util.today() - timedelta(days=2) stale = util.today() - timedelta(days=400) monkeypatch.setattr( discovery, "fetch_feed", lambda url, timeout=30.0: b"ignored", ) monkeypatch.setattr( discovery, "parse_entries", lambda payload: [ {"video_id": "new1", "title": "new", "published": recent, "description": ""}, {"video_id": "old1", "title": "old", "published": stale, "description": ""}, ], ) stats = discovery.poll_channel(conn, settings, channel) assert stats["queued"] == 1 assert stats["old"] == 1 assert videos.get(conn, "new1")["state"] == videos.PENDING assert videos.get(conn, "old1")["state"] == videos.SKIPPED_OLD def test_falls_back_to_channel_feed_when_uulf_is_empty( self, conn, settings, channel, monkeypatch ): seen_urls = [] def fake_fetch(url, timeout=30.0): seen_urls.append(url) return None if "playlist_id" in url else b"feed" monkeypatch.setattr(discovery, "fetch_feed", fake_fetch) monkeypatch.setattr( discovery, "parse_entries", lambda payload: [ { "video_id": "fb1", "title": "fallback", "published": util.today(), "description": "", } ], ) stats = discovery.poll_channel(conn, settings, channel) assert stats["source"] == videos.SOURCE_UC assert any("playlist_id" in url for url in seen_urls) assert any("channel_id" in url for url in seen_urls) assert videos.get(conn, "fb1")["discovery_source"] == videos.SOURCE_UC def test_feed_failure_increments_the_counter(self, conn, settings, channel, monkeypatch): def boom(url, timeout=30.0): raise discovery.FeedUnavailable("HTTP 503") monkeypatch.setattr(discovery, "fetch_feed", boom) stats = discovery.poll_channel(conn, settings, channel) assert "error" in stats row = conn.execute( "SELECT * FROM channel WHERE id = ?", (channel["id"],) ).fetchone() assert row["consecutive_poll_failures"] == 1 assert row["last_poll_ok"] == 0 def test_success_resets_the_failure_counter(self, conn, settings, channel, monkeypatch): with conn: conn.execute( "UPDATE channel SET consecutive_poll_failures = 4 WHERE id = ?", (channel["id"],), ) monkeypatch.setattr(discovery, "fetch_feed", lambda url, timeout=30.0: b"x") monkeypatch.setattr(discovery, "parse_entries", lambda payload: []) discovery.poll_channel(conn, settings, channel) row = conn.execute( "SELECT * FROM channel WHERE id = ?", (channel["id"],) ).fetchone() assert row["consecutive_poll_failures"] == 0 assert row["last_poll_ok"] == 1 class TestSkippedShortRepair: """A fallback-discovered video wrongly rejected as a Short must come back once the authoritative UULF feed lists it.""" def _poll_with(self, monkeypatch, entry): monkeypatch.setattr(discovery, "fetch_feed", lambda url, timeout=30.0: b"x") monkeypatch.setattr(discovery, "parse_entries", lambda payload: [entry]) def test_repairs_a_uc_discovered_skipped_short( self, conn, settings, channel, monkeypatch ): add_video( conn, channel["id"], "short1", state=videos.SKIPPED_SHORT, discovery_source=videos.SOURCE_UC, upload_date=util.today().isoformat(), ) self._poll_with( monkeypatch, { "video_id": "short1", "title": "t", "published": util.today(), "description": "", }, ) stats = discovery.poll_channel(conn, settings, channel) assert stats["repaired"] == 1 row = videos.get(conn, "short1") assert row["state"] == videos.PENDING assert row["discovery_source"] == videos.SOURCE_UULF def test_does_not_repair_one_discovered_via_uulf( self, conn, settings, channel, monkeypatch ): add_video( conn, channel["id"], "short2", state=videos.SKIPPED_SHORT, discovery_source=videos.SOURCE_UULF, ) self._poll_with( monkeypatch, {"video_id": "short2", "title": "t", "published": util.today(), "description": ""}, ) discovery.poll_channel(conn, settings, channel) assert videos.get(conn, "short2")["state"] == videos.SKIPPED_SHORT def test_never_resurrects_a_deleted_tombstone( self, conn, settings, channel, monkeypatch ): add_video( conn, channel["id"], "gone1", state=videos.DELETED, discovery_source=videos.SOURCE_UC, ) self._poll_with( monkeypatch, {"video_id": "gone1", "title": "t", "published": util.today(), "description": ""}, ) discovery.poll_channel(conn, settings, channel) assert videos.get(conn, "gone1")["state"] == videos.DELETED class TestRescan: def test_requeues_skipped_old_inside_the_window(self, conn, settings, channel): inside = (util.today() - timedelta(days=5)).isoformat() add_video(conn, channel["id"], "v1", state=videos.SKIPPED_OLD, upload_date=inside) assert discovery.rescan_channel(conn, settings, channel) == 1 assert videos.get(conn, "v1")["state"] == videos.PENDING def test_leaves_videos_outside_the_window_alone(self, conn, settings, channel): outside = (util.today() - timedelta(days=200)).isoformat() add_video(conn, channel["id"], "v2", state=videos.SKIPPED_OLD, upload_date=outside) assert discovery.rescan_channel(conn, settings, channel) == 0 assert videos.get(conn, "v2")["state"] == videos.SKIPPED_OLD def test_honours_a_per_channel_override(self, conn, settings, channel): age = (util.today() - timedelta(days=30)).isoformat() add_video(conn, channel["id"], "v3", state=videos.SKIPPED_OLD, upload_date=age) # Default retention is 14 days, so nothing moves. assert discovery.rescan_channel(conn, settings, channel) == 0 with conn: conn.execute( "UPDATE channel SET retention_days = 60 WHERE id = ?", (channel["id"],) ) widened = conn.execute( "SELECT * FROM channel WHERE id = ?", (channel["id"],) ).fetchone() assert discovery.rescan_channel(conn, settings, widened) == 1 def test_never_resurrects_a_tombstone(self, conn, settings, channel): add_video( conn, channel["id"], "v4", state=videos.DELETED, upload_date=util.today().isoformat(), ) assert discovery.rescan_channel(conn, settings, channel) == 0 assert videos.get(conn, "v4")["state"] == videos.DELETED class TestEffectiveRetention: def test_override_wins(self, settings, channel, conn): with conn: conn.execute( "UPDATE channel SET retention_days = 60 WHERE id = ?", (channel["id"],) ) row = conn.execute("SELECT * FROM channel WHERE id = ?", (channel["id"],)).fetchone() assert discovery.effective_retention_days(settings, row) == 60 def test_falls_back_to_the_global_default(self, settings, channel): assert discovery.effective_retention_days(settings, channel) == 14