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 aiohttp
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)

    def get(self, url, **kw):
        return self._next("GET", url)


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


@pytest.fixture(autouse=True)
def no_recheck_delay(monkeypatch):
    """La segunda pasada de la comprobación espera 3 s en producción, que es lo
    que tarda YouTube en indexar. Aquí no se espera a nada."""
    monkeypatch.setattr(yt, "_VISIBILITY_RECHECK_DELAY", 0)


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


#: Lo que oEmbed contesta de un vídeo que no se ve sin sesión.
OEMBED_HIDDEN = FakeResp(404, body="Not Found")
#: Y de uno que sí.
OEMBED_VISIBLE = FakeResp(200, {"title": "JAL 1628", "type": "video"})


def patch(client, routes):
    """El servidor falso, con la comprobación de visibilidad ya enrutada.

    `upload()` la hace siempre, así que todo test que suba pasa por oEmbed. Por
    defecto contesta "no se ve", que es lo que se espera de un vídeo privado; el
    test que quiera el caso malo pone su propia ruta `/oembed`.
    """
    session = FakeSession({"/oembed": OEMBED_HIDDEN, **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")
    # Token, metadatos, bytes, y la comprobación de visibilidad — que se
    # reintenta porque el primer 404 puede ser YouTube todavía indexando.
    assert [c[0] for c in session.calls] == ["POST", "POST", "PUT", "GET", "GET"]
    assert len(seen) == 4, "cada etapa avisa: autenticar, abrir, subir, comprobar"


@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
    # Sigue habiendo descripción: las fuentes y los hashtags.
    assert "FAA · 5 MARCH 1987" in description
    assert "#Shorts" in description


def test_the_research_topic_never_reaches_the_description():
    """El `topic` es la consulta de investigación, no prosa que describa el vídeo.

    Ocurrió: el Short de Trans-en-Provence subió al canal en INGLÉS con la
    descripción encabezada por "Trans-en-Provence Francia 1981 GEPAN CNES
    análisis suelo evidencia física" — la consulta tal cual, en español y en
    sopa de palabras clave. El título ya describe el vídeo y ése sí lo escribe
    el modelo en inglés (`meta.title`).
    """
    topic = "Trans-en-Provence Francia 1981 GEPAN CNES análisis suelo evidencia física"
    meta = build_metadata(SPEC, topic)
    assert topic not in meta["snippet"]["description"]
    for palabra in ("análisis", "suelo", "evidencia"):
        assert palabra not in meta["snippet"]["description"]


def test_the_description_survives_a_spec_with_nothing_to_cite():
    """Sin tema y sin citas queda el mínimo, no una cadena vacía."""
    description = build_metadata({"meta": {"title": "T"}, "shots": []},
                                 "cualquier tema")["snippet"]["description"]
    assert description.strip() == "#Shorts #UAP #UFO"


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 Radars of the FAA in Alaska")["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 "radars" in lowered, "el vídeo lo dice en el título"
    assert "faa" in lowered, "el vídeo lo dibuja en un plano"
    assert sum(len(t) + 1 for t in tags) <= yt.MAX_TAGS_CHARS


def test_tags_never_carry_the_spanish_that_only_lived_in_the_search_query():
    """El caso real: el Short de Trans-en-Provence, canal en inglés.

    Se etiquetó con `análisis`, `suelo`, `evidencia` y `física` porque las
    etiquetas salían del `topic`, que es la consulta de investigación. No se
    arregla con una lista de palabras en español —eso es el parche sin fin—
    sino cambiando de fuente: ahora salen del spec, que es inglés por
    construcción, y del tema sólo lo que el vídeo de verdad dice.
    """
    spec = {"meta": {"id": "trans_en_provence_1981",
                     "title": "Trans-en-Provence: Ground Trace, Lab Analysis"},
            "shots": [{"template": "signal_strips",
                       "props": {"headline": "GEPAN ANALYSIS · SOIL"}}]}
    topic = "Trans-en-Provence Francia 1981 GEPAN CNES análisis suelo evidencia física"
    tags = build_metadata(spec, topic)["snippet"]["tags"]
    lowered = {t.casefold() for t in tags}

    for basura in ("análisis", "suelo", "evidencia", "física", "francia", "cnes"):
        assert basura not in lowered, f"{basura} sólo estaba en la consulta"
    # Y lo que el vídeo sí dice sobrevive.
    assert "gepan" in lowered
    assert "trans-en-provence" in lowered


def test_a_topic_word_survives_if_the_video_says_it():
    """La criba no es por idioma, es por si el vídeo lo dice.

    Zimbabwe, Brazil y Texas sólo estaban en la consulta y son justo lo que se
    busca; `BASE AÉREA TALAVERA` está en español y también, porque es el nombre
    de la base y sale dibujado. Un filtro por idioma habría tirado los dos.
    """
    spec = {"meta": {"id": "talavera_1976", "title": "Green Humanoid at the Air Base"},
            "shots": [{"template": "data_card",
                       "props": {"card_title": "BASE AÉREA TALAVERA"}}]}
    tags = build_metadata(spec, "Talavera 1976 OVNI humanoide Base Aérea")["snippet"]["tags"]
    lowered = {t.casefold() for t in tags}

    assert "aérea" in lowered, "está dibujado en pantalla"
    assert "ovni" not in lowered and "humanoide" not in lowered


def test_the_phrase_tag_is_never_cut_mid_word():
    """En cinco de los ocho Shorts generados la frase salía partida —"...GEPAN
    CNES análisis suelo evi"— y una frase partida no la busca nadie."""
    largo = "_".join(["palabra"] * 12)          # muy por encima de MAX_TAG
    spec = {"meta": {"id": largo, "title": "T"}, "shots": []}
    frase = build_metadata(spec, "x")["snippet"]["tags"][len(yt.BASE_TAGS)]

    assert len(frase) <= yt.MAX_TAG
    assert not frase.endswith("palabr"), "cortada a media palabra"
    assert frase.split()[-1] == "palabra"


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"


# --- la visibilidad, comprobada en vez de creída ----------------------------

@pytest.mark.asyncio
async def test_a_video_anyone_can_watch_is_detected(uploader, video):
    """El caso que existe para pillar: la API dice privado y el vídeo se ve.

    Todo el flujo de revisión — informe de fundamento primero, publicar después
    — descansa en que subir NO publique. Si eso deja de ser cierto hay que
    enterarse por el parte de la subida, no por una visita al canal.
    """
    patch(uploader, {
        "/token": TOKEN_OK,
        "/upload/youtube": FakeResp(200, {}, headers={"Location": "https://up/x"}),
        "https://up/x": FakeResp(200, VIDEO_OK),
        "/oembed": OEMBED_VISIBLE,
    })

    result = await uploader.upload(video, build_metadata(SPEC, "x"))

    assert result.privacy_status == "private", "la API sigue diciendo privado"
    assert result.reachable is True
    assert result.visibility_contradiction


@pytest.mark.asyncio
async def test_a_private_video_reports_no_contradiction(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"))

    assert result.reachable is False
    assert not result.visibility_contradiction


@pytest.mark.asyncio
async def test_a_video_visible_only_on_the_second_look_still_counts(uploader):
    """Segundos después de subirlo, oEmbed devuelve 404 de un vídeo que sí se
    ve: aún no está indexado. Un solo vistazo daría por privado justo el vídeo
    que hay que gritar."""
    session = patch(uploader, {"/oembed": [OEMBED_HIDDEN, OEMBED_VISIBLE]})

    assert await uploader.reachable("abc123") is True
    assert len(session.calls) == 2


@pytest.mark.asyncio
async def test_two_hidden_looks_are_enough_to_stop_asking(uploader):
    session = patch(uploader, {"/oembed": OEMBED_HIDDEN})

    assert await uploader.reachable("abc123") is False
    assert len(session.calls) == 2


@pytest.mark.asyncio
async def test_a_network_failure_is_not_knowing_rather_than_privacy(uploader):
    """No se pudo comprobar NO es lo mismo que no se ve. Devolver False aquí
    sería inventarse una garantía a partir de un fallo de red."""
    class Broken:
        async def __aenter__(self): return self
        async def __aexit__(self, *a): return False

        def get(self, url, **kw):
            raise aiohttp.ClientError("sin red")

    uploader._session = lambda total: Broken()

    assert await uploader.reachable("abc123") is None


@pytest.mark.asyncio
async def test_an_upload_without_an_id_is_not_checked(uploader):
    session = patch(uploader, {"/oembed": OEMBED_VISIBLE})

    assert await uploader.reachable("") is None
    assert session.calls == []