ytstream/tests/test_api.py

"""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"<html>nope</html>")
    )
    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