tawasuyu/brahman

1//! End-to-end tests for `nakui run` — bind a socket from the main test
2//! thread, drive it from a client thread with line-JSON requests, and
3//! assert behaviour through the wire.
4//!
5//! Why server-on-main / client-on-thread: `Executor` is `!Send` (Rhai
6//! caches AST in a `RefCell`). Moving it across thread boundaries is a
7//! compile-time error, so the test thread runs the server and a worker
8//! thread plays the client. The worker calls `shutdown` last, which lets
9//! the main thread return from `run_server` and join.
10
11use std::io::{BufRead, BufReader, Write};
12use std::os::unix::net::UnixStream;
13use std::path::{Path, PathBuf};
14use std::thread;
15use std::time::Duration;
16
17use nakui_core::event_log::{EventLog, execute_and_log, seed_and_log};
18use nakui_core::executor::Executor;
19use nakui_core::run::run_server;
20use nakui_core::store::MemoryStore;
21use serde_json::{Value, json};
22use uuid::Uuid;
23
24fn workspace_root() -> PathBuf {
25    Path::new(env!("CARGO_MANIFEST_DIR"))
26        .parent()
27        .expect("workspace root above core/")
28        .to_path_buf()
29}
30
31fn treasury_module() -> PathBuf {
32    workspace_root().join("modules/treasury")
33}
34
35fn fresh_log_path() -> PathBuf {
36    std::env::temp_dir().join(format!("nakui_run_log_{}.jsonl", Uuid::new_v4()))
37}
38
39fn fresh_socket_path() -> PathBuf {
40    std::env::temp_dir().join(format!("nakui_run_{}.sock", Uuid::new_v4()))
41}
42
43/// One client connection: keeps a single BufReader alive across
44/// exchanges so buffered bytes from one response don't get dropped
45/// before the next read.
46struct Conn {
47    writer: UnixStream,
48    reader: BufReader<UnixStream>,
49}
50
51impl Conn {
52    fn connect_with_retry(path: &Path) -> Self {
53        for _ in 0..100 {
54            if let Ok(stream) = UnixStream::connect(path) {
55                let reader_stream = stream.try_clone().expect("clone for reader");
56                return Self {
57                    writer: stream,
58                    reader: BufReader::new(reader_stream),
59                };
60            }
61            thread::sleep(Duration::from_millis(20));
62        }
63        panic!("server never started accepting on {}", path.display());
64    }
65
66    fn exchange(&mut self, req: Value) -> Value {
67        let mut s = serde_json::to_vec(&req).expect("serialize request");
68        s.push(b'\n');
69        self.writer.write_all(&s).expect("write request");
70        let mut line = String::new();
71        let n = self.reader.read_line(&mut line).expect("read response");
72        assert!(n > 0, "server closed connection without responding");
73        serde_json::from_str(line.trim()).expect("parse response")
74    }
75}
76
77#[test]
78fn run_server_full_protocol_round_trip() {
79    let log_path = fresh_log_path();
80    let socket_path = fresh_socket_path();
81
82    let caja_id = Uuid::new_v4();
83    {
84        let executor = Executor::load_module(treasury_module()).expect("load module");
85        let mut log = EventLog::open(&log_path).expect("open log");
86        let mut store = MemoryStore::new();
87        seed_and_log(
88            &executor,
89            &mut store,
90            &mut log,
91            "Caja",
92            caja_id,
93            json!({
94                "id": caja_id.to_string(),
95                "name": "A",
96                "saldo": 100_000_i64,
97                "currency": "USD",
98            }),
99        )
100        .expect("seed");
101    }
102
103    let executor = Executor::load_module(treasury_module()).expect("load module");
104    let log = EventLog::open(&log_path).expect("reopen log");
105    let store = MemoryStore::new();
106
107    let socket_for_client = socket_path.clone();
108    let client = thread::spawn(move || -> Result<(), String> {
109        let mut conn = Conn::connect_with_retry(&socket_for_client);
110
111        let resp = conn.exchange(json!({"op": "describe"}));
112        if resp["ok"] != json!(true) {
113            return Err(format!("describe not ok: {}", resp));
114        }
115        if resp["module"] != json!("treasury") {
116            return Err(format!("module mismatch: {}", resp));
117        }
118        if resp["protocol"] != json!(1) {
119            return Err(format!("protocol mismatch: {}", resp));
120        }
121        let morphisms = resp["morphisms"]
122            .as_array()
123            .ok_or("morphisms not array")?;
124        if !morphisms.iter().any(|m| m["name"] == "register_cash_move") {
125            return Err("register_cash_move missing from describe".into());
126        }
127
128        let resp = conn.exchange(json!({
129            "op": "load",
130            "entity": "Caja",
131            "id": caja_id.to_string(),
132        }));
133        if resp["value"]["saldo"].as_i64() != Some(100_000) {
134            return Err(format!("initial saldo wrong: {}", resp));
135        }
136
137        let resp = conn.exchange(json!({
138            "op": "execute",
139            "morphism": "register_cash_move",
140            "inputs": {"caja": caja_id.to_string()},
141            "params": {
142                "monto": 5_000_i64,
143                "tipo": "in",
144                "timestamp": "2026-05-04T10:00:00Z",
145                "memo": "via run",
146                "movimiento_id": Uuid::new_v4().to_string(),
147            },
148        }));
149        if resp["ok"] != json!(true) {
150            return Err(format!("execute failed: {}", resp));
151        }
152        if resp["seq"].as_u64().is_none() {
153            return Err(format!("execute missing seq: {}", resp));
154        }
155        if resp["ops"].as_array().map(|a| a.is_empty()).unwrap_or(true) {
156            return Err(format!("execute missing ops: {}", resp));
157        }
158
159        let resp = conn.exchange(json!({
160            "op": "load",
161            "entity": "Caja",
162            "id": caja_id.to_string(),
163        }));
164        if resp["value"]["saldo"].as_i64() != Some(105_000) {
165            return Err(format!("post-execute saldo wrong: {}", resp));
166        }
167
168        // Kernel rejection: returns ok=false with stage=pre_log.
169        let other = Uuid::new_v4();
170        let resp = conn.exchange(json!({
171            "op": "execute",
172            "morphism": "transfer_between_cajas",
173            "inputs": {"source": caja_id.to_string(), "dest": other.to_string()},
174            "params": {
175                "monto": 999_999_999_i64,
176                "timestamp": "2026-05-04T10:30:00Z",
177                "memo": "overdraw",
178                "transfer_id": Uuid::new_v4().to_string(),
179            },
180        }));
181        if resp["ok"] != json!(false) || resp["stage"] != json!("pre_log") {
182            return Err(format!("expected pre_log rejection: {}", resp));
183        }
184
185        // Bad JSON — connection survives, server keeps serving.
186        conn.writer.write_all(b"not json\n").map_err(|e| e.to_string())?;
187        let mut line = String::new();
188        conn.reader.read_line(&mut line).map_err(|e| e.to_string())?;
189        let parsed: Value = serde_json::from_str(line.trim()).map_err(|e| e.to_string())?;
190        if parsed["ok"] != json!(false) {
191            return Err(format!("bad request didn't get error: {}", parsed));
192        }
193
194        let resp = conn.exchange(json!({"op": "verify"}));
195        if resp["ok"] != json!(true) {
196            return Err(format!("verify failed: {}", resp));
197        }
198        if resp["entries"].as_u64() != Some(2) {
199            return Err(format!("verify entries wrong: {}", resp));
200        }
201
202        let resp = conn.exchange(json!({"op": "shutdown"}));
203        if resp["ok"] != json!(true) || resp["shutdown"] != json!(true) {
204            return Err(format!("shutdown response wrong: {}", resp));
205        }
206        Ok(())
207    });
208
209    run_server(executor, log, store, None, &socket_path).expect("server clean exit");
210    client.join().expect("client thread joined").expect("client assertions");
211
212    assert!(
213        !socket_path.exists(),
214        "shutdown must remove the socket file"
215    );
216    let _ = std::fs::remove_file(&log_path);
217}
218
219#[test]
220fn run_server_reconciles_drifted_store_on_startup() {
221    let log_path = fresh_log_path();
222    let socket_path = fresh_socket_path();
223
224    let caja_id = Uuid::new_v4();
225    {
226        let executor = Executor::load_module(treasury_module()).expect("load");
227        let mut log = EventLog::open(&log_path).expect("open log");
228        let mut store = MemoryStore::new();
229        seed_and_log(
230            &executor,
231            &mut store,
232            &mut log,
233            "Caja",
234            caja_id,
235            json!({
236                "id": caja_id.to_string(),
237                "name": "A",
238                "saldo": 200_000_i64,
239                "currency": "USD",
240            }),
241        )
242        .expect("seed");
243        execute_and_log(
244            &executor,
245            &mut store,
246            &mut log,
247            "register_cash_move",
248            &[("caja", caja_id)],
249            json!({
250                "monto": 1_500_i64,
251                "tipo": "in",
252                "timestamp": "2026-05-04T09:00:00Z",
253                "memo": "pre-run",
254                "movimiento_id": Uuid::new_v4().to_string(),
255            }),
256        )
257        .expect("deposit");
258    }
259
260    let executor = Executor::load_module(treasury_module()).expect("load");
261    let log = EventLog::open(&log_path).expect("reopen");
262    let empty_store = MemoryStore::new();
263
264    let socket_for_client = socket_path.clone();
265    let client = thread::spawn(move || -> Result<(), String> {
266        let mut conn = Conn::connect_with_retry(&socket_for_client);
267        let resp = conn.exchange(json!({
268            "op": "load",
269            "entity": "Caja",
270            "id": caja_id.to_string(),
271        }));
272        if resp["value"]["saldo"].as_i64() != Some(201_500) {
273            return Err(format!(
274                "expected saldo 201_500 (200k seed + 1.5k replayed deposit), got {}",
275                resp
276            ));
277        }
278        conn.exchange(json!({"op": "shutdown"}));
279        Ok(())
280    });
281
282    run_server(executor, log, empty_store, None, &socket_path).expect("clean exit");
283    client.join().unwrap().expect("client assertions");
284
285    let _ = std::fs::remove_file(&log_path);
286}