sergio/takana

1//! E2E del bus de agente (Fase 5):
2//!
3//! Arrancamos `serve_agent_bus` en un thread con un socket bajo tempdir, conectamos como
4//! cliente, completamos el handshake y ejercitamos los caminos no-pesados (`query`, errores).
5//!
6//! `compile` real no se ejercita aquí — depende del lab + red, ya cubierto por el e2e de
7//! grep gated en `HAMMER_NETWORK_TESTS`. Lo que sí verificamos es que un `compile` SIN la
8//! capacidad devuelve `Error{code:"no_cap"}`, lo que prueba el dispatcher + el gate.
9
10// El binario no expone una API pública; importamos sus módulos privados desde
11// `path = ...`. `proto` se movió a hammer-core en Fase 6.
12#[path = "../src/events.rs"]
13mod events;
14#[path = "../src/control.rs"]
15mod control;
16#[path = "../src/bus.rs"]
17mod bus;
18
19use std::io::{BufRead, BufReader, Write};
20use std::os::unix::net::UnixStream;
21use std::path::PathBuf;
22use std::sync::Arc;
23use std::thread;
24use std::time::Duration;
25
26use hammer_core::proto::{Cap, Command, Event, Peer, RecipeInline};
27
28/// Helper: lanza un bus en background y devuelve la ruta del socket. Limpia al final del
29/// test gracias al `tempfile::TempDir` que se le pasa al caller.
30fn start_bus(
31    sock: &std::path::Path,
32    policy: bus::CapsPolicy,
33    store_root: PathBuf,
34    init_control: PathBuf,
35) -> events::EventBus {
36    let events = events::EventBus::new();
37    let ctx = bus::BusContext {
38        store_root,
39        init_control,
40        events: events.clone(),
41    };
42    let sock = sock.to_path_buf();
43    thread::Builder::new()
44        .name("bus-test".into())
45        .spawn(move || {
46            let _ = bus::serve_agent_bus(&sock, policy, ctx);
47        })
48        .unwrap();
49    // El caller usa `wait_for_sock` para esperar a que `bind` exista en disco.
50    events
51}
52
53fn wait_for_sock(sock: &std::path::Path) -> UnixStream {
54    for _ in 0..200 {
55        if let Ok(s) = UnixStream::connect(sock) {
56            return s;
57        }
58        std::thread::sleep(Duration::from_millis(10));
59    }
60    panic!("no pude conectar al bus en {}", sock.display());
61}
62
63fn send(stream: &mut UnixStream, cmd: &Command) {
64    let line = serde_json::to_string(cmd).unwrap();
65    stream.write_all(line.as_bytes()).unwrap();
66    stream.write_all(b"\n").unwrap();
67    stream.flush().unwrap();
68}
69
70fn recv_event(reader: &mut BufReader<UnixStream>) -> Event {
71    let mut line = String::new();
72    reader.read_line(&mut line).expect("read event");
73    serde_json::from_str(line.trim()).expect("event JSON válido")
74}
75
76fn permissive_policy() -> bus::CapsPolicy {
77    Arc::new(|_peer: &Peer| {
78        vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init]
79    })
80}
81
82fn read_only_policy() -> bus::CapsPolicy {
83    Arc::new(|_peer: &Peer| vec![Cap::Query])
84}
85
86#[test]
87fn handshake_returns_welcome_with_peer_creds() {
88    let d = tempfile::tempdir().unwrap();
89    let sock = d.path().join("agent.sock");
90    start_bus(
91        &sock,
92        permissive_policy(),
93        d.path().join("store"),
94        d.path().join("init.ctl"),
95    );
96
97    let stream = wait_for_sock(&sock);
98    let mut writer = stream.try_clone().unwrap();
99    let mut reader = BufReader::new(stream);
100    send(
101        &mut writer,
102        &Command::Hello { ver: 1, client: "test".into() },
103    );
104    let ev = recv_event(&mut reader);
105    match ev {
106        Event::Welcome { ver, caps, peer } => {
107            assert_eq!(ver, 1);
108            assert!(caps.contains(&Cap::Query));
109            assert_eq!(peer.uid, unsafe { libc::getuid() });
110            assert_eq!(peer.pid, std::process::id() as i32);
111        }
112        other => panic!("esperaba Welcome, llegó {other:?}"),
113    }
114}
115
116#[test]
117fn missing_cap_returns_error_no_cap() {
118    let d = tempfile::tempdir().unwrap();
119    let sock = d.path().join("agent.sock");
120    start_bus(
121        &sock,
122        read_only_policy(),
123        d.path().join("store"),
124        d.path().join("init.ctl"),
125    );
126
127    let stream = wait_for_sock(&sock);
128    let mut writer = stream.try_clone().unwrap();
129    let mut reader = BufReader::new(stream);
130    send(
131        &mut writer,
132        &Command::Hello { ver: 1, client: "test".into() },
133    );
134    let _ = recv_event(&mut reader); // welcome
135
136    send(
137        &mut writer,
138        &Command::Compile {
139            recipe: RecipeInline {
140                name: "x".into(),
141                repo: Some("git://x".into()),
142                commit: Some("abc".into()),
143                tarball: None,
144                sha256: None,
145                patch: None,
146                compiler: "zig-cc".into(),
147                target: "x86_64-linux-musl".into(),
148                link: "static".into(),
149                flags: vec![],
150                evidence: Default::default(),
151            },
152        },
153    );
154    match recv_event(&mut reader) {
155        Event::Error { code, msg } => {
156            assert_eq!(code, "no_cap");
157            assert!(msg.contains("Compile"), "{msg}");
158        }
159        other => panic!("esperaba Error{{no_cap}}, llegó {other:?}"),
160    }
161}
162
163#[test]
164fn query_file_returns_value_for_existing_path() {
165    let d = tempfile::tempdir().unwrap();
166    let sock = d.path().join("agent.sock");
167    start_bus(
168        &sock,
169        permissive_policy(),
170        d.path().join("store"),
171        d.path().join("init.ctl"),
172    );
173
174    let target = d.path().join("data.txt");
175    std::fs::write(&target, b"contenido").unwrap();
176
177    let stream = wait_for_sock(&sock);
178    let mut writer = stream.try_clone().unwrap();
179    let mut reader = BufReader::new(stream);
180    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
181    let _ = recv_event(&mut reader); // welcome
182
183    send(
184        &mut writer,
185        &Command::Query {
186            what: "file".into(),
187            path: Some(target.display().to_string()),
188            name: None,
189            expr: None,
190        },
191    );
192    match recv_event(&mut reader) {
193        Event::QueryResult { what, value } => {
194            assert_eq!(what, "file");
195            assert_eq!(value["exists"], serde_json::json!(true));
196            assert_eq!(value["size"], serde_json::json!(9));
197        }
198        other => panic!("esperaba QueryResult, llegó {other:?}"),
199    }
200}
201
202#[test]
203fn query_expr_evaluates_against_host_path() {
204    let d = tempfile::tempdir().unwrap();
205    let sock = d.path().join("agent.sock");
206    start_bus(
207        &sock,
208        permissive_policy(),
209        d.path().join("store"),
210        d.path().join("init.ctl"),
211    );
212
213    // file:<path> sobre un archivo real; el daemon evalúa con fs_root=None ⇒ ruta absoluta.
214    let target = d.path().join("payload.bin");
215    std::fs::write(&target, b"X").unwrap();
216
217    let stream = wait_for_sock(&sock);
218    let mut writer = stream.try_clone().unwrap();
219    let mut reader = BufReader::new(stream);
220    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
221    let _ = recv_event(&mut reader); // welcome
222
223    send(
224        &mut writer,
225        &Command::Query {
226            what: "expr".into(),
227            path: None,
228            name: None,
229            expr: Some(format!("file:{}", target.display())),
230        },
231    );
232    match recv_event(&mut reader) {
233        Event::QueryResult { what, value } => {
234            assert_eq!(what, "expr");
235            assert_eq!(value["found"], serde_json::json!(true));
236            assert_eq!(value["size"], serde_json::json!(1));
237        }
238        other => panic!("esperaba QueryResult, llegó {other:?}"),
239    }
240
241    // Y una expresión inválida vuelve como JSON con `error`, no como Event::Error
242    // (la query es bien formada; el problema es el contenido).
243    send(
244        &mut writer,
245        &Command::Query {
246            what: "expr".into(),
247            path: None,
248            name: None,
249            expr: Some("foo:bar".into()),
250        },
251    );
252    match recv_event(&mut reader) {
253        Event::QueryResult { what, value } => {
254            assert_eq!(what, "expr");
255            assert!(value["error"].as_str().unwrap().contains("kind 'foo'"));
256        }
257        other => panic!("esperaba QueryResult, llegó {other:?}"),
258    }
259}
260
261#[test]
262fn modified_events_fan_out_to_subscribers() {
263    // Validamos sólo el camino del fan-out a través del bus, sin watcher real:
264    // disparamos `publish(Modified)` directamente y vemos que llega por el socket.
265    let d = tempfile::tempdir().unwrap();
266    let sock = d.path().join("agent.sock");
267    let bus_handle = start_bus(
268        &sock,
269        permissive_policy(),
270        d.path().join("store"),
271        d.path().join("init.ctl"),
272    );
273
274    let stream = wait_for_sock(&sock);
275    let mut writer = stream.try_clone().unwrap();
276    let mut reader = BufReader::new(stream);
277    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
278    let _ = recv_event(&mut reader); // welcome
279
280    // El forwarder de eventos arranca dentro de handle_connection; damos un margen para
281    // que la suscripción esté registrada antes de publicar.
282    std::thread::sleep(Duration::from_millis(50));
283    bus_handle.publish(&Event::Modified {
284        path: "/bin/grep".into(),
285        op: "replace".into(),
286        ts: "2026-06-09T00:00:00Z".into(),
287    });
288    match recv_event(&mut reader) {
289        Event::Modified { path, op, ts } => {
290            assert_eq!(path, "/bin/grep");
291            assert_eq!(op, "replace");
292            assert_eq!(ts, "2026-06-09T00:00:00Z");
293        }
294        other => panic!("esperaba Modified, llegó {other:?}"),
295    }
296}
297
298#[test]
299fn init_command_writes_to_fifo() {
300    let d = tempfile::tempdir().unwrap();
301    let sock = d.path().join("agent.sock");
302    let fifo = d.path().join("init.ctl");
303    control::ensure_fifo(&fifo).unwrap();
304
305    // Reader del FIFO en background: lee una línea y termina.
306    let fifo_for_reader = fifo.clone();
307    let reader_thread = std::thread::spawn(move || {
308        let f = std::fs::OpenOptions::new()
309            .read(true)
310            .open(&fifo_for_reader)
311            .unwrap();
312        let mut r = std::io::BufReader::new(f);
313        let mut s = String::new();
314        std::io::BufRead::read_line(&mut r, &mut s).unwrap();
315        s
316    });
317    std::thread::sleep(Duration::from_millis(50));
318
319    start_bus(
320        &sock,
321        permissive_policy(),
322        d.path().join("store"),
323        fifo.clone(),
324    );
325    let stream = wait_for_sock(&sock);
326    let mut writer = stream.try_clone().unwrap();
327    let mut reader = BufReader::new(stream);
328    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
329    let _ = recv_event(&mut reader);
330
331    send(&mut writer, &Command::Init { cmd: "start web".into() });
332    match recv_event(&mut reader) {
333        Event::InitAck { cmd } => assert_eq!(cmd, "start web"),
334        other => panic!("esperaba InitAck, llegó {other:?}"),
335    }
336    let from_fifo = reader_thread.join().unwrap();
337    assert_eq!(from_fifo, "start web\n");
338}