#!/usr/bin/env python3
"""probar-uevent-relay.py — ¿un uevent del kernel LLEGA al grupo que mira libudev-zero?

El síntoma que esto cubre es «los accesorios no aparecen cuando se conectan después de iniciar»
(metal, 2026-10-15). La causa está en `crates/uevent-relay/src/main.rs`: libudev-zero abre su
monitor en el grupo netlink 0x4 y sin `udevd` nadie emite ahí nunca.

**El brazo de control es gratis y por eso es obligatorio:** en una máquina con udev de verdad,
`udevd` reenvía al grupo 0x2 — nunca al 0x4. Así que todo lo que aparezca en el 0x4 lo puso este
relay y nada más. La prueba corre las dos mitades:

    1. CON el relay vivo  ⇒ el evento disparado a mano aparece en el grupo 4.
    2. SIN el relay       ⇒ no aparece (si apareciera, la prueba no estaría midiendo el relay).

No necesita hardware: el uevent se dispara escribiendo `change` en el `uevent` de un dispositivo
que ya existe, que es exactamente el mensaje que el kernel emite al enchufar uno nuevo.

    sudo python3 scripts/probar-uevent-relay.py [ruta-al-binario]
"""
import os
import socket
import subprocess
import sys
import time

NETLINK_KOBJECT_UEVENT = 15
GRUPO_UDEV = 4          # el que mira libudev-zero
ESPERA = 4.0

BIN = sys.argv[1] if len(sys.argv) > 1 else "target/debug/uevent-relay"


def fallar(msg):
    print(f"✗ FALLO: {msg}")
    sys.exit(1)


def escucha_grupo_4():
    s = socket.socket(socket.AF_NETLINK, socket.SOCK_DGRAM, NETLINK_KOBJECT_UEVENT)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
    s.bind((0, GRUPO_UDEV))
    s.settimeout(ESPERA)
    return s


def un_dispositivo_cualquiera():
    """Alguno que exista y tenga `uevent` escribible. Un input es lo más parecido al caso real."""
    for base in ("/sys/class/input", "/sys/class/net", "/sys/block"):
        if not os.path.isdir(base):
            continue
        for n in sorted(os.listdir(base)):
            p = os.path.join(base, n, "uevent")
            if os.access(p, os.W_OK):
                return p
    return None


def disparar(uevent_path):
    with open(uevent_path, "w") as f:
        f.write("change")


def recibir(s):
    try:
        return s.recv(16384)
    except socket.timeout:
        return None


if os.geteuid() != 0:
    print("este test necesita root (bind a un grupo netlink y escribir en /sys)")
    sys.exit(2)
if not os.path.exists(BIN):
    print(f"no encuentro {BIN} — cargo build -p uevent-relay")
    sys.exit(2)

uevent = un_dispositivo_cualquiera()
if not uevent:
    print("no encontré ningún dispositivo con `uevent` escribible: no se puede medir")
    sys.exit(2)
print(f"== disparando sobre {uevent} ==")

# ── 1. EL BRAZO DE CONTROL, PRIMERO: sin el relay, el grupo 4 está mudo ───────────────────────
# Va primero a propósito: si se corriera después, un evento rezagado del relay anterior podría
# colarse y el control pasaría por casualidad.
s = escucha_grupo_4()
disparar(uevent)
sobra = recibir(s)
s.close()
if sobra is not None:
    fallar(
        "SIN el relay ya llega algo al grupo 4: esta prueba no está midiendo el relay\n"
        f"    recibido: {sobra[:120]!r}"
    )
print("  ✓ control: sin el relay, el grupo 4 no recibe nada (udevd emite al 2, no al 4)")

# ── 2. CON el relay: el mismo evento aparece ──────────────────────────────────────────────────
s = escucha_grupo_4()
env = dict(os.environ, UEVENT_RELAY_DEBUG="1")
relay = subprocess.Popen([BIN], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
time.sleep(0.6)   # que ligue su socket antes de disparar
if relay.poll() is not None:
    fallar(f"el relay murió al arrancar:\n{relay.stderr.read()}")

disparar(uevent)
got = recibir(s)
s.close()
relay.terminate()
try:
    _, err = relay.communicate(timeout=3)
except subprocess.TimeoutExpired:
    relay.kill()
    _, err = relay.communicate()

if got is None:
    fallar(f"el evento NO llegó al grupo 4 en {ESPERA}s\n    stderr del relay: {err[:400]}")

texto = got.decode("utf-8", "replace")
campos = dict(
    p.split("=", 1) for p in texto.split("\0") if "=" in p
)
print(f"  ✓ llegó al grupo 4: {len(got)} bytes")
for clave in ("ACTION", "DEVPATH", "SUBSYSTEM", "SEQNUM"):
    if clave not in campos:
        fallar(f"el evento reenviado no trae {clave} — libudev-zero lo descartaría (exige las cuatro)")
print(f"  ✓ trae las cuatro claves que libudev-zero exige: "
      f"ACTION={campos['ACTION']} SUBSYSTEM={campos['SUBSYSTEM']} DEVPATH={campos['DEVPATH'][:48]}…")

# El mensaje reenviado tiene que ser IDÉNTICO en forma al del kernel: si el relay lo reescribiera,
# libudev-zero sintetizaría propiedades distintas de las que sintetiza al enumerar, y un
# dispositivo se comportaría distinto según si estaba enchufado al arrancar o no.
if not texto.startswith(("add@", "change@", "remove@", "bind@", "unbind@", "move@", "online@", "offline@")):
    fallar(f"el mensaje reenviado no empieza con la línea de acción del kernel: {texto[:60]!r}")
print("  ✓ el mensaje va tal cual lo emitió el kernel (misma forma ⇒ mismas propiedades sintetizadas)")

print("\n✓✓ probar-uevent-relay: el hotplug llega a donde libudev-zero lo está esperando")