researchowl/tests/test_shortsmith_live.py

"""Prueba de fontanería contra un shortsmith VIVO.

Se salta salvo que se le dé una URL alcanzable desde donde corren los tests:

    SHORTSMITH_LIVE_URL=http://10.43.86.57:8080 pytest tests/test_shortsmith_live.py -v

(dentro del cluster es `http://shortsmith-svc.shortsmith.svc.cluster.local:8080`;
desde el nodo, la ClusterIP de `kubectl get svc -n shortsmith`).

Renderiza el ejemplo de referencia entero — ~30 s de CPU en el pod — y comprueba
que vuelve un MP4. Es el paso 1 del §12 del spec de fase 2: probar el transporte
antes de generar nada.
"""
import json
import os
from pathlib import Path

import pytest

from src.generator.shortsmith import ShortsmithClient

LIVE_URL = os.environ.get("SHORTSMITH_LIVE_URL")
pytestmark = pytest.mark.skipif(
    not LIVE_URL, reason="define SHORTSMITH_LIVE_URL para probar contra el servicio vivo")

EXAMPLE = Path(__file__).resolve().parents[1] / "src/generator/examples/jal1628.json"


@pytest.mark.asyncio
async def test_healthz_and_templates():
    client = ShortsmithClient(LIVE_URL)

    health = await client.health()
    assert health["status"] == "ok"
    assert health["templates"] >= 1

    templates = await client.templates(refresh=True)
    # No se comprueban nombres concretos a propósito: el contrato es de
    # shortsmith y añadir plantillas allí no debe romper aquí.
    assert templates, "GET /templates devolvió vacío"
    for name, schema in templates.items():
        assert schema.get("type") == "object", f"{name} no publica un esquema de objeto"
        assert "properties" in schema

    # Los dos presupuestos de texto, en el contrato SERVIDO. `unreadable_notes`
    # se calla contra un esquema que no los traiga —no puede inventarse el
    # número—, así que un shortsmith anterior a 289d50e apagaría la comprobación
    # entera sin un solo error. Esto es lo único que lo nota.
    sin_tope = [
        f"{name}.{prop}"
        for name, schema in templates.items()
        for owner in [schema, *(schema.get("$defs") or {}).values()]
        for prop, node in (owner.get("properties") or {}).items()
        if "x-fits" in node and "x-fits-hard" not in node
    ]
    assert sin_tope == [], f"campos con x-fits pero sin x-fits-hard: {sin_tope}"


@pytest.mark.asyncio
async def test_the_envelope_served_is_the_envelope_validated_against():
    """El sobre, contra el servicio de verdad — que es donde derivó.

    La copia local decía «pequeña y estable» y aceptaba 1920x1080 días después
    de que shortsmith dejara de renderizarlo. Un test de unidad no podía verlo:
    los dos lados eran coherentes consigo mismos. Esto compara con el servido.

    Contra un shortsmith anterior a `GET /limits` el cliente devuelve el suelo
    sin error, y entonces esto sólo comprueba que el suelo es el estrecho — que
    es exactamente lo que se quiere de un fallback.
    """
    from src.generator.spec_contract import BASELINE_LIMITS, validate_spec, SpecInvalid

    client = ShortsmithClient(LIVE_URL)
    limits = await client.limits(refresh=True)

    resoluciones = [tuple(pair) for pair in limits["resolutions"]]
    assert (1080, 1920) in resoluciones, "el servicio ya no acepta el vertical"

    # Lo que el sobre NO trae, el validador local tiene que rechazarlo. Si
    # shortsmith deja de aceptar una resolución, esto falla el día que pasa y no
    # la generación siguiente.
    spec = json.loads(EXAMPLE.read_text())
    templates = await client.templates(refresh=True)
    for size in [(1920, 1080), (1080, 1080)]:
        if size in resoluciones:
            continue
        doc = json.loads(json.dumps(spec))
        doc["meta"]["width"], doc["meta"]["height"] = size
        with pytest.raises(SpecInvalid):
            validate_spec(doc, templates, limits=limits)

    # Y el suelo nunca puede ser más ancho que lo servido: si lo fuera, un
    # shortsmith caído dejaría pasar specs que el vivo rechaza.
    suelo = {tuple(pair) for pair in BASELINE_LIMITS["resolutions"]}
    assert suelo <= set(resoluciones), (
        f"el suelo {sorted(suelo)} acepta más que el servicio {sorted(resoluciones)}")


@pytest.mark.asyncio
async def test_render_the_reference_example_end_to_end(tmp_path):
    client = ShortsmithClient(LIVE_URL)
    spec = json.loads(EXAMPLE.read_text())

    job_id = await client.render(spec)
    seen = []

    async def on_progress(fraction, status):
        seen.append(fraction)

    result = await client.poll(job_id, on_progress=on_progress)
    assert result.ok, f"el job terminó en {result.status}: {result.error}"
    assert seen and max(seen) == 1.0

    video = await client.fetch_video(job_id)
    # ftyp en los primeros bytes: es un MP4 de verdad, no una página de error.
    assert b"ftyp" in video[:32]
    assert len(video) > 100_000, f"solo {len(video)} bytes — sospechosamente corto"

    out = tmp_path / "jal1628.mp4"
    out.write_bytes(video)
    print(f"\nrenderizado {len(video)/1e6:.2f} MB en {out}")