researchowl/tests/test_youtube.py

"""Subida a YouTube: auth, los dos pasos del resumable, y los metadatos.

Todo contra un servidor falso. No hay test en vivo: cualquier ejecución real
sube un vídeo a un canal de verdad, y eso no es algo que deba pasar por teclear
`pytest`.
"""
import json

import pytest

from src.generator import youtube as yt
from src.generator.youtube import (
    UploadedVideo, YouTubeAuthError, YouTubeDisabled, YouTubeError,
    YouTubeNotConfigured, YouTubeQuotaExceeded, YouTubeRejected,
    YouTubeUploader, build_metadata,
)

SPEC = {
    "meta": {"id": "jal1628", "title": "JAL 1628: Three Radars, One Object"},
    "shots": [
        {"template": "scale_bars", "props": {
            "headline": "REPORTED SCALE",
            "attribution": "— CAPT. TERAUCHI, ESTIMATE"}},
        {"template": "document_quote", "props": {
            "source": "FAA · 5 MARCH 1987", "quote_a": "“SPLIT RADAR IMAGE”"}},
        {"template": "counter_close", "props": {"count_to": 1500}},
    ],
}


class FakeResp:
    def __init__(self, status, payload=None, body=None, headers=None):
        self.status = status
        self.headers = headers or {}
        if body is None:
            body = json.dumps(payload) if payload is not None else ""
        self._body = body

    async def __aenter__(self):
        return self

    async def __aexit__(self, *a):
        return False

    async def text(self):
        return self._body

    async def json(self):
        return json.loads(self._body)


class FakeSession:
    def __init__(self, routes):
        self.routes = routes
        self.calls = []

    async def __aenter__(self):
        return self

    async def __aexit__(self, *a):
        return False

    def _next(self, method, url):
        self.calls.append((method, url))
        for pattern, responses in self.routes.items():
            if pattern in url:
                if isinstance(responses, list):
                    return responses.pop(0) if len(responses) > 1 else responses[0]
                return responses
        raise AssertionError(f"ruta no simulada: {method} {url}")

    def post(self, url, **kw):
        return self._next("POST", url)

    def put(self, url, **kw):
        return self._next("PUT", url)


@pytest.fixture(autouse=True)
def clean_token_cache():
    yt._token_cache.clear()
    yield
    yt._token_cache.clear()


@pytest.fixture
def uploader():
    return YouTubeUploader(client_id="cid", client_secret="secret",
                           refresh_token="refresh")


def patch(client, routes):
    session = FakeSession(routes)
    client._session = lambda total: session
    return session


@pytest.fixture
def video(tmp_path):
    path = tmp_path / "166.mp4"
    path.write_bytes(b"\x00\x00\x00 ftypisom" + b"\x00" * 4096)
    return path


TOKEN_OK = FakeResp(200, {"access_token": "at-1", "expires_in": 3600})
VIDEO_OK = {"id": "abc123", "snippet": {"title": "JAL 1628"},
            "status": {"privacyStatus": "private", "uploadStatus": "uploaded"}}


# --- auth ------------------------------------------------------------------

@pytest.mark.asyncio
async def test_access_token_is_cached_across_instances(uploader):
    session = patch(uploader, {"/token": TOKEN_OK})
    assert await uploader.access_token() == "at-1"

    # Otro uploader, mismo client_id: el bot construye uno nuevo por comando y
    # no debe pagar un refresco cada vez.
    twin = YouTubeUploader(client_id="cid", client_secret="s", refresh_token="r")
    patch(twin, {})  # sin rutas: si intentara pedirlo, reventaría
    assert await twin.access_token() == "at-1"
    assert len(session.calls) == 1


@pytest.mark.asyncio
async def test_expired_token_is_refreshed(uploader):
    patch(uploader, {"/token": [
        FakeResp(200, {"access_token": "at-1", "expires_in": 0}),
        FakeResp(200, {"access_token": "at-2", "expires_in": 3600}),
    ]})
    assert await uploader.access_token() == "at-1"
    assert await uploader.access_token() == "at-2"


