"""Guards that stop one bad YouTube answer from cascading: the subscription prune guard, the
playlist-push circuit breaker, and RSS telling a failure apart from an empty feed."""
import logging
import httpx
import pytest
from app.sync import runner
from app.sync.playlists import FAILED, _Breaker
from app.sync.subscriptions import should_prune
from app.youtube import rss
from app.youtube.client import YouTubeError
class TestShouldPrune:
def test_a_normal_fetch_prunes(self):
assert should_prune(fetched_count=12, existing_count=15) is True
def test_an_empty_fetch_against_existing_rows_never_prunes(self):
# The data-loss case: one anomalous empty-but-200 response must not wipe 15 rows.
assert should_prune(fetched_count=0, existing_count=15) is False
def test_an_empty_fetch_with_nothing_local_is_fine(self):
# A brand-new account: nothing to lose, so the (no-op) prune may run.
assert should_prune(fetched_count=0, existing_count=0) is True
def test_a_single_remaining_subscription_still_prunes(self):
assert should_prune(fetched_count=1, existing_count=200) is True
class TestBreaker:
def test_open_until_the_limit_is_reached(self):
b = _Breaker(limit=3)
for _ in range(2):
b.hit(False)
assert b.open()
b.hit(False)
assert not b.open() and b.tripped
def test_a_success_re_arms_the_streak(self):
b = _Breaker(limit=3)
b.hit(False)
b.hit(False)
b.hit(True) # one write went through — this isn't an account-wide failure
b.hit(False)
b.hit(False)
assert b.open()
def test_stays_tripped_once_tripped(self):
b = _Breaker(limit=2)
b.hit(False)
b.hit(False)
b.hit(True)
assert not b.open()
def _quota_dead():
"""A write op that fails the way a quota-exhausted / scope-revoked account fails."""
raise YouTubeError("quota exceeded")
class TestBreakerRun:
def test_returns_the_ops_result_and_leaves_failures_alone(self):
b, failed = _Breaker(), []
assert b.run(lambda: "item-1", lambda: failed.append("v1")) == "item-1"
assert failed == []
def test_a_youtube_error_records_the_failure_and_counts_towards_the_trip(self):
b, failed = _Breaker(limit=2), []
assert b.run(_quota_dead, lambda: failed.append("v1")) is FAILED
assert failed == ["v1"] and b.open()
b.run(_quota_dead, lambda: failed.append("v2"))
assert not b.open()
def test_a_tripped_breaker_never_calls_the_op(self):
b, failed, calls = _Breaker(limit=1), [], []
b.run(_quota_dead, lambda: failed.append("v1"))
assert b.run(lambda: calls.append("called"), lambda: failed.append("v2")) is FAILED
assert calls == [] # the whole point: no more quota spent
assert failed == ["v1", "v2"] # …but the item is still reported
def test_other_exceptions_are_not_swallowed(self):
# Only YouTubeError is a "write failed" signal; a bug must still surface.
def boom():
raise ValueError("bug")
with pytest.raises(ValueError):
_Breaker().run(boom, lambda: None)
class _StubDb:
"""Just enough Session for run_rss_poll when the caller supplies the channel list."""
def __init__(self):
self.rollbacks = 0
def rollback(self):
self.rollbacks += 1
class _StubChannel:
def __init__(self, cid):
self.id = cid
class TestRssPollResult:
"""`failed` must count every channel that did not get polled — feed-side AND apply-side.
Counting only the fetch failures would move the "a broken run looks like a quiet hour" blind
spot rather than remove it."""
def _run(self, monkeypatch, *, fetch, apply_, channels=("UC1", "UC2")):
monkeypatch.setattr(runner, "fetch_channel_feed", fetch)
monkeypatch.setattr(runner, "apply_rss_feed", apply_)
# The API fallback has its own DB-lane tests (test_feed_fallback.py); here it reads nothing.
monkeypatch.setattr(
runner, "run_feed_fallback", lambda db, ids: {"read": 0, "new": 0, "failed": 0}
)
return runner.run_rss_poll(_StubDb(), [_StubChannel(c) for c in channels])
def test_a_healthy_run_reports_no_failures(self, monkeypatch):
out = self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=lambda db, ch, e: 1)
assert out == {"new": 2, "failed": 0, "fallback": 0}
def test_unreadable_feeds_are_counted(self, monkeypatch):
# One dead channel among healthy ones: counted, but the run itself is still a run.
def fetch(cid):
if cid == "UC_dead":
raise rss.RssError("404", "http_404")
return [{"id": "v"}]
out = self._run(
monkeypatch, fetch=fetch, apply_=lambda db, ch, e: 1, channels=("UC1", "UC_dead")
)
assert out == {"new": 1, "failed": 1, "fallback": 0}
def test_apply_failures_are_counted_too(self, monkeypatch):
# A per-row write failure (a constraint clash on one channel) while the rest store fine.
def apply_(db, ch, entries):
if ch.id == "UC_bad":
raise RuntimeError("constraint")
return 1
out = self._run(
monkeypatch,
fetch=lambda cid: [{"id": "v"}],
apply_=apply_,
channels=("UC1", "UC_bad"),
)
assert out == {"new": 1, "failed": 1, "fallback": 0}, (
"an unstored channel must not read as ok"
)
def test_many_failures_write_one_warning_with_counts_not_one_per_channel(
self, monkeypatch, caplog
):
# YouTube's RSS outage of 2026-09-16: ~900 of 979 feeds failed on every 20-minute poll and
# each wrote its own WARNING — 11k lines in four hours, which the log_events table would
# have stored too. One summary per poll, per-channel detail at DEBUG.
channels = [f"UC{i}" for i in range(40)]
def fetch(cid):
n = int(cid[2:])
if n < 30:
raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404", "http_404")
if n < 35:
raise rss.RssError(f"RSS fetch failed for {cid}: timed out", "transport")
return [{"id": "v"}]
caplog.set_level(logging.DEBUG, logger="siftlode.sync")
out = self._run(monkeypatch, fetch=fetch, apply_=lambda db, ch, e: 1, channels=channels)
assert out == {"new": 5, "failed": 35, "fallback": 0}
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
assert len(warnings) == 1
assert warnings[0].event == "sync.rss.failed"
assert warnings[0].attrs["reasons"] == {"http_404": 30, "transport": 5}
assert len(warnings[0].attrs["examples"]) == runner.RSS_FAILURE_EXAMPLES
assert sum(r.levelno == logging.DEBUG for r in caplog.records) == 35
def test_a_failure_on_our_side_gets_one_traceback_per_poll_not_one_per_channel(
self, monkeypatch, caplog
):
# A read-only database fails every channel's apply identically. The first gets a traceback
# at ERROR; repeating it for each of 979 channels would be the RSS flood all over again.
def apply_(db, ch, entries):
raise RuntimeError("read-only transaction")
caplog.set_level(logging.DEBUG, logger="siftlode.sync")
with pytest.raises(RuntimeError):
self._run(
monkeypatch,
fetch=lambda cid: [{"id": "v"}],
apply_=apply_,
channels=("A", "B", "C"),
)
errors = [r for r in caplog.records if r.levelno == logging.ERROR]
assert len(errors) == 1 and errors[0].exc_info is not None
assert (
sum(r.levelno == logging.DEBUG and r.exc_info is not None for r in caplog.records) == 2
)
def test_a_quiet_hour_is_distinguishable_from_an_outage(self, monkeypatch):
quiet = self._run(monkeypatch, fetch=lambda cid: [], apply_=lambda db, ch, e: 0)
assert quiet == {
"new": 0,
"failed": 0,
"fallback": 0,
} # same "new", different "failed" — that's the point
def test_a_total_outage_raises_instead_of_returning_ok(self, monkeypatch):
# Every feed unreachable (the tinyproxy-lost-the-boot-race incident). Returning a result
# here recorded the scheduler run as "ok", with the outage only visible inside the summary
# text — a green pill for a run that did nothing at all.
def boom(cid):
raise rss.RssError("Connection refused", "transport")
with pytest.raises(RuntimeError, match="all 2 channels"):
self._run(monkeypatch, fetch=boom, apply_=lambda db, ch, e: 1)
def test_the_outage_message_carries_the_underlying_error(self, monkeypatch):
# "all N channels failed" is ALSO what one dead feed id looks like on a small instance.
# The card has to say which: `HTTP 404` = fix the subscription, `Connection refused` =
# the egress is down.
def dead(cid):
raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404", "http_404")
with pytest.raises(RuntimeError, match="HTTP 404"):
self._run(monkeypatch, fetch=dead, apply_=lambda db, ch, e: 1, channels=("UC_dead",))
def test_a_read_only_database_is_a_total_outage_too(self, monkeypatch):
# Every feed reads fine, nothing can be stored: just as un-polled, just as red.
def boom(db, ch, entries):
raise RuntimeError("read-only transaction")
with pytest.raises(RuntimeError, match="read-only transaction"):
self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=boom)
def test_only_the_unreadable_feeds_go_to_the_api_and_what_it_reads_counts(
self, monkeypatch, caplog
):
# A feed that answered 404 is the fallback's; a channel whose rows failed to STORE is not —
# the API would hand the same rows to the same broken write.
seen = []
def fetch(cid):
if cid == "UC_dark":
raise rss.RssError("404", "http_404")
return [{"id": "v"}]
def apply_(db, ch, entries):
if ch.id == "UC_badrow":
raise RuntimeError("constraint")
return 1
def fallback(db, ids):
seen.extend(ids)
return {"read": 1, "new": 3, "failed": 0}
monkeypatch.setattr(runner, "fetch_channel_feed", fetch)
monkeypatch.setattr(runner, "apply_rss_feed", apply_)
monkeypatch.setattr(runner, "run_feed_fallback", fallback)
out = runner.run_rss_poll(
_StubDb(), [_StubChannel(c) for c in ("UC_ok", "UC_dark", "UC_badrow")]
)
assert seen == ["UC_dark"]
assert out == {"new": 4, "failed": 2, "fallback": 1}
# ...and the one line the stored log keeps says so (an INFO line outside a request is not).
(warning,) = [r for r in caplog.records if getattr(r, "event", None) == "sync.rss.failed"]
assert warning.attrs["fallback_read"] == 1 and warning.attrs["fallback_new"] == 3
def test_no_channels_at_all_is_not_an_outage(self, monkeypatch):
# A fresh instance with nothing subscribed must not report an error every poll.
out = self._run(monkeypatch, fetch=lambda cid: [], apply_=lambda db, ch, e: 0, channels=())
assert out == {"new": 0, "failed": 0, "fallback": 0}
class TestRssErrorVsEmpty:
"""A failed poll must RAISE (so the caller leaves `last_rss_at` alone); only a genuinely
empty feed may return []."""
def _patch(self, monkeypatch, **kwargs):
def fake_get(url, **_):
if "exc" in kwargs:
raise kwargs["exc"]
return httpx.Response(
kwargs["status"],
content=kwargs.get("body", b""),
request=httpx.Request("GET", url),
)
monkeypatch.setattr(rss.httpx, "get", fake_get)
monkeypatch.setattr(rss.time, "sleep", lambda s: None)
def _answers(self, monkeypatch, *answers):
"""The endpoint answers each try with the next of `answers` (a status or an exception);
returns the list of tries made and the waits between them."""
tries, waits = [], []
queue = list(answers)
def fake_get(url, **_):
answer = queue.pop(0)
tries.append(answer)
if isinstance(answer, Exception):
raise answer
body = self.FEED if answer == 200 else b""
return httpx.Response(answer, content=body, request=httpx.Request("GET", url))
monkeypatch.setattr(rss.httpx, "get", fake_get)
monkeypatch.setattr(rss.time, "sleep", waits.append)
return tries, waits
FEED = b"""
abc123Hello
"""
def test_a_feed_that_fails_and_then_answers_is_read(self, monkeypatch):
# The endpoint's measured behaviour: the same channel 404s, 500s, then answers.
tries, waits = self._answers(monkeypatch, 404, 500, 200)
assert [e["id"] for e in rss.fetch_channel_feed("UC_flaky")] == ["abc123"]
assert (len(tries), waits) == (3, list(rss.RETRY_DELAYS))
def test_a_dropped_connection_is_tried_again_too(self, monkeypatch):
tries, _ = self._answers(monkeypatch, httpx.ConnectError("reset"), 200)
assert rss.fetch_channel_feed("UC_blip")
assert len(tries) == 2
def test_a_timeout_is_not_tried_again(self, monkeypatch):
# A hung request already cost the full timeout; retrying it would triple a poll over a
# dead egress.
tries, waits = self._answers(monkeypatch, httpx.ReadTimeout("hung"), 200)
with pytest.raises(rss.RssError) as err:
rss.fetch_channel_feed("UC_hung")
assert (len(tries), waits, err.value.reason) == (1, [], "transport")
def test_a_feed_that_keeps_failing_raises_its_last_failure(self, monkeypatch):
tries, _ = self._answers(monkeypatch, 500, 404, 404)
with pytest.raises(rss.RssError) as err:
rss.fetch_channel_feed("UC_gone")
assert (len(tries), err.value.reason) == (3, "http_404")
def test_a_refusal_is_not_tried_again(self, monkeypatch):
tries, waits = self._answers(monkeypatch, 403)
with pytest.raises(rss.RssError) as err:
rss.fetch_channel_feed("UC_forbidden")
assert (len(tries), waits, err.value.reason) == (1, [], "http_403")
def test_non_200_raises(self, monkeypatch):
self._patch(monkeypatch, status=404)
with pytest.raises(rss.RssError):
rss.fetch_channel_feed("UC_dead")
def test_transport_error_raises(self, monkeypatch):
self._patch(monkeypatch, exc=httpx.ConnectError("no route"))
with pytest.raises(rss.RssError):
rss.fetch_channel_feed("UC_offline")
def test_an_empty_but_valid_feed_returns_empty(self, monkeypatch):
feed = b""""""
self._patch(monkeypatch, status=200, body=feed)
assert rss.fetch_channel_feed("UC_quiet") == []
def test_entries_are_parsed(self, monkeypatch):
feed = b"""
abc123Hello
2026-07-23T10:00:00+00:00
"""
self._patch(monkeypatch, status=200, body=feed)
out = rss.fetch_channel_feed("UC_live")
assert [e["id"] for e in out] == ["abc123"]
assert out[0]["channel_id"] == "UC_live"