ContentAutomation/tests/test_youtube_resumable_upload.py

from __future__ import annotations

import json
from datetime import UTC, datetime, timedelta
from pathlib import Path

from content_automation.adapters.social.youtube import (
    YouTubeAdapter,
    YouTubeDataApiClient,
    YouTubeSnippet,
    YouTubeStatus,
    YouTubeVideoInsertPayload,
)


class FakeInsertRequest:
    def __init__(self, response: dict[str, object]) -> None:
        self._response = response

    def execute(self) -> dict[str, object]:
        return self._response


class FakeResumableRequest:
    def __init__(self) -> None:
        self._calls = 0

    def next_chunk(self):
        self._calls += 1
        if self._calls == 1:
            return object(), None
        return None, {"id": "video-123"}


class FakeVideosResource:
    def __init__(self, request) -> None:
        self._request = request

    def insert(self, part: str, body: dict, media_body=None):
        return self._request


class FakeService:
    def __init__(self, request) -> None:
        self._videos = FakeVideosResource(request)

    def videos(self) -> FakeVideosResource:
        return self._videos


def test_resumable_upload_happy_path(monkeypatch, tmp_path: Path) -> None:
    media_file = tmp_path / "clip.mp4"
    media_file.write_bytes(b"abcdef")

    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.build",
        lambda *args, **kwargs: FakeService(FakeResumableRequest()),
    )
    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.MediaFileUpload",
        lambda *args, **kwargs: object(),
    )

    adapter = YouTubeAdapter(
        access_token="token",
        use_resumable_upload=True,
        resumable_chunk_size=3,
    )

    post_id = adapter.post_media(media_url=media_file.as_uri(), caption="caption")

    assert post_id == "video-123"


def test_regular_upload_happy_path(monkeypatch, tmp_path: Path) -> None:
    media_file = tmp_path / "clip.mp4"
    media_file.write_bytes(b"abcdef")

    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.build",
        lambda *args, **kwargs: FakeService(
            FakeInsertRequest({"id": "video-regular-123"})
        ),
    )
    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.MediaFileUpload",
        lambda *args, **kwargs: object(),
    )

    adapter = YouTubeAdapter(
        access_token="token",
        use_resumable_upload=False,
    )

    post_id = adapter.post_media(media_url=media_file.as_uri(), caption="caption")

    assert post_id == "video-regular-123"


def test_insert_video_happy_path_for_non_local_url(monkeypatch) -> None:
    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.build",
        lambda *args, **kwargs: FakeService(
            FakeInsertRequest({"id": "video-insert-123"})
        ),
    )

    adapter = YouTubeAdapter(access_token="token")
    post_id = adapter.post_media(
        media_url="https://cdn.example.com/path/to/video.mp4", caption="caption"
    )

    assert post_id == "video-insert-123"


def test_client_refreshes_expired_token_before_request(monkeypatch) -> None:
    refreshed_tokens: list[str] = []

    def fake_refresh(self, request) -> None:
        self.token = "new-token"
        self.expiry = datetime.now(UTC).replace(tzinfo=None) + timedelta(minutes=30)
        refreshed_tokens.append(self.token)

    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.Credentials.refresh",
        fake_refresh,
    )
    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.build",
        lambda *args, **kwargs: FakeService(
            FakeInsertRequest({"id": "video-refreshed"})
        ),
    )

    client = YouTubeDataApiClient(
        access_token="expired-token",
        category_id="22",
        privacy_status="public",
        refresh_token="refresh-token",
        client_id="client-id",
        client_secret="client-secret",
        expiry="2024-01-01T00:00:00Z",
    )

    payload = YouTubeVideoInsertPayload(
        snippet=YouTubeSnippet(
            title="title",
            description="description",
            categoryId="22",
        ),
        status=YouTubeStatus(privacyStatus="public"),
        sourceUrl="https://cdn.example.com/video.mp4",
    )
    response = client.insert_video(part="snippet,status", payload=payload)

    assert response["id"] == "video-refreshed"
    assert refreshed_tokens == ["new-token"]


def test_obtain_credentials_from_client_secret_file(
    monkeypatch, tmp_path: Path
) -> None:
    captured: dict[str, object] = {}

    class FakeCredentials:
        def to_json(self) -> str:
            return json.dumps(
                {
                    "token": "token-123",
                    "refresh_token": "refresh-123",
                    "token_uri": "https://oauth2.googleapis.com/token",
                    "client_id": "client-123",
                    "client_secret": "secret-123",
                    "scopes": ["https://www.googleapis.com/auth/youtube.upload"],
                }
            )

    class FakeFlow:
        def run_local_server(self):
            return FakeCredentials()

    def fake_from_client_secrets_file(client_secret_file: str, scopes: list[str]):
        captured["client_secret_file"] = client_secret_file
        captured["scopes"] = scopes
        return FakeFlow()

    monkeypatch.setattr(
        "content_automation.adapters.social.youtube.InstalledAppFlow.from_client_secrets_file",
        fake_from_client_secrets_file,
    )

    client_secret_path = tmp_path / "client_secret.json"
    token_output_path = tmp_path / "youtube_credentials.json"

    credentials_payload = YouTubeAdapter.obtain_credentials_from_client_secret_file(
        client_secret_file_path=client_secret_path,
        scopes=["https://www.googleapis.com/auth/youtube.upload"],
        token_output_path=token_output_path,
    )

    assert captured["client_secret_file"] == str(client_secret_path)
    assert captured["scopes"] == ["https://www.googleapis.com/auth/youtube.upload"]
    assert credentials_payload["token"] == "token-123"
    assert token_output_path.exists()
    assert (
        json.loads(token_output_path.read_text(encoding="utf-8"))["token"]
        == "token-123"
    )