sergio/takana

1//! E2E del cliente del bus (Fase 6) contra un servidor stub.
2//!
3//! No depende de `hammerd` ni del lab: un mini-servidor JSON-líneas que conoce el
4//! handshake y responde a `Compile` con `BuildReady` sintético. Demuestra que el cliente
5//! síncrono multiplexa correctamente respuestas (BuildReady) y eventos asíncronos (Modified).
6
7use std::io::{BufRead, BufReader, Write};
8use std::os::unix::net::{UnixListener, UnixStream};
9use std::path::PathBuf;
10use std::thread;
11use std::time::Duration;
12
13use hammer_core::proto::{Cap, Command, Event, Peer, RecipeInline, PROTOCOL_VERSION};
14
15fn start_stub_bus(sock: PathBuf) -> std::sync::mpsc::Sender<Event> {
16    let (tx_async, rx_async) = std::sync::mpsc::channel::<Event>();
17    let _ = std::fs::remove_file(&sock);
18    let listener = UnixListener::bind(&sock).expect("bind stub");
19    thread::Builder::new()
20        .name("stub-bus".into())
21        .spawn(move || {
22            // Sólo aceptamos una conexión: el Receiver no es Clone, se mueve al handler.
23            if let Some(Ok(stream)) = listener.incoming().next() {
24                thread::spawn(move || stub_serve(stream, rx_async));
25            }
26        })
27        .unwrap();
28    tx_async
29}
30
31fn stub_serve(stream: UnixStream, rx_async: std::sync::mpsc::Receiver<Event>) {
32    let reader_stream = stream.try_clone().unwrap();
33    let mut writer = stream;
34    let writer_clone = writer.try_clone().unwrap();
35
36    // Hilo que reenvía los eventos asíncronos al cliente. Mover el writer impide que
37    // se mezclen los bytes con las respuestas (cada Sender los serializa atómicamente).
38    let async_writer = std::sync::Arc::new(std::sync::Mutex::new(writer_clone));
39    let aw_for_async = async_writer.clone();
40    thread::spawn(move || {
41        for ev in rx_async {
42            let line = serde_json::to_string(&ev).unwrap();
43            let mut g = aw_for_async.lock().unwrap();
44            let _ = g.write_all(line.as_bytes());
45            let _ = g.write_all(b"\n");
46            let _ = g.flush();
47        }
48    });
49
50    let mut reader = BufReader::new(reader_stream);
51    let mut first = String::new();
52    reader.read_line(&mut first).unwrap();
53    let cmd: Command = serde_json::from_str(first.trim()).unwrap();
54    let Command::Hello { ver: _, client: _ } = cmd else {
55        return;
56    };
57    // welcome
58    let w = Event::Welcome {
59        ver: PROTOCOL_VERSION,
60        caps: vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init],
61        peer: Peer { uid: 1000, gid: 1000, pid: 1 },
62    };
63    let mut g = async_writer.lock().unwrap();
64    g.write_all(serde_json::to_string(&w).unwrap().as_bytes()).unwrap();
65    g.write_all(b"\n").unwrap();
66    g.flush().unwrap();
67    drop(g);
68
69    // Loop de comandos.
70    loop {
71        let mut line = String::new();
72        match reader.read_line(&mut line) {
73            Ok(0) => break,
74            Ok(_) => {}
75            Err(_) => break,
76        }
77        if line.trim().is_empty() {
78            continue;
79        }
80        let cmd: Command = match serde_json::from_str(line.trim()) {
81            Ok(c) => c,
82            Err(_) => continue,
83        };
84        let response = match cmd {
85            Command::Compile { recipe } => Event::BuildReady {
86                recipe: recipe.name.clone(),
87                artifact: format!(
88                    "b3:stub-{}",
89                    recipe.commit.as_deref().or(recipe.sha256.as_deref()).unwrap_or("nosrc")
90                ),
91            },
92            Command::Query { what, .. } => Event::QueryResult {
93                what,
94                value: serde_json::json!({"stub": true}),
95            },
96            Command::Init { cmd } => Event::InitAck { cmd },
97            Command::Inject { artifact, target, .. } => Event::Injected {
98                artifact,
99                target,
100                files: 1,
101            },
102            Command::Hello { .. } => Event::Error {
103                code: "duplicate_hello".into(),
104                msg: "ya hicimos hello".into(),
105            },
106        };
107        let mut g = async_writer.lock().unwrap();
108        g.write_all(serde_json::to_string(&response).unwrap().as_bytes())
109            .unwrap();
110        g.write_all(b"\n").unwrap();
111        g.flush().unwrap();
112        drop(g);
113
114        // Cierra solo si el cliente cierra; aquí seguimos.
115        if !writer.flush().is_ok() {
116            break;
117        }
118    }
119}
120
121#[test]
122fn client_handshake_compile_and_async_modified() {
123    let d = tempfile::tempdir().unwrap();
124    let sock = d.path().join("stub.sock");
125    let async_tx = start_stub_bus(sock.clone());
126
127    // Pequeño wait para que el listener bindee.
128    for _ in 0..50 {
129        if sock.exists() { break }
130        thread::sleep(Duration::from_millis(10));
131    }
132
133    let mut client = hammer_agent::AgentClient::connect(&sock).expect("connect");
134    assert_eq!(client.welcome.ver, PROTOCOL_VERSION);
135    assert!(client.welcome.caps.contains(&Cap::Compile));
136
137    let recipe = RecipeInline {
138        name: "grep".into(),
139        repo: Some("git://example/grep.git".into()),
140        commit: Some("abc123".into()),
141        tarball: None,
142        sha256: None,
143        patch: None,
144        compiler: "zig-cc".into(),
145        target: "x86_64-linux-musl".into(),
146        link: "static".into(),
147        flags: vec![],
148    };
149    let artifact = client
150        .compile(recipe, Duration::from_secs(2))
151        .expect("BuildReady");
152    assert_eq!(artifact, "b3:stub-abc123");
153
154    // El stub también puede empujar eventos asíncronos. Inyectamos un Modified y vemos
155    // que el cliente lo entrega por drain_async()/next_async().
156    async_tx
157        .send(Event::Modified {
158            path: "/bin/grep".into(),
159            op: "replace".into(),
160            ts: "2026-06-09T00:00:00Z".into(),
161        })
162        .unwrap();
163    let ev = client.next_async(Duration::from_secs(2)).expect("async");
164    match ev {
165        Event::Modified { path, .. } => assert_eq!(path, "/bin/grep"),
166        other => panic!("esperaba Modified, llegó {other:?}"),
167    }
168}