sergio/takana

1//! E2E del adaptador de transporte arje-bus → bus de agente (último tramo de B.2).
2//!
3//! No necesita un init vivo: levantamos un **socket falso de arje-bus** que habla el mismo
4//! wire (frames `u32-BE-len + postcard`), validamos que el adaptador manda el `Subscribe`
5//! correcto, le inyectamos un `EnteCrashed`, y verificamos que sale como `Event::Crashed`
6//! por el `EventBus`. Así se ejercita la ruta real de `arje_link::run`: connect → subscribe →
7//! leer frame → traducir → publicar.
8//!
9//! El contrato de bytes con arje-bus (que acá replicamos como `Tx*`) está fijado del lado de
10//! arje por `arje-bus::contrato_wire` — si arje reordena un enum, ese test falla primero.
11
12// Importamos los módulos privados del binario (mismo patrón que `bus_e2e.rs`). El orden
13// importa: `arje_link` y `crashes` referencian `crate::events`.
14#[path = "../src/events.rs"]
15mod events;
16#[path = "../src/crashes.rs"]
17mod crashes;
18#[path = "../src/arje_link.rs"]
19mod arje_link;
20
21use std::io::{Read, Write};
22use std::os::unix::net::UnixListener;
23use std::path::PathBuf;
24use std::thread;
25use std::time::Duration;
26
27use takana_core::proto::Event;
28use serde::Serialize;
29use ulid::Ulid;
30
31// ── Lado emisor del wire (lo que arje-bus serializaría). Discriminantes fijados por
32//    `arje-bus::contrato_wire`: BusPayload Event=2, BusEvent EnteCrashed=0, LifecycleStatus
33//    Killed=1. ─────────────────────────────────────────────────────────────────────────────
34#[derive(Serialize)]
35struct TxMsg {
36    from: Option<Ulid>,
37    seq: u64,
38    payload: TxPayload,
39}
40
41#[derive(Serialize)]
42enum TxPayload {
43    #[allow(dead_code)]
44    Request,
45    #[allow(dead_code)]
46    Response,
47    Event(TxEvent),
48}
49
50#[derive(Serialize)]
51enum TxEvent {
52    EnteCrashed { id: Ulid, label: String, status: TxStatus },
53}
54
55#[derive(Serialize)]
56enum TxStatus {
57    #[allow(dead_code)]
58    Exited(i32),
59    Killed(i32),
60}
61
62fn frame(body: &[u8]) -> Vec<u8> {
63    let mut v = (body.len() as u32).to_be_bytes().to_vec();
64    v.extend_from_slice(body);
65    v
66}
67
68#[test]
69fn crashed_de_un_bus_falso_llega_como_event_crashed() {
70    let tmp = tempfile::tempdir().unwrap();
71    let sock = tmp.path().join("ente-bus.sock");
72    let listener = UnixListener::bind(&sock).unwrap();
73
74    // Servidor falso de arje-bus.
75    let server = thread::spawn(move || {
76        let (mut conn, _) = listener.accept().unwrap();
77
78        // 1) El adaptador debe mandar el frame Subscribe exacto: [None, seq=1, Request, Subscribe=13].
79        let mut len = [0u8; 4];
80        conn.read_exact(&mut len).unwrap();
81        let n = u32::from_be_bytes(len) as usize;
82        let mut body = vec![0u8; n];
83        conn.read_exact(&mut body).unwrap();
84        assert_eq!(body, vec![0x00, 0x01, 0x00, 0x0E], "frame Subscribe inesperado");
85
86        // 2) Le inyectamos un EnteCrashed por señal (SIGSEGV=11 → 128+11=139).
87        let id = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").unwrap();
88        let ev = postcard::to_stdvec(&TxMsg {
89            from: Some(id),
90            seq: 0,
91            payload: TxPayload::Event(TxEvent::EnteCrashed {
92                id,
93                label: "demonio".into(),
94                status: TxStatus::Killed(11),
95            }),
96        })
97        .unwrap();
98        conn.write_all(&frame(&ev)).unwrap();
99        conn.flush().unwrap();
100        // Al salir del closure se cierra `conn` → el adaptador sale por EOF.
101    });
102
103    let bus = events::EventBus::new();
104    let rx = bus.subscribe();
105    let bus_for_run = bus.clone();
106    let sock_for_run = PathBuf::from(&sock);
107    let client = thread::spawn(move || {
108        // Devuelve Err por el EOF final del servidor; es el cierre esperado.
109        let _ = arje_link::run(&sock_for_run, &bus_for_run);
110    });
111
112    // El crash debe materializarse como Event::Crashed en el bus de agente.
113    let ev = rx
114        .recv_timeout(Duration::from_secs(5))
115        .expect("el adaptador no publicó el crash");
116    match ev {
117        Event::Crashed { service, code } => {
118            assert_eq!(service, "demonio");
119            assert_eq!(code, 139, "Killed(11) → 128+11");
120        }
121        other => panic!("esperaba Event::Crashed, fue {other:?}"),
122    }
123
124    server.join().unwrap();
125    client.join().unwrap();
126}