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 takana_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 } => {
86                // El stub refleja la evidencia declarada como un veredicto "todo pasa" para
87                // ejercitar el camino H1c del cliente (sin lab real que la ejecute).
88                let verdict = (!recipe.evidence.is_empty()).then(|| {
89                    takana_core::proto::EvidenceVerdict {
90                        all_passed: true,
91                        max_level: recipe.evidence.checks.iter().map(|c| c.kind).max(),
92                        checks: recipe
93                            .evidence
94                            .checks
95                            .iter()
96                            .map(|c| takana_core::proto::EvidenceCheckVerdict {
97                                kind: c.kind,
98                                cmd: c.cmd.clone(),
99                                passed: true,
100                                detail: "stub ok".into(),
101                            })
102                            .collect(),
103                    }
104                });
105                Event::BuildReady {
106                    recipe: recipe.name.clone(),
107                    artifact: format!(
108                        "b3:stub-{}",
109                        recipe.commit.as_deref().or(recipe.sha256.as_deref()).unwrap_or("nosrc")
110                    ),
111                    verdict,
112                }
113            }
114            Command::Query { what, .. } => Event::QueryResult {
115                what,
116                value: serde_json::json!({"stub": true}),
117            },
118            Command::Init { cmd } => Event::InitAck { cmd },
119            Command::Inject { artifact, target, .. } => Event::Injected {
120                artifact,
121                target,
122                files: 1,
123            },
124            Command::Hello { .. } => Event::Error {
125                code: "duplicate_hello".into(),
126                msg: "ya hicimos hello".into(),
127            },
128        };
129        let mut g = async_writer.lock().unwrap();
130        g.write_all(serde_json::to_string(&response).unwrap().as_bytes())
131            .unwrap();
132        g.write_all(b"\n").unwrap();
133        g.flush().unwrap();
134        drop(g);
135
136        // Cierra solo si el cliente cierra; aquí seguimos.
137        if !writer.flush().is_ok() {
138            break;
139        }
140    }
141}
142
143#[test]
144fn client_handshake_compile_and_async_modified() {
145    let d = tempfile::tempdir().unwrap();
146    let sock = d.path().join("stub.sock");
147    let async_tx = start_stub_bus(sock.clone());
148
149    // Pequeño wait para que el listener bindee.
150    for _ in 0..50 {
151        if sock.exists() { break }
152        thread::sleep(Duration::from_millis(10));
153    }
154
155    let mut client = takana_agent::AgentClient::connect(&sock).expect("connect");
156    assert_eq!(client.welcome.ver, PROTOCOL_VERSION);
157    assert!(client.welcome.caps.contains(&Cap::Compile));
158
159    let recipe = RecipeInline {
160        name: "grep".into(),
161        repo: Some("git://example/grep.git".into()),
162        commit: Some("abc123".into()),
163        tarball: None,
164        sha256: None,
165        patch: None,
166        compiler: "zig-cc".into(),
167        target: "x86_64-linux-musl".into(),
168        link: "static".into(),
169        flags: vec![],
170        evidence: takana_core::Evidence {
171            checks: vec![takana_core::EvidenceCheck {
172                kind: takana_core::EvidenceKind::CmdExit,
173                cmd: "grep --version".into(),
174                expected_exit: 0,
175                expected_output: None,
176            }],
177        },
178    };
179    let outcome = client
180        .compile(recipe, Duration::from_secs(2))
181        .expect("BuildReady");
182    assert_eq!(outcome.artifact, "b3:stub-abc123");
183    // [H1c] el veredicto de la evidencia llega por BuildReady y el cliente lo expone.
184    let verdict = outcome.verdict.expect("verdict presente (declaramos evidencia)");
185    assert!(verdict.all_passed);
186    assert_eq!(verdict.checks.len(), 1);
187    assert_eq!(verdict.max_level, Some(takana_core::EvidenceKind::CmdExit));
188
189    // El stub también puede empujar eventos asíncronos. Inyectamos un Modified y vemos
190    // que el cliente lo entrega por drain_async()/next_async().
191    async_tx
192        .send(Event::Modified {
193            path: "/bin/grep".into(),
194            op: "replace".into(),
195            ts: "2026-06-09T00:00:00Z".into(),
196        })
197        .unwrap();
198    let ev = client.next_async(Duration::from_secs(2)).expect("async");
199    match ev {
200        Event::Modified { path, .. } => assert_eq!(path, "/bin/grep"),
201        other => panic!("esperaba Modified, llegó {other:?}"),
202    }
203}