@pytest.mark.asyncio
async def test_invalid_grant_points_at_the_testing_screen(uploader):
    """El fallo que se va a encontrar de verdad, y el que menos se adivina."""
    patch(uploader, {"/token": FakeResp(
        400, body=json.dumps({"error": "invalid_grant",
                              "error_description": "Token has been expired or revoked."}))})

    with pytest.raises(YouTubeAuthError) as exc:
        await uploader.access_token()
    message = str(exc.value)
    assert "Testing" in message and "7 días" in message


@pytest.mark.asyncio
async def test_unconfigured_uploader_says_so():
    bare = YouTubeUploader(client_id="", client_secret="", refresh_token="")
    assert not bare.is_configured()
    with pytest.raises(YouTubeNotConfigured):
        await bare.access_token()


# --- subida ----------------------------------------------------------------

@pytest.mark.asyncio
async def test_upload_does_metadata_then_bytes(uploader, video):
    session = patch(uploader, {
        "/token": TOKEN_OK,
        "/upload/youtube": FakeResp(200, {}, headers={"Location": "https://up/xyz"}),
        "https://up/xyz": FakeResp(200, VIDEO_OK),
    })

    seen = []
    result = await uploader.upload(video, build_metadata(SPEC, "JAL 1628"),
                                   on_progress=lambda t: seen.append(t))

    assert result.video_id == "abc123"
    assert result.watch_url == "https://youtube.com/shorts/abc123"
    assert result.studio_url.endswith("/abc123/edit")
    assert [c[0] for c in session.calls] == ["POST", "POST", "PUT"]
    assert len(seen) == 3, "cada etapa avisa: autenticar, abrir, subir"


@pytest.mark.asyncio
async def test_missing_location_header_is_fatal(uploader, video):
    """Sin Location no hay dónde mandar los bytes. Falla claro en vez de
    intentar un PUT contra la nada."""
    patch(uploader, {"/token": TOKEN_OK,
                     "/upload/youtube": FakeResp(200, {})})

    with pytest.raises(YouTubeError, match="Location"):
        await uploader.upload(video, build_metadata(SPEC, "x"))


@pytest.mark.asyncio
async def test_forced_private_is_detected(uploader, video):
    """Se pide público, YouTube devuelve privado: la firma del candado del
    proyecto sin auditar. Hay que verlo, no tragárselo."""
    patch(uploader, {
        "/token": TOKEN_OK,
        "/upload/youtube": FakeResp(200, {}, headers={"Location": "https://up/x"}),
        "https://up/x": FakeResp(200, VIDEO_OK),
    })
    meta = build_metadata(SPEC, "x", privacy_status="public")

    result = await uploader.upload(video, meta)
    assert result.privacy_status == "private"
    assert result.forced_private


@pytest.mark.asyncio
async def test_private_request_is_not_reported_as_forced(uploader, video):
    patch(uploader, {
        "/token": TOKEN_OK,
        "/upload/youtube": FakeResp(200, {}, headers={"Location": "https://up/x"}),
        "https://up/x": FakeResp(200, VIDEO_OK),
    })
    result = await uploader.upload(video, build_metadata(SPEC, "x",
                                                         privacy_status="private"))
    assert not result.forced_private


@pytest.mark.asyncio
async def test_quota_exceeded_is_its_own_error(uploader, video):
    patch(uploader, {"/token": TOKEN_OK, "/upload/youtube": FakeResp(403, {
        "error": {"code": 403, "message": "The request cannot be completed.",
                  "errors": [{"reason": "quotaExceeded"}]}})})

    with pytest.raises(YouTubeQuotaExceeded, match="Pacífico"):
        await uploader.upload(video, build_metadata(SPEC, "x"))


@pytest.mark.asyncio
async def test_bad_metadata_is_rejected_with_the_reason(uploader, video):
    patch(uploader, {"/token": TOKEN_OK, "/upload/youtube": FakeResp(400, {
        "error": {"code": 400, "message": "Invalid video title.",
                  "errors": [{"reason": "invalidTitle"}]}})})

    with pytest.raises(YouTubeRejected) as exc:
        await uploader.upload(video, build_metadata(SPEC, "x"))
    assert exc.value.reason == "invalidTitle"
    assert "Invalid video title" in str(exc.value)


@pytest.mark.asyncio
async def test_401_on_upload_is_an_auth_error(uploader, video):
    patch(uploader, {"/token": TOKEN_OK, "/upload/youtube": FakeResp(401, {
        "error": {"code": 401, "message": "Invalid Credentials"}})})

    with pytest.raises(YouTubeAuthError):
        await uploader.upload(video, build_metadata(SPEC, "x"))


