//! Tests for `bxctl chat`, against a fake `loopd` that speaks the channel protocol. Do not edit. use bxctl::chat::{ChatError, Printer, new_session_id, run_turn}; use proto::{ DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnDone, TurnEvent, Usage, WireError, read_frame, write_frame, }; use std::os::unix::net::UnixListener; use std::path::PathBuf; use std::process::Command; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; static NEXT: AtomicU32 = AtomicU32::new(0); /// What the fake `loopd` sends back for one turn, after the events: done or an error. #[derive(Clone)] enum End { Done(TurnDone), Error(ErrorCode, &'static str), } struct FakeLoopd { socket: PathBuf, turns: Arc>>, } fn usage() -> Usage { Usage { cache_n: 10, prompt_n: 5, predicted_n: 7, reasoning_tokens: 3, thinking_capped: false, } } /// Serves every connection with the same script: the given events, then `end`. Records the /// turns it received. fn fake_loopd(events: Vec, end: End) -> FakeLoopd { let n = NEXT.fetch_add(1, Ordering::SeqCst); let dir = std::env::temp_dir().join(format!("bxctl-test-{}-{n}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); let socket = dir.join("loop.sock"); let listener = UnixListener::bind(&socket).unwrap(); let turns = Arc::new(Mutex::new(Vec::new())); let seen = Arc::clone(&turns); thread::spawn(move || { for stream in listener.incoming() { let mut stream = stream.unwrap(); let request = read_frame(&mut stream).unwrap(); let Message::Turn(turn) = request.msg else { panic!("not a turn") }; seen.lock().unwrap().push(turn.clone()); let id = request.id; let end = if turn.resume && turn.content == "trigger-no-such-session" { End::Error(ErrorCode::NoSuchSession, "session x does not exist") } else { end.clone() }; for e in &events { write_frame( &mut stream, &Envelope { v: PROTOCOL_VERSION, id, r#final: false, msg: Message::TurnEvent(e.clone()), }, ) .unwrap(); } let last = match end { End::Done(done) => Message::TurnDone(done), End::Error(code, detail) => Message::Error(WireError { code, detail: detail.to_string(), }), }; write_frame( &mut stream, &Envelope { v: PROTOCOL_VERSION, id, r#final: true, msg: last, }, ) .unwrap(); } }); FakeLoopd { socket, turns } } fn events() -> Vec { vec![ TurnEvent::Queued { ahead: 1 }, TurnEvent::Waiting { slot_busy: true }, TurnEvent::Progress { total: 100, cache: 50, processed: 75, }, TurnEvent::Reasoning { text: "let me ".to_string(), }, TurnEvent::Reasoning { text: "think".to_string(), }, TurnEvent::ToolCallStarted { name: "clock".to_string(), }, TurnEvent::ToolResult { name: "clock".to_string(), class: DataClass::Public, truncated: true, }, TurnEvent::ThinkingCapped { tokens: 4096 }, TurnEvent::Retrying { attempt: 2, after_ms: 1500, error: "the server went silent".to_string(), }, TurnEvent::CacheLoss { expected: 500, got: 20, }, TurnEvent::Content { text: "It is ".to_string(), }, TurnEvent::Content { text: "noon.".to_string(), }, ] } fn done() -> End { End::Done(TurnDone { content: "It is noon.".to_string(), usage: usage(), }) } fn id(s: &str) -> SessionId { SessionId::new(s).unwrap() } #[test] fn run_turn_delivers_every_event_in_order_then_the_answer() { let fake = fake_loopd(events(), done()); let mut seen = Vec::new(); let done = run_turn( &fake.socket, &id("s1"), "what time is it?", false, &mut |e| seen.push(e.clone()), ) .unwrap(); assert_eq!(seen, events()); assert_eq!(done.content, "It is noon."); assert_eq!(done.usage, usage()); let turns = fake.turns.lock().unwrap(); assert_eq!(turns.len(), 1); assert_eq!( turns[0], Turn { session: id("s1"), content: "what time is it?".to_string(), resume: false } ); } #[test] fn an_error_frame_is_refused_with_its_code_and_detail() { let fake = fake_loopd( vec![TurnEvent::Content { text: "partial".to_string(), }], End::Error(ErrorCode::SessionFull, "this conversation is full"), ); let mut seen = 0; let err = run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| seen += 1).unwrap_err(); assert_eq!(seen, 1, "events before the error are still delivered"); match err { ChatError::Refused(w) => { assert_eq!(w.code, ErrorCode::SessionFull); assert_eq!(w.detail, "this conversation is full"); } other => panic!("{other:?}"), } let e: Box = Box::new(run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| {}).unwrap_err()); assert!(e.to_string().contains("session full"), "{e}"); } #[test] fn no_loopd_is_a_connect_error() { let missing = std::env::temp_dir().join("bxctl-no-such-loopd.sock"); assert!(matches!( run_turn(&missing, &id("s1"), "x", false, &mut |_| {}), Err(ChatError::Connect(_)) )); } #[test] fn new_session_ids_are_valid_and_distinct() { let a = new_session_id(); let b = new_session_id(); assert!(a.as_str().starts_with("chat-")); assert_ne!(a, b); } #[test] fn the_printer_formats_each_event_kind() { let mut out = Vec::new(); let mut p = Printer::new(true, false); for e in events() { p.event(&mut out, &e).unwrap(); } p.end_reasoning(&mut out).unwrap(); let text = String::from_utf8(out).unwrap(); assert!( text.contains("\x1b[2mlet me think\x1b[0m\n"), "reasoning dimmed, joined, and ended once: {text:?}" ); assert!(text.contains("[tool clock]\n"), "{text:?}"); assert!(text.contains("[clock: Public, truncated]\n"), "{text:?}"); assert!(text.contains("[waiting: slot busy]\n"), "{text:?}"); assert!( text.contains("[retrying: attempt 2 in 1500 ms: the server went silent]\n"), "{text:?}" ); assert!( text.contains("[thinking capped at 4096 tokens]\n"), "{text:?}" ); assert!(text.contains("[cache loss: 20 of 500]\n"), "{text:?}"); assert!( text.ends_with("It is noon."), "content streams as it is: {text:?}" ); assert!( !text.contains("Queued") && !text.contains("Progress"), "queued and progress are silent: {text:?}" ); let mut out = Vec::new(); let mut p = Printer::new(false, false); for e in events() { p.event(&mut out, &e).unwrap(); } let text = String::from_utf8(out).unwrap(); assert!( !text.contains("let me"), "--no-thinking hides reasoning: {text:?}" ); assert!( !text.contains("\x1b["), "and no escape codes are left: {text:?}" ); let mut out = Vec::new(); let mut p = Printer::new(true, true); for e in events() { p.event(&mut out, &e).unwrap(); } let text = String::from_utf8(out).unwrap(); let lines: Vec<&str> = text.lines().collect(); assert_eq!( lines.len(), events().len(), "--json: one line per event, none skipped" ); let first: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); assert_eq!(first, serde_json::json!({"event": "queued", "ahead": 1})); } #[test] fn say_prints_only_the_answer_on_stdout() { let fake = fake_loopd(events(), done()); let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["chat", "--socket"]) .arg(&fake.socket) .args(["--session", "scripted-1", "--say", "what time is it?"]) .output() .unwrap(); assert!(output.status.success()); assert_eq!(String::from_utf8_lossy(&output.stdout), "It is noon.\n"); let stderr = String::from_utf8_lossy(&output.stderr); assert!( stderr.contains("[tool clock]"), "events go to stderr: {stderr}" ); let turns = fake.turns.lock().unwrap(); assert_eq!(turns.len(), 1); assert_eq!(turns[0].session, id("scripted-1")); assert!( turns[0].resume, "a session given on the command line is resumed" ); } #[test] fn say_creates_a_named_session_that_does_not_exist_yet() { let fake = fake_loopd(vec![], done()); let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["chat", "--socket"]) .arg(&fake.socket) .args(["--session", "fresh", "--say", "trigger-no-such-session"]) .output() .unwrap(); assert!( output.status.success(), "{}", String::from_utf8_lossy(&output.stderr) ); let turns = fake.turns.lock().unwrap(); assert_eq!(turns.len(), 2, "resume was refused, so it was created"); assert!(turns[0].resume && !turns[1].resume); } #[test] fn say_reports_an_error_on_stderr_with_status_1() { let fake = fake_loopd( vec![], End::Error(ErrorCode::TurnLimit, "the turn hit a limit"), ); let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["chat", "--socket"]) .arg(&fake.socket) .args(["--say", "x"]) .output() .unwrap(); assert_eq!(output.status.code(), Some(1)); assert_eq!(String::from_utf8_lossy(&output.stdout), ""); assert!(String::from_utf8_lossy(&output.stderr).contains("turn limit: the turn hit a limit")); } #[test] fn json_mode_prints_frames_as_json_lines() { let fake = fake_loopd(events(), done()); let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["chat", "--socket"]) .arg(&fake.socket) .args(["--session", "j", "--json", "--say", "x"]) .output() .unwrap(); assert!(output.status.success()); let stderr = String::from_utf8_lossy(&output.stderr); let lines: Vec<&str> = stderr.lines().collect(); assert_eq!( lines.len(), events().len() + 1, "every event, then the done frame: {stderr}" ); let last: serde_json::Value = serde_json::from_str(lines[lines.len() - 1]).unwrap(); assert_eq!(last["content"], "It is noon."); assert_eq!(last["usage"]["cache_n"], 10); } #[test] fn interactive_mode_reads_lines_until_quit() { let fake = fake_loopd( vec![TurnEvent::Content { text: "ok".to_string(), }], End::Done(TurnDone { content: "ok".to_string(), usage: usage(), }), ); let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["chat", "--socket"]) .arg(&fake.socket) .stdin(std::process::Stdio::piped()) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .spawn() .unwrap(); { let mut stdin = child.stdin.take().unwrap(); std::io::Write::write_all(&mut stdin, b"first\n\nsecond\n/quit\nnever sent\n").unwrap(); } let output = child.wait_with_output().unwrap(); assert!(output.status.success()); let stdout = String::from_utf8_lossy(&output.stdout); assert!(stdout.starts_with("session chat-"), "{stdout}"); assert_eq!(stdout.matches("ok").count(), 2, "{stdout}"); let turns = fake.turns.lock().unwrap(); assert_eq!( turns.len(), 2, "a blank line sends nothing, and /quit stops" ); assert!( !turns[0].resume && turns[1].resume, "the first turn creates, the second resumes" ); assert_eq!(turns[0].session, turns[1].session); } #[test] fn bad_arguments_print_usage() { let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["chat", "--session", "Not Valid!"]) .output() .unwrap(); assert_eq!(output.status.code(), Some(2)); assert!(String::from_utf8_lossy(&output.stderr).contains("usage")); let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) .args(["dance"]) .output() .unwrap(); assert_eq!(output.status.code(), Some(2)); }