takana/scripts/test-atuq-medios.py

#!/usr/bin/env python3
"""¿Un medio se abre en el reproductor de la distro y no en una pestaña? (SDD 26 §6.6)

    python3 scripts/test-atuq-medios.py                     # el vídeo sale por mpv
    python3 scripts/test-atuq-medios.py --negative-control   # SIN mpv: la navegación vuelve al navegador

── QUÉ SE EJERCITA ───────────────────────────────────────────────────────────────────────────
Un servidor HTTP sirve un audio real (`audio/wav`); el navegador navega a él; la extensión ve por la
cabecera que es un medio, **cancela la navegación** y le pasa la URL al host, que lanza el reproductor
de la distro. Se comprueban las tres cosas: que la extensión lo detectó y canceló, que el host
contestó con el **pid** del proceso, y que **el reproductor de verdad abrió esa URL** — su propia
salida aparece en el log, porque hereda la del navegador.

── EL CONTROL NEGATIVO PRUEBA LA PROMESA QUE MÁS IMPORTA ─────────────────────────────────────
`--negative-control` borra `/usr/bin/mpv` dentro de la jaula y exige que la navegación **vuelva al
navegador**. La decisión de cancelar es síncrona y la respuesta del host llega después, así que el
único modo de fallo inaceptable es quedarse sin vídeo Y sin pestaña. Sin este control, una extensión
que cancelara siempre se vería idéntica a una que funciona — hasta el día que falte el reproductor.

REQUISITOS: el rootfs de un escritorio (el mínimo de `atuq-nested.sh` NO trae `mpv`), y `atuq` +
`puriy-costura` sellados.
"""
import http.server
import os
import shutil
import socket
import struct
import subprocess
import sys
import tempfile
import threading

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RFS = os.environ.get("RFS", "/mnt/cosecha/escritorios/sway-rootfs")
ALPINE = os.path.join(ROOT, ".dev-fs/alpine")
HAMMER = os.path.join(ROOT, "target/release/takana")
STORE = os.environ.get("HAMMER_STORE", os.path.join(ROOT, "store"))
MPV = "/usr/bin/mpv"


def fatal(msg):
    print(f"✗ {msg}", file=sys.stderr)
    raise SystemExit(1)


def artefacto(receta, nombre, escotilla=None):
    forzado = os.environ.get(escotilla) if escotilla else None
    if forzado:
        if not os.path.isdir(forzado):
            fatal(f"{escotilla}={forzado} no existe")
        print(f"⚠ {escotilla}: se mide {os.path.basename(forzado)}, que NO es necesariamente el vigente")
        return forzado
    r = subprocess.run([HAMMER, "--store", STORE, "hash", f"recipes/{receta}.toml"],
                       cwd=ROOT, capture_output=True, text=True)
    if r.returncode != 0:
        fatal(f"no pude hashear {receta}: {r.stderr.strip()}")
    h = r.stdout.strip().splitlines()[-1].removeprefix("b3:")
    d = os.path.join(STORE, f"{h}-{nombre}")
    if not os.path.isdir(d):
        fatal(f"{nombre} b3:{h[:8]} no está sellado — construilo antes")
    return d


def wav(segundos=0.3, hz=8000):
    """Un WAV real, hecho a mano: sin esto habría que traer un binario de prueba al
    repo, y un fichero opaco en un test es una dependencia que nadie revisa."""
    n = int(segundos * hz)
    datos = b"\x00" * n
    cab = (b"RIFF" + (36 + len(datos)).to_bytes(4, "little") + b"WAVEfmt "
           + (16).to_bytes(4, "little") + (1).to_bytes(2, "little") + (1).to_bytes(2, "little")
           + hz.to_bytes(4, "little") + hz.to_bytes(4, "little")
           + (1).to_bytes(2, "little") + (8).to_bytes(2, "little")
           + b"data" + len(datos).to_bytes(4, "little"))
    return cab + datos


AUDIO = wav()