@pytest.mark.asyncio
async def test_missing_and_empty_files_never_reach_the_network(uploader, tmp_path):
    patch(uploader, {})  # cualquier petición reventaría
    with pytest.raises(YouTubeError, match="no existe"):
        await uploader.upload(tmp_path / "nope.mp4", {})

    empty = tmp_path / "empty.mp4"
    empty.write_bytes(b"")
    with pytest.raises(YouTubeError, match="vacío"):
        await uploader.upload(empty, {})


@pytest.mark.asyncio
async def test_kill_switch(uploader, video, monkeypatch):
    monkeypatch.setattr(yt.settings, "youtube_enabled", False)
    patch(uploader, {})
    with pytest.raises(YouTubeDisabled):
        await uploader.upload(video, {})


# --- metadatos -------------------------------------------------------------

def test_metadata_carries_the_article_link_and_the_citations():
    meta = build_metadata(SPEC, "JAL 1628 Alaska sighting",
                          article_url="https://theexclusionzone.com/jal-1628/")
    description = meta["snippet"]["description"]

    assert "https://theexclusionzone.com/jal-1628/" in description
    # Verbatim: suavizar mayúsculas convierte FAA en Faa.
    assert "FAA · 5 MARCH 1987" in description
    assert "CAPT. TERAUCHI, ESTIMATE" in description
    # El guión de la atribución no se duplica con el de la lista.
    assert "— — " not in description


def test_metadata_without_article_url_still_builds():
    description = build_metadata(SPEC, "JAL 1628")["snippet"]["description"]
    assert "Full investigation" not in description
    assert "JAL 1628" in description


def test_title_comes_from_the_spec_and_is_truncated():
    long_spec = {"meta": {"title": "A" * 200}, "shots": []}
    assert len(build_metadata(long_spec, "x")["snippet"]["title"]) == 100


def test_title_falls_back_to_the_topic():
    assert build_metadata({"shots": []}, "Socorro 1964")["snippet"]["title"] \
        == "Socorro 1964"


def test_tags_drop_stopwords_and_duplicates_and_respect_the_limit():
    tags = build_metadata(SPEC, "The Landing of the UFO in Socorro New Mexico"
                          )["snippet"]["tags"]
    lowered = [t.casefold() for t in tags]

    assert "the" not in lowered and "of" not in lowered and "in" not in lowered
    assert len(lowered) == len(set(lowered))
    assert "socorro" in lowered
    assert sum(len(t) + 1 for t in tags) <= yt.MAX_TAGS_CHARS


def test_the_whole_topic_is_one_tag():
    """Partido en palabras deja "New" y "Mexico" sueltas, que no buscan igual."""
    tags = build_metadata(SPEC, "Socorro New Mexico 1964")["snippet"]["tags"]
    assert "Socorro New Mexico 1964" in tags
    assert "Socorro" in tags, "las sueltas también, que cuestan poco"


def test_tags_stay_under_the_limit_with_an_absurd_topic():
    tags = build_metadata(SPEC, " ".join(f"palabra{i}" for i in range(200))
                          )["snippet"]["tags"]
    assert sum(len(t) + 1 for t in tags) <= yt.MAX_TAGS_CHARS


def test_made_for_kids_is_declared():
    """Sin declararlo la subida puede quedar en un limbo que no se ve por API."""
    assert build_metadata(SPEC, "x")["status"]["selfDeclaredMadeForKids"] is False


def test_privacy_defaults_to_private():
    assert build_metadata(SPEC, "x")["status"]["privacyStatus"] == "private"


def test_description_is_capped():
    spec = {"meta": {"title": "t"},
            "shots": [{"props": {"source": "S" * 400}} for _ in range(40)]}
    description = build_metadata(spec, "x")["snippet"]["description"]
    assert len(description) <= yt.MAX_DESCRIPTION


def test_uploaded_video_urls():
    video = UploadedVideo(video_id="xyz", title="t", privacy_status="private")
    assert video.watch_url == "https://youtube.com/shorts/xyz"
    assert video.studio_url == "https://studio.youtube.com/video/xyz/edit"