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            },
151        },
152    );
153    match recv_event(&mut reader) {
154        Event::Error { code, msg } => {
155            assert_eq!(code, "no_cap");
156            assert!(msg.contains("Compile"), "{msg}");
157        }
158        other => panic!("esperaba Error{{no_cap}}, llegó {other:?}"),
159    }
160}
161
162#[test]
163fn query_file_returns_value_for_existing_path() {
164    let d = tempfile::tempdir().unwrap();
165    let sock = d.path().join("agent.sock");
166    start_bus(
167        &sock,
168        permissive_policy(),
169        d.path().join("store"),
170        d.path().join("init.ctl"),
171    );
172
173    let target = d.path().join("data.txt");
174    std::fs::write(&target, b"contenido").unwrap();
175
176    let stream = wait_for_sock(&sock);
177    let mut writer = stream.try_clone().unwrap();
178    let mut reader = BufReader::new(stream);
179    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
180    let _ = recv_event(&mut reader); // welcome
181
182    send(
183        &mut writer,
184        &Command::Query {
185            what: "file".into(),
186            path: Some(target.display().to_string()),
187            name: None,
188            expr: None,
189        },
190    );
191    match recv_event(&mut reader) {
192        Event::QueryResult { what, value } => {
193            assert_eq!(what, "file");
194            assert_eq!(value["exists"], serde_json::json!(true));
195            assert_eq!(value["size"], serde_json::json!(9));
196        }
197        other => panic!("esperaba QueryResult, llegó {other:?}"),
198    }
199}
200
201#[test]
202fn query_expr_evaluates_against_host_path() {
203    let d = tempfile::tempdir().unwrap();
204    let sock = d.path().join("agent.sock");
205    start_bus(
206        &sock,
207        permissive_policy(),
208        d.path().join("store"),
209        d.path().join("init.ctl"),
210    );
211
212    // file:<path> sobre un archivo real; el daemon evalúa con fs_root=None ⇒ ruta absoluta.
213    let target = d.path().join("payload.bin");
214    std::fs::write(&target, b"X").unwrap();
215
216    let stream = wait_for_sock(&sock);
217    let mut writer = stream.try_clone().unwrap();
218    let mut reader = BufReader::new(stream);
219    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
220    let _ = recv_event(&mut reader); // welcome
221
222    send(
223        &mut writer,
224        &Command::Query {
225            what: "expr".into(),
226            path: None,
227            name: None,
228            expr: Some(format!("file:{}", target.display())),
229        },
230    );
231    match recv_event(&mut reader) {
232        Event::QueryResult { what, value } => {
233            assert_eq!(what, "expr");
234            assert_eq!(value["found"], serde_json::json!(true));
235            assert_eq!(value["size"], serde_json::json!(1));
236        }
237        other => panic!("esperaba QueryResult, llegó {other:?}"),
238    }
239
240    // Y una expresión inválida vuelve como JSON con `error`, no como Event::Error
241    // (la query es bien formada; el problema es el contenido).
242    send(
243        &mut writer,
244        &Command::Query {
245            what: "expr".into(),
246            path: None,
247            name: None,
248            expr: Some("foo:bar".into()),
249        },
250    );
251    match recv_event(&mut reader) {
252        Event::QueryResult { what, value } => {
253            assert_eq!(what, "expr");
254            assert!(value["error"].as_str().unwrap().contains("kind 'foo'"));
255        }
256        other => panic!("esperaba QueryResult, llegó {other:?}"),
257    }
258}
259
260#[test]
261fn modified_events_fan_out_to_subscribers() {
262    // Validamos sólo el camino del fan-out a través del bus, sin watcher real:
263    // disparamos `publish(Modified)` directamente y vemos que llega por el socket.
264    let d = tempfile::tempdir().unwrap();
265    let sock = d.path().join("agent.sock");
266    let bus_handle = start_bus(
267        &sock,
268        permissive_policy(),
269        d.path().join("store"),
270        d.path().join("init.ctl"),
271    );
272
273    let stream = wait_for_sock(&sock);
274    let mut writer = stream.try_clone().unwrap();
275    let mut reader = BufReader::new(stream);
276    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
277    let _ = recv_event(&mut reader); // welcome
278
279    // El forwarder de eventos arranca dentro de handle_connection; damos un margen para
280    // que la suscripción esté registrada antes de publicar.
281    std::thread::sleep(Duration::from_millis(50));
282    bus_handle.publish(&Event::Modified {
283        path: "/bin/grep".into(),
284        op: "replace".into(),
285        ts: "2026-06-09T00:00:00Z".into(),
286    });
287    match recv_event(&mut reader) {
288        Event::Modified { path, op, ts } => {
289            assert_eq!(path, "/bin/grep");
290            assert_eq!(op, "replace");
291            assert_eq!(ts, "2026-06-09T00:00:00Z");
292        }
293        other => panic!("esperaba Modified, llegó {other:?}"),
294    }
295}
296
297#[test]
298fn init_command_writes_to_fifo() {
299    let d = tempfile::tempdir().unwrap();
300    let sock = d.path().join("agent.sock");
301    let fifo = d.path().join("init.ctl");
302    control::ensure_fifo(&fifo).unwrap();
303
304    // Reader del FIFO en background: lee una línea y termina.
305    let fifo_for_reader = fifo.clone();
306    let reader_thread = std::thread::spawn(move || {
307        let f = std::fs::OpenOptions::new()
308            .read(true)
309            .open(&fifo_for_reader)
310            .unwrap();
311        let mut r = std::io::BufReader::new(f);
312        let mut s = String::new();
313        std::io::BufRead::read_line(&mut r, &mut s).unwrap();
314        s
315    });
316    std::thread::sleep(Duration::from_millis(50));
317
318    start_bus(
319        &sock,
320        permissive_policy(),
321        d.path().join("store"),
322        fifo.clone(),
323    );
324    let stream = wait_for_sock(&sock);
325    let mut writer = stream.try_clone().unwrap();
326    let mut reader = BufReader::new(stream);
327    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
328    let _ = recv_event(&mut reader);
329
330    send(&mut writer, &Command::Init { cmd: "start web".into() });
331    match recv_event(&mut reader) {
332        Event::InitAck { cmd } => assert_eq!(cmd, "start web"),
333        other => panic!("esperaba InitAck, llegó {other:?}"),
334    }
335    let from_fifo = reader_thread.join().unwrap();
336    assert_eq!(from_fifo, "start web\n");
337}