peter
2a7f49c981
- HLS sweep only deletes directories matching the session-name shape; the root is admin-configurable and may be a shared path - playlist unwrap picks the first entry that actually downloaded, not entries[0] - concat quoting keeps backslashes literal (ffmpeg treats them so inside '') - ArrowUp/Down defer while the focused element can still scroll that way - the staging fallback skips yt-dlp working files (.part-Frag/.ytdl/.temp/.fNNN) - the boot HLS sweep runs off the event loop - volume persistence is debounced (was a setItem per wheel notch) and a spin no longer loses notches to the iframe's lagging getVolume - _Breaker.run owns the whole open/call/record protocol; four call sites shrink - should_prune extracted and unit-tested, plus RSS error-vs-empty tests - the worker exits non-zero when the schema never lands
121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
"""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 httpx
|
|
import pytest
|
|
|
|
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()
|
|
|
|
|
|
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(lambda: (_ for _ in ()).throw(YouTubeError("quota")), lambda: failed.append("v1")) is FAILED
|
|
assert failed == ["v1"] and b.open()
|
|
b.run(lambda: (_ for _ in ()).throw(YouTubeError("quota")), 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(lambda: (_ for _ in ()).throw(YouTubeError("quota")), 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.
|
|
with pytest.raises(ValueError):
|
|
_Breaker().run(lambda: (_ for _ in ()).throw(ValueError("bug")), lambda: None)
|
|
|
|
|
|
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)
|
|
|
|
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"""<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom"></feed>"""
|
|
self._patch(monkeypatch, status=200, body=feed)
|
|
assert rss.fetch_channel_feed("UC_quiet") == []
|
|
|
|
def test_entries_are_parsed(self, monkeypatch):
|
|
feed = b"""<?xml version="1.0"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom" xmlns:yt="http://www.youtube.com/xml/schemas/2015">
|
|
<entry><yt:videoId>abc123</yt:videoId><title>Hello</title>
|
|
<published>2026-07-23T10:00:00+00:00</published></entry>
|
|
</feed>"""
|
|
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"
|