"""Test fixtures. Every path the application uses is redirected into a tmpdir. No test touches the network, a real yt-dlp, a real Jellyfin, or the real media tree — the YouTube API client is always a stub (see `FakeApi`), because the point of a test suite here is to pin down behaviour that only shows up on the failure paths: a 403, an empty response, a video that ages out and must not come back. """ from __future__ import annotations import os import tempfile from pathlib import Path # config resolves its paths at import time, so the environment has to be set # before anything from the package is imported. _SANDBOX = Path(tempfile.mkdtemp(prefix="yts-tests-")) os.environ.setdefault("YTS_STATE_DIR", str(_SANDBOX / "state")) os.environ.setdefault("YTS_MEDIA_ROOT", str(_SANDBOX / "media")) os.environ.setdefault("YTS_DB_PATH", str(_SANDBOX / "state" / "ytstream.db")) os.environ.setdefault("YTS_LOCK_PATH", str(_SANDBOX / "state" / "run.lock")) os.environ.setdefault("YTS_VENV_BIN", str(_SANDBOX / "venv" / "bin")) import pytest # noqa: E402 from ytstream import api, config, db, util, videos # noqa: E402 from ytstream.settings import Settings # noqa: E402 FIXTURES = Path(__file__).parent / "fixtures" # A real-looking channel id: UC + 22 chars. CHANNEL_ID = "UCW7jUEpYT_t0Gsf632d6_wQ" @pytest.fixture() def media_root(tmp_path, monkeypatch): """Point the media root at a per-test tmpdir.""" root = tmp_path / "media" root.mkdir(parents=True) monkeypatch.setattr(config, "MEDIA_ROOT", root) return root @pytest.fixture() def conn(tmp_path): connection = db.connect(tmp_path / "ytstream.db") yield connection connection.close() @pytest.fixture() def settings(conn): settings = Settings(conn) # Tests that reach the API go through FakeApi, but the code refuses to call # out at all without a key, so give it one that is never used for real. settings.set("youtube_api_key", "test-key") return settings @pytest.fixture() def channel(conn): """One subscribed channel, returned as a row.""" return add_channel(conn, CHANNEL_ID, "clabretro", "clabretro") @pytest.fixture() def no_network(monkeypatch): """Fail loudly if anything tries to open a socket. Belt and braces: a test that accidentally hits the network would pass locally and fail in a different week for reasons nobody could reproduce. """ import urllib.request def forbidden(*args, **kwargs): raise AssertionError("test attempted a network call") monkeypatch.setattr(urllib.request, "urlopen", forbidden) # --------------------------------------------------------------------- helpers def add_channel(conn, channel_id: str, title: str, dir_name: str, **kwargs): fields = {"source": "youtube", "backfilled": 1, "uploads_playlist": "UULF"} fields.update(kwargs) columns = ", ".join(fields) marks = ", ".join("?" * len(fields)) with conn: conn.execute( f"INSERT INTO channel (channel_id, handle, title, description, dir_name, " f"added_at, {columns}) VALUES (?, ?, ?, ?, ?, ?, {marks})", (channel_id, f"@{dir_name}", title, "A channel", dir_name, util.utcnow_iso(), *fields.values()), ) return conn.execute( "SELECT * FROM channel WHERE channel_id = ?", (channel_id,) ).fetchone() def add_video(conn, channel_pk, video_id, **kwargs): """Insert a video row with sensible defaults.""" defaults = { "title": f"Video {video_id}", "upload_date": "2026-08-01", "state": videos.LISTED, "discovery_source": videos.SOURCE_UULF, "duration": 900, } defaults.update(kwargs) videos.insert(conn, channel_pk=channel_pk, video_id=video_id, **defaults) return videos.get(conn, video_id) def feed_bytes(name: str) -> bytes: return (FIXTURES / name).read_bytes() def make_feed(entries: list[dict], *, playlist_published: str = "2019-01-01T00:00:00+00:00") -> bytes: """Build an Atom feed shaped like YouTube's. Includes the feed-level that is NOT an entry, because scraping timestamps instead of walking atom:entry picks it up and yields nonsense upload rates — a mistake made once while measuring. """ items = "".join( f""" yt:video:{e['video_id']} {e['video_id']} {e.get('title', 'Untitled')} {e['published']} """ for e in entries ) return f""" Videos {playlist_published}{items} """.encode() class FakeApi: """Stand-in for `api.Api` with scriptable responses. Counts calls so tests can assert the batching actually batches — the whole quota argument in the plan rests on 50 ids per call, and a regression to one call per video would be silent and expensive. """ def __init__(self, *, subs=None, uploads=None, durations=None, channel=None, raises=None): self._subs = subs self._uploads = uploads or [] self._durations = durations or {} self._channel = channel self._raises = raises self.calls = 0 self.duration_calls = 0 self.subscription_calls = 0 def subscriptions(self, channel_id): self.subscription_calls += 1 self.calls += 1 if isinstance(self._raises, Exception): raise self._raises return list(self._subs or []) def uploads(self, channel_id, *, kind="UULF", since=None, limit=None, page_token=None): self.calls += 1 produced = 0 for entry, token in self._uploads: if since is not None and entry["published"] < since: return yield entry, token produced += 1 if limit is not None and produced >= limit: return def durations(self, video_ids): self.duration_calls += 1 self.calls += 1 return {vid: self._durations[vid] for vid in video_ids if vid in self._durations} def channel(self, channel_id): self.calls += 1 return self._channel def resolve_handle(self, handle): self.calls += 1 return self._channel def patch_api(monkeypatch, module, fake: FakeApi): """Make `module.api.Api(...)` return `fake` regardless of arguments.""" monkeypatch.setattr(module.api, "Api", lambda *a, **kw: fake) return fake