"""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 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" # --- 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 == []