class Servidor(http.server.SimpleHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_GET(self):
        if self.path.startswith("/audio.wav"):
            datos, tipo = AUDIO, "audio/wav"
        elif self.path.startswith("/ir.html"):
            # ⚠ NO se navega al medio desde la línea de comandos, y la razón está
            # medida: una URL en el argumento se pide en el ARRANQUE y compite con la
            # inicialización de la extensión — la misma carrera que ya apareció en el
            # guardián de `sct`, sólo que acá hay una sola carga y no hay segunda
            # oportunidad. Esta página espera tres segundos y recién entonces va al
            # medio, que además es lo que hace una persona: seguir un enlace.
            datos = (b'<!doctype html><html><head><meta charset="utf-8">'
                     b'<meta http-equiv="refresh" content="3;url=/audio.wav">'
                     b"</head><body>yendo al medio</body></html>")
            tipo = "text/html; charset=utf-8"
        else:
            datos, tipo = b"<html><body>no soy un medio</body></html>", "text/html"
        self.send_response(200)
        self.send_header("Content-Type", tipo)
        self.send_header("Content-Length", str(len(datos)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(datos)


def puerto_libre():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p


def main():
    negativo = "--negative-control" in sys.argv[1:]
    if negativo:
        print("== MODO CONTROL NEGATIVO: sin mpv; la navegación tiene que VOLVER al navegador")
    if not os.path.isdir(RFS):
        fatal(f"no está el rootfs en {RFS} — hidratá un perfil de escritorio (trae `mpv`)")
    if not os.path.isfile(os.path.join(RFS, MPV.lstrip("/"))):
        fatal(f"{RFS} no trae {MPV}: este guardián necesita el reproductor de la imagen")
    atuq = artefacto("atuq", "atuq", "ATUQ_DIR")
    costura = artefacto("puriy-costura", "puriy-costura")
    print(f"artefactos: {os.path.basename(atuq)}\n            {os.path.basename(costura)}")

    puerto = puerto_libre()
    servidor = http.server.ThreadingHTTPServer(("127.0.0.1", puerto), Servidor)
    threading.Thread(target=servidor.serve_forever, daemon=True).start()
    url = f"http://127.0.0.1:{puerto}/audio.wav"
    partida = f"http://127.0.0.1:{puerto}/ir.html"

    tmp = tempfile.mkdtemp(prefix="atuq-medios-")
    salida = os.path.join(tmp, "salida")
    os.makedirs(os.path.join(salida, "perfil"))
    # La configuración del reproductor la pone el TEST, no el producto: sin salida de
    # vídeo ni de audio (no hay pantalla ni tarjeta en la jaula) y con un tope de
    # duración para que termine solo. Meter esto en el host sería cablear el arnés
    # dentro del artefacto.
    conf = os.path.join(salida, "hogar", ".config", "mpv")
    os.makedirs(conf)
    with open(os.path.join(conf, "mpv.conf"), "w") as fh:
        # `log-file` es la evidencia: la salida del reproductor ya NO va al stdout del
        # host —eso era el bug que este guardián encontró, ver §6.6— y Gecko no vuelca
        # el stderr del host al del navegador, así que hay que pedirle al propio
        # reproductor que deje su rastro donde se pueda leer.
        fh.write("vo=null\nao=null\nlength=0.2\nmsg-level=all=info\nlog-file=/salida/mpv.log\n")
    try:
        with open(os.path.join(salida, "perfil", "user.js"), "w") as fh:
            fh.write('user_pref("browser.dom.window.dump.enabled", true);\n')
            fh.write('user_pref("browser.shell.checkDefaultBrowser", false);\n')

        borrar = f"rm -f {MPV}" if negativo else "true"
        guion = f"""
        mkdir -p /root /salida/hogar
        [ -s /etc/machine-id ] || tr -d - < /proc/sys/kernel/random/uuid > /etc/machine-id
        export HOME=/salida/hogar
        {borrar}
        timeout 50 /usr/bin/atuq --profile /salida/perfil --no-remote --headless '{partida}' \
            > /salida/consola.log 2>&1
        exit 0
        """
        subprocess.run([
            "bwrap",
            "--overlay-src", ALPINE, "--overlay-src", RFS,
            "--overlay-src", atuq, "--overlay-src", costura,
            "--tmp-overlay", "/",
            "--dev-bind", "/dev", "/dev", "--proc", "/proc", "--bind", "/sys", "/sys",
            "--tmpfs", "/tmp", "--tmpfs", "/run", "--unshare-pid",
            "--bind", salida, "/salida",
            "--uid", "0", "--gid", "0",
            "--setenv", "PATH", "/usr/bin:/bin:/usr/sbin:/sbin",
            "--setenv", "XDG_RUNTIME_DIR", "/run",
            "--setenv", "XDG_DATA_DIRS", "/usr/share", "--setenv", "MOZ_HEADLESS", "1",
            "--setenv", "MOZ_DISABLE_CONTENT_SANDBOX", "1",
            "--setenv", "MOZ_DISABLE_GMP_SANDBOX", "1",
            "--setenv", "MOZ_DISABLE_RDD_SANDBOX", "1",
            "--setenv", "MOZ_DISABLE_SOCKET_PROCESS_SANDBOX", "1",
            "--setenv", "MOZ_DISABLE_UTILITY_SANDBOX", "1",
            "sh", "-c", guion,
        ], check=False)

        log = open(os.path.join(salida, "consola.log"), errors="replace").read()
        lineas = [l[len("MEDIOS "):] for l in log.splitlines() if l.startswith("MEDIOS ")]
        for l in lineas:
            print("   " + l)
        if not lineas:
            destino = os.path.join(ROOT, "work", "atuq-medios.log")
            os.makedirs(os.path.dirname(destino), exist_ok=True)
            open(destino, "w").write(log)
            fatal(f"la extensión no dijo nada — log en {destino}")

        detecto = any(l.startswith("MEDIO ") for l in lineas)
        cancelo = any(l.startswith("CANCELADA ") for l in lineas)
        abierto = [l for l in lineas if l.startswith("ABIERTO ")]
        volvio = [l for l in lineas if l.startswith("VUELVE AL NAVEGADOR")]
        sin_repro = [l for l in lineas if l.startswith("SIN REPRODUCTOR")]

        if not detecto:
            fatal("la extensión no reconoció el `audio/wav` como medio")

        if negativo:
            if abierto:
                fatal(f"sin mpv dijo que abrió algo: {abierto}")
            if not sin_repro:
                fatal("sin mpv, el host tenía que fallar DICIÉNDOLO y la extensión decirlo")
            if not volvio:
                fatal("sin reproductor, la navegación NO volvió al navegador —"
                      " el usuario se quedaría sin vídeo y sin pestaña")
            print("\n✓ control negativo: sin reproductor no se finge nada y la navegación vuelve"
                  " al navegador")
            return 0

        if not cancelo:
            fatal("la extensión no canceló la navegación: el vídeo seguiría en una pestaña")
        if not abierto:
            fatal("el host no confirmó haber abierto el reproductor")
        if MPV not in abierto[0]:
            fatal(f"abrió algo que no es el reproductor de la imagen: {abierto[0]}")
        if url not in abierto[0]:
            fatal(f"abrió otra URL: {abierto[0]}")
        if volvio:
            fatal(f"volvió al navegador habiendo reproductor: {volvio}")

        # Y la prueba de que el reproductor CORRIÓ de verdad, y no que hubo un `spawn`:
        # su propio log. Su salida ya no va al stdout del host (eso era el bug) y Gecko
        # no vuelca el stderr del host al del navegador, así que la evidencia se le pide
        # al reproductor.
        ruta_mpv = os.path.join(salida, "mpv.log")
        if not os.path.isfile(ruta_mpv):
            destino = os.path.join(ROOT, "work", "atuq-medios.log")
            os.makedirs(os.path.dirname(destino), exist_ok=True)
            open(destino, "w").write(log)
            fatal(f"el reproductor no dejó su log: no llegó a correr — consola en {destino}")
        propio = open(ruta_mpv, errors="replace").read()
        if url not in propio:
            fatal(f"el reproductor corrió pero no abrió esa URL — su log dice:\n{propio[:400]}")
        for l in [x for x in propio.splitlines() if "Playing" in x or "AO:" in x][:3]:
            print("   mpv: " + l.strip()[:110])

        print("\n✓ una navegación a un medio se CANCELA y la abre el reproductor de la distro")
        print("✓ y el reproductor corrió de verdad: su propia salida lo dice, no un `spawn` a ciegas")
        return 0
    finally:
        servidor.shutdown()
        shutil.rmtree(tmp, ignore_errors=True)


sys.exit(main())