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: "git://x".into(),
142                commit: "abc".into(),
143                patch: None,
144                compiler: "zig-cc".into(),
145                target: "x86_64-linux-musl".into(),
146                link: "static".into(),
147                flags: vec![],
148            },
149        },
150    );
151    match recv_event(&mut reader) {
152        Event::Error { code, msg } => {
153            assert_eq!(code, "no_cap");
154            assert!(msg.contains("Compile"), "{msg}");
155        }
156        other => panic!("esperaba Error{{no_cap}}, llegó {other:?}"),
157    }
158}
159
160#[test]
161fn query_file_returns_value_for_existing_path() {
162    let d = tempfile::tempdir().unwrap();
163    let sock = d.path().join("agent.sock");
164    start_bus(
165        &sock,
166        permissive_policy(),
167        d.path().join("store"),
168        d.path().join("init.ctl"),
169    );
170
171    let target = d.path().join("data.txt");
172    std::fs::write(&target, b"contenido").unwrap();
173
174    let stream = wait_for_sock(&sock);
175    let mut writer = stream.try_clone().unwrap();
176    let mut reader = BufReader::new(stream);
177    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
178    let _ = recv_event(&mut reader); // welcome
179
180    send(
181        &mut writer,
182        &Command::Query {
183            what: "file".into(),
184            path: Some(target.display().to_string()),
185            name: None,
186        },
187    );
188    match recv_event(&mut reader) {
189        Event::QueryResult { what, value } => {
190            assert_eq!(what, "file");
191            assert_eq!(value["exists"], serde_json::json!(true));
192            assert_eq!(value["size"], serde_json::json!(9));
193        }
194        other => panic!("esperaba QueryResult, llegó {other:?}"),
195    }
196}
197
198#[test]
199fn modified_events_fan_out_to_subscribers() {
200    // Validamos sólo el camino del fan-out a través del bus, sin watcher real:
201    // disparamos `publish(Modified)` directamente y vemos que llega por el socket.
202    let d = tempfile::tempdir().unwrap();
203    let sock = d.path().join("agent.sock");
204    let bus_handle = start_bus(
205        &sock,
206        permissive_policy(),
207        d.path().join("store"),
208        d.path().join("init.ctl"),
209    );
210
211    let stream = wait_for_sock(&sock);
212    let mut writer = stream.try_clone().unwrap();
213    let mut reader = BufReader::new(stream);
214    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
215    let _ = recv_event(&mut reader); // welcome
216
217    // El forwarder de eventos arranca dentro de handle_connection; damos un margen para
218    // que la suscripción esté registrada antes de publicar.
219    std::thread::sleep(Duration::from_millis(50));
220    bus_handle.publish(&Event::Modified {
221        path: "/bin/grep".into(),
222        op: "replace".into(),
223        ts: "2026-06-09T00:00:00Z".into(),
224    });
225    match recv_event(&mut reader) {
226        Event::Modified { path, op, ts } => {
227            assert_eq!(path, "/bin/grep");
228            assert_eq!(op, "replace");
229            assert_eq!(ts, "2026-06-09T00:00:00Z");
230        }
231        other => panic!("esperaba Modified, llegó {other:?}"),
232    }
233}
234
235#[test]
236fn init_command_writes_to_fifo() {
237    let d = tempfile::tempdir().unwrap();
238    let sock = d.path().join("agent.sock");
239    let fifo = d.path().join("init.ctl");
240    control::ensure_fifo(&fifo).unwrap();
241
242    // Reader del FIFO en background: lee una línea y termina.
243    let fifo_for_reader = fifo.clone();
244    let reader_thread = std::thread::spawn(move || {
245        let f = std::fs::OpenOptions::new()
246            .read(true)
247            .open(&fifo_for_reader)
248            .unwrap();
249        let mut r = std::io::BufReader::new(f);
250        let mut s = String::new();
251        std::io::BufRead::read_line(&mut r, &mut s).unwrap();
252        s
253    });
254    std::thread::sleep(Duration::from_millis(50));
255
256    start_bus(
257        &sock,
258        permissive_policy(),
259        d.path().join("store"),
260        fifo.clone(),
261    );
262    let stream = wait_for_sock(&sock);
263    let mut writer = stream.try_clone().unwrap();
264    let mut reader = BufReader::new(stream);
265    send(&mut writer, &Command::Hello { ver: 1, client: "t".into() });
266    let _ = recv_event(&mut reader);
267
268    send(&mut writer, &Command::Init { cmd: "start web".into() });
269    match recv_event(&mut reader) {
270        Event::InitAck { cmd } => assert_eq!(cmd, "start web"),
271        other => panic!("esperaba InitAck, llegó {other:?}"),
272    }
273    let from_fifo = reader_thread.join().unwrap();
274    assert_eq!(from_fifo, "start web\n");
275}