sergio/takana

1#!/usr/bin/env python3
2"""probar-uevent-relay.py — ¿un uevent del kernel LLEGA al grupo que mira libudev-zero?
3
4El síntoma que esto cubre es «los accesorios no aparecen cuando se conectan después de iniciar»
5(metal, 2026-10-15). La causa está en `crates/uevent-relay/src/main.rs`: libudev-zero abre su
6monitor en el grupo netlink 0x4 y sin `udevd` nadie emite ahí nunca.
7
8**El brazo de control es gratis y por eso es obligatorio:** en una máquina con udev de verdad,
9`udevd` reenvía al grupo 0x2 — nunca al 0x4. Así que todo lo que aparezca en el 0x4 lo puso este
10relay y nada más. La prueba corre las dos mitades:
11
12    1. CON el relay vivo  ⇒ el evento disparado a mano aparece en el grupo 4.
13    2. SIN el relay       ⇒ no aparece (si apareciera, la prueba no estaría midiendo el relay).
14
15No necesita hardware: el uevent se dispara escribiendo `change` en el `uevent` de un dispositivo
16que ya existe, que es exactamente el mensaje que el kernel emite al enchufar uno nuevo.
17
18    sudo python3 scripts/probar-uevent-relay.py [ruta-al-binario]
19"""
20import os
21import socket
22import subprocess
23import sys
24import time
25
26NETLINK_KOBJECT_UEVENT = 15
27GRUPO_UDEV = 4          # el que mira libudev-zero
28ESPERA = 4.0
29
30BIN = sys.argv[1] if len(sys.argv) > 1 else "target/debug/uevent-relay"
31
32
33def fallar(msg):
34    print(f"✗ FALLO: {msg}")
35    sys.exit(1)
36
37
38def escucha_grupo_4():
39    s = socket.socket(socket.AF_NETLINK, socket.SOCK_DGRAM, NETLINK_KOBJECT_UEVENT)
40    s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
41    s.bind((0, GRUPO_UDEV))
42    s.settimeout(ESPERA)
43    return s
44
45
46def un_dispositivo_cualquiera():
47    """Alguno que exista y tenga `uevent` escribible. Un input es lo más parecido al caso real."""
48    for base in ("/sys/class/input", "/sys/class/net", "/sys/block"):
49        if not os.path.isdir(base):
50            continue
51        for n in sorted(os.listdir(base)):
52            p = os.path.join(base, n, "uevent")
53            if os.access(p, os.W_OK):
54                return p
55    return None
56
57
58def disparar(uevent_path):
59    with open(uevent_path, "w") as f:
60        f.write("change")
61
62
63def recibir(s):
64    try:
65        return s.recv(16384)
66    except socket.timeout:
67        return None
68
69
70if os.geteuid() != 0:
71    print("este test necesita root (bind a un grupo netlink y escribir en /sys)")
72    sys.exit(2)
73if not os.path.exists(BIN):
74    print(f"no encuentro {BIN} — cargo build -p uevent-relay")
75    sys.exit(2)
76
77uevent = un_dispositivo_cualquiera()
78if not uevent:
79    print("no encontré ningún dispositivo con `uevent` escribible: no se puede medir")
80    sys.exit(2)
81print(f"== disparando sobre {uevent} ==")
82
83# ── 1. EL BRAZO DE CONTROL, PRIMERO: sin el relay, el grupo 4 está mudo ───────────────────────
84# Va primero a propósito: si se corriera después, un evento rezagado del relay anterior podría
85# colarse y el control pasaría por casualidad.
86s = escucha_grupo_4()
87disparar(uevent)
88sobra = recibir(s)
89s.close()
90if sobra is not None:
91    fallar(
92        "SIN el relay ya llega algo al grupo 4: esta prueba no está midiendo el relay\n"
93        f"    recibido: {sobra[:120]!r}"
94    )
95print("  ✓ control: sin el relay, el grupo 4 no recibe nada (udevd emite al 2, no al 4)")
96
97# ── 2. CON el relay: el mismo evento aparece ──────────────────────────────────────────────────
98s = escucha_grupo_4()
99env = dict(os.environ, UEVENT_RELAY_DEBUG="1")
100relay = subprocess.Popen([BIN], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
101time.sleep(0.6)   # que ligue su socket antes de disparar
102if relay.poll() is not None:
103    fallar(f"el relay murió al arrancar:\n{relay.stderr.read()}")
104
105disparar(uevent)
106got = recibir(s)
107s.close()
108relay.terminate()
109try:
110    _, err = relay.communicate(timeout=3)
111except subprocess.TimeoutExpired:
112    relay.kill()
113    _, err = relay.communicate()
114
115if got is None:
116    fallar(f"el evento NO llegó al grupo 4 en {ESPERA}s\n    stderr del relay: {err[:400]}")
117
118texto = got.decode("utf-8", "replace")
119campos = dict(
120    p.split("=", 1) for p in texto.split("\0") if "=" in p
121)
122print(f"  ✓ llegó al grupo 4: {len(got)} bytes")
123for clave in ("ACTION", "DEVPATH", "SUBSYSTEM", "SEQNUM"):
124    if clave not in campos:
125        fallar(f"el evento reenviado no trae {clave} — libudev-zero lo descartaría (exige las cuatro)")
126print(f"  ✓ trae las cuatro claves que libudev-zero exige: "
127      f"ACTION={campos['ACTION']} SUBSYSTEM={campos['SUBSYSTEM']} DEVPATH={campos['DEVPATH'][:48]}…")
128
129# El mensaje reenviado tiene que ser IDÉNTICO en forma al del kernel: si el relay lo reescribiera,
130# libudev-zero sintetizaría propiedades distintas de las que sintetiza al enumerar, y un
131# dispositivo se comportaría distinto según si estaba enchufado al arrancar o no.
132if not texto.startswith(("add@", "change@", "remove@", "bind@", "unbind@", "move@", "online@", "offline@")):
133    fallar(f"el mensaje reenviado no empieza con la línea de acción del kernel: {texto[:60]!r}")
134print("  ✓ el mensaje va tal cual lo emitió el kernel (misma forma ⇒ mismas propiedades sintetizadas)")
135
136print("\n✓✓ probar-uevent-relay: el hotplug llega a donde libudev-zero lo está esperando")