"""The YouTube Data API client, with no YouTube involved. Two things here earn their keep beyond ordinary coverage: * **Error classification.** Three distinct operator mistakes all arrive as HTTP 403 `forbidden`, and the only reliable signal is `error.details[].reason`. That cost three attempts to diagnose by hand once; it should cost nobody a second one. * **`totalResults` is not the list length.** It reported 127 against 119 actually returned, because terminated and private channels still count as subscriptions. """ from __future__ import annotations import io import json import urllib.error import pytest from ytstream import api # ------------------------------------------------------------------- durations @pytest.mark.parametrize("text,expected", [ ("PT1H2M3S", 3723), ("PT21M", 1260), ("PT45S", 45), ("P1DT2H", 93600), ("P1W", 604800), ("PT0S", 0), # live and upcoming videos report this ("", None), ("garbage", None), ("1H2M", None), ]) def test_parse_duration(text, expected): assert api.parse_duration(text) == expected def test_parse_published_keeps_the_exact_string(): """The date drives naming; the exact stamp is kept because approximate_date is wrong by up to two days and episode numbers cannot be re-derived later.""" day, exact = api.parse_published("2026-08-11T16:32:10Z") assert day.isoformat() == "2026-08-11" assert exact == "2026-08-11T16:32:10Z" def test_parse_published_normalises_to_utc(): day, _ = api.parse_published("2026-08-11T23:30:00-08:00") assert day.isoformat() == "2026-08-12" def test_parse_published_tolerates_rubbish(): assert api.parse_published("") == (None, None) assert api.parse_published("not a date")[0] is None # ------------------------------------------------------------------- playlists def test_uploads_playlist_id(): assert api.uploads_playlist_id("UCjCJ2LaOIsPzOoXUTMDI3wg") == "UULFjCJ2LaOIsPzOoXUTMDI3wg" assert api.uploads_playlist_id("UCjCJ2LaOIsPzOoXUTMDI3wg", "UU") == "UUjCJ2LaOIsPzOoXUTMDI3wg" def test_uploads_playlist_id_rejects_non_channel_ids(): with pytest.raises(ValueError): api.uploads_playlist_id("PLnotachannel") # -------------------------------------------------------- error classification def _http_error(code, body): return urllib.error.HTTPError( "https://example/", code, "Forbidden", {}, io.BytesIO(json.dumps(body).encode()), ) def _classify(code, body): return api.Api("k")._classify(_http_error(code, body)) def test_service_disabled_is_not_configured(): """The API is not enabled on the project. Carries an activation URL.""" error = _classify(403, {"error": { "code": 403, "message": "YouTube Data API v3 has not been used in project 510818173753 " "before or it is disabled.", "errors": [{"reason": "accessNotConfigured"}], "details": [{"reason": "SERVICE_DISABLED", "metadata": {"activationUrl": "https://console/enable"}}], }}) assert isinstance(error, api.NotConfigured) def test_key_service_blocked_is_not_configured(): """A different mistake entirely — the key's own API restrictions — and it reports only `forbidden` in errors[].""" error = _classify(403, {"error": { "code": 403, "message": "Requests to this API youtube method " "youtube.api.v3.V3DataVideoService.List are blocked.", "errors": [{"reason": "forbidden"}], "details": [{"reason": "API_KEY_SERVICE_BLOCKED"}], }}) assert isinstance(error, api.NotConfigured) def test_subscription_forbidden_is_its_own_type(): """Must never be mistaken for an empty subscription list.""" error = _classify(403, {"error": { "code": 403, "message": "The requester is not allowed to access the requested subscriptions.", "errors": [{"reason": "subscriptionForbidden"}], }}) assert isinstance(error, api.SubscriptionsPrivate) assert not isinstance(error, api.NotConfigured) def test_transient_server_error_is_plain_api_error(): error = _classify(500, {"error": { "code": 500, "message": "Backend Error", "errors": [{"reason": "backendError"}], }}) assert type(error) is api.ApiError def test_quota_exceeded_is_plain_api_error(): """Retrying tomorrow helps, so it must not be classed as misconfiguration.""" error = _classify(403, {"error": { "code": 403, "message": "The request cannot be completed because you have " "exceeded your quota.", "errors": [{"reason": "quotaExceeded"}], }}) assert type(error) is api.ApiError def test_unparseable_body_still_yields_an_error(): exc = urllib.error.HTTPError( "https://example/", 502, "Bad Gateway", {}, io.BytesIO(b"nope") ) error = api.Api("k")._classify(exc) assert isinstance(error, api.ApiError) assert error.status == 502 def test_missing_key_raises_before_any_request(monkeypatch): import urllib.request monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not be called"))) with pytest.raises(api.NotConfigured): api.Api("").durations(["dQw4w9WgXcQ"]) # -------------------------------------------------------------- request shapes class Recorder: """Captures the URLs requested and replays canned pages.""" def __init__(self, pages): self.pages = list(pages) self.urls = [] def __call__(self, request, timeout=None): self.urls.append(request.full_url) payload = json.dumps(self.pages.pop(0)).encode() class Response(io.BytesIO): def __enter__(self_inner): return self_inner def __exit__(self_inner, *exc): return False return Response(payload) @pytest.fixture() def recorder(monkeypatch): def install(pages): rec = Recorder(pages) monkeypatch.setattr(api.urllib.request, "urlopen", rec) return rec return install def test_subscriptions_pages_to_exhaustion(recorder): rec = recorder([ {"pageInfo": {"totalResults": 127}, "items": [{"snippet": {"title": f"C{i}", "resourceId": {"channelId": f"UC{i:022d}"}}} for i in range(50)], "nextPageToken": "T2"}, {"pageInfo": {"totalResults": 127}, "items": [{"snippet": {"title": "Last", "resourceId": {"channelId": "UC" + "z" * 22}}}]}, ]) found = api.Api("k").subscriptions("UCbrother") assert len(found) == 51 assert len(rec.urls) == 2 assert "pageToken=T2" in rec.urls[1] def test_subscriptions_returns_the_fetched_list_not_total_results(recorder): """Measured: totalResults said 127, pagination returned 119.""" recorder([{"pageInfo": {"totalResults": 127}, "items": [{"snippet": {"title": "One", "resourceId": {"channelId": "UC" + "a" * 22}}}]}]) found = api.Api("k").subscriptions("UCbrother") assert len(found) == 1 def test_subscriptions_deduplicates(recorder): recorder([{"items": [ {"snippet": {"title": "Dup", "resourceId": {"channelId": "UC" + "a" * 22}}}, {"snippet": {"title": "Dup", "resourceId": {"channelId": "UC" + "a" * 22}}}, ]}]) assert len(api.Api("k").subscriptions("UCbrother")) == 1 def test_subscriptions_skips_entries_without_a_channel_id(recorder): recorder([{"items": [ {"snippet": {"title": "Broken", "resourceId": {}}}, {"snippet": {"title": "Fine", "resourceId": {"channelId": "UC" + "a" * 22}}}, ]}]) assert [entry["title"] for entry in api.Api("k").subscriptions("UCb")] == ["Fine"] def test_durations_batches_fifty_ids_per_call(recorder): """The whole quota argument rests on this. One call per video would be a silent 50x regression.""" ids = [f"vid{i:08d}" for i in range(120)] rec = recorder([ {"items": [{"id": vid, "contentDetails": {"duration": "PT10M"}} for vid in ids[start:start + 50]]} for start in (0, 50, 100) ]) found = api.Api("k").durations(ids) assert len(found) == 120 assert len(rec.urls) == 3 def test_durations_flags_livestreams(recorder): recorder([{"items": [ {"id": "live0000001", "contentDetails": {"duration": "PT0S"}, "liveStreamingDetails": {"actualStartTime": "2026-08-01T00:00:00Z"}}, {"id": "vod00000001", "contentDetails": {"duration": "PT30M"}}, ]}]) found = api.Api("k").durations(["live0000001", "vod00000001"]) assert found["live0000001"]["is_live"] is True assert found["vod00000001"]["is_live"] is False def test_uploads_stops_at_the_since_date(recorder): recorder([{"items": [ {"contentDetails": {"videoId": "new00000001", "videoPublishedAt": "2026-08-10T00:00:00Z"}}, {"contentDetails": {"videoId": "old00000001", "videoPublishedAt": "2026-01-01T00:00:00Z"}}, {"contentDetails": {"videoId": "old00000002", "videoPublishedAt": "2025-01-01T00:00:00Z"}}, ], "nextPageToken": "T2"}]) from datetime import date got = list(api.Api("k").uploads("UC" + "a" * 22, since=date(2026, 8, 1))) assert [entry["video_id"] for entry, _ in got] == ["new00000001"] def test_uploads_honours_the_limit(recorder): recorder([{"items": [ {"contentDetails": {"videoId": f"vid{i:08d}", "videoPublishedAt": "2026-08-10T00:00:00Z"}} for i in range(10) ]}]) got = list(api.Api("k").uploads("UC" + "a" * 22, limit=3)) assert len(got) == 3 def test_uploads_skips_entries_with_no_publish_date(recorder): """A private or deleted video keeps its playlist slot but loses its date. Skipping is right; stopping would truncate the backfill at that point.""" recorder([{"items": [ {"contentDetails": {"videoId": "priv0000001"}}, {"contentDetails": {"videoId": "good0000001", "videoPublishedAt": "2026-08-10T00:00:00Z"}}, ]}]) got = list(api.Api("k").uploads("UC" + "a" * 22)) assert [entry["video_id"] for entry, _ in got] == ["good0000001"] def test_uploads_yields_the_next_page_token_for_resumability(recorder): recorder([{"items": [ {"contentDetails": {"videoId": "vid00000001", "videoPublishedAt": "2026-08-10T00:00:00Z"}} ], "nextPageToken": "CARRY_ON"}]) # Take only the first entry: the generator would otherwise follow the token # on to a page the recorder has not been given, which is correct paging # behaviour and covered elsewhere. _, token = next(api.Api("k").uploads("UC" + "a" * 22)) assert token == "CARRY_ON" def test_call_counter_tracks_quota(recorder): recorder([{"items": []}, {"items": []}]) client = api.Api("k") client.durations(["a"]) client.durations(["b"]) assert client.calls == 2