diff --git a/Cargo.lock b/Cargo.lock index 91e054d..6b2f1e2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1066,6 +1066,7 @@ dependencies = [ "libc", "maestro-config", "maestro-protocol", + "maestro-testing", "nix", "serde_json", "tempfile", @@ -1122,6 +1123,7 @@ dependencies = [ "chrono", "futures-util", "maestro-protocol", + "maestro-testing", "maestro-tui", "serde", "serde_json", @@ -1156,6 +1158,20 @@ dependencies = [ "uuid", ] +[[package]] +name = "maestro-testing" +version = "0.1.0" +dependencies = [ + "axum", + "chrono", + "futures-util", + "maestro-protocol", + "serde", + "serde_json", + "tokio", + "uuid", +] + [[package]] name = "maestro-tui" version = "0.1.0" @@ -1167,6 +1183,7 @@ dependencies = [ "futures-util", "maestro-config", "maestro-protocol", + "maestro-testing", "ratatui", "serde_json", "thiserror 1.0.69", @@ -1189,6 +1206,7 @@ dependencies = [ "include_dir", "maestro-config", "maestro-protocol", + "maestro-testing", "mime_guess", "reqwest", "serde_json", @@ -1198,6 +1216,7 @@ dependencies = [ "tokio-tungstenite 0.21.0", "tracing", "tracing-subscriber", + "uuid", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index b33b5bb..a5cc5ef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ futures-util = { version = "0.3", default-features = false, features = ["sink", libc = "0.2" maestro-config = { path = "crates/maestro-config" } maestro-protocol = { path = "crates/maestro-protocol" } +maestro-testing = { path = "crates/maestro-testing" } maestro-tui = { path = "crates/maestro-tui" } nix = { version = "0.29", default-features = false, features = ["term", "signal"] } rand = "0.8" diff --git a/crates/maestro-cli/Cargo.toml b/crates/maestro-cli/Cargo.toml index 6ded490..9dc2320 100644 --- a/crates/maestro-cli/Cargo.toml +++ b/crates/maestro-cli/Cargo.toml @@ -29,4 +29,7 @@ tracing-subscriber.workspace = true uuid.workspace = true [dev-dependencies] +maestro-testing.workspace = true tempfile.workspace = true +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +uuid.workspace = true diff --git a/crates/maestro-cli/tests/against_mock_daemon.rs b/crates/maestro-cli/tests/against_mock_daemon.rs new file mode 100644 index 0000000..955cc8b --- /dev/null +++ b/crates/maestro-cli/tests/against_mock_daemon.rs @@ -0,0 +1,107 @@ +//! The CLI against a daemon. It had no such tests: `maestro list`'s formatting +//! was unit-tested, but nothing exercised the client, the error decoding, or +//! the unreachable-daemon path without compiling and forking a real daemon. + +use maestro_protocol::{DaemonClient, DaemonInfo}; +use maestro_testing::{session, MockDaemon}; +use uuid::Uuid; + +#[tokio::test] +async fn list_reads_the_daemon_catalog() { + let mock = MockDaemon::start().await; + mock.add_session(session(Uuid::from_u128(1), "/home/mock/work", "claude")); + + let sessions = mock.client().list_sessions().await.expect("list"); + + assert_eq!(sessions.len(), 1); + assert_eq!(sessions[0].command, "claude"); +} + +#[tokio::test] +async fn info_carries_version_and_presets() { + let mock = MockDaemon::start().await; + mock.set_info(DaemonInfo { + hostname: "konishi".into(), + version: "9.9.9".into(), + home_dir: "/home/naps".into(), + presets: vec![maestro_protocol::SpawnPreset { + id: "claude".into(), + label: "Claude".into(), + command: "claude".into(), + }], + }); + + let info = mock.client().daemon_info().await.expect("info"); + + // The CLI's own DaemonInfo used to drop both of these fields. + assert_eq!(info.version, "9.9.9"); + assert_eq!(info.presets.len(), 1); +} + +#[tokio::test] +async fn a_structured_failure_surfaces_the_daemon_message() { + let mock = MockDaemon::start().await; + mock.force_error(502, "tmux_failed", "no server running"); + + let err = mock + .client() + .list_sessions() + .await + .expect_err("should fail"); + + assert_eq!( + err.daemon_message(), + Some("no server running"), + "the daemon's own message must survive: {err}" + ); + assert!(!err.is_unreachable()); +} + +#[tokio::test] +async fn an_unreachable_daemon_is_distinguishable_from_a_rejection() { + // Port 1 has nothing listening on it. + let client = DaemonClient::new("http://127.0.0.1:1"); + + let err = client.list_sessions().await.expect_err("should fail"); + + assert!( + err.is_unreachable(), + "connection refused must not read as an HTTP rejection: {err}" + ); +} + +#[tokio::test] +async fn killing_an_unknown_session_reports_the_daemon_reason() { + let mock = MockDaemon::start().await; + + let err = mock + .client() + .kill_session(Uuid::new_v4()) + .await + .expect_err("should fail"); + + assert_eq!(err.daemon_message(), Some("no such session on this daemon")); +} + +#[tokio::test] +async fn spawn_round_trips_through_the_daemon() { + let mock = MockDaemon::start().await; + + let created = mock + .client() + .create_session(&maestro_protocol::CreateSessionRequest { + cwd: "/home/mock".into(), + command: "bash".into(), + env: Default::default(), + metadata: Default::default(), + worktree: None, + }) + .await + .expect("create"); + + assert_eq!(created.command, "bash"); + assert_eq!(mock.session_count(), 1); + + mock.client().kill_session(created.id).await.expect("kill"); + assert_eq!(mock.session_count(), 0); +} diff --git a/crates/maestro-e2e/Cargo.toml b/crates/maestro-e2e/Cargo.toml index 5e44fe5..56a4d21 100644 --- a/crates/maestro-e2e/Cargo.toml +++ b/crates/maestro-e2e/Cargo.toml @@ -14,6 +14,7 @@ path = "src/lib.rs" chrono.workspace = true futures-util.workspace = true maestro-protocol = { workspace = true, features = ["client"] } +maestro-testing.workspace = true serde.workspace = true serde_json.workspace = true tempfile.workspace = true diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index bc21c07..5cb166a 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -445,14 +445,13 @@ fn run_cli(args: &[&str], envs: &[(&str, &str)]) -> std::process::Output { #[tokio::test(flavor = "multi_thread")] async fn cli_list_reads_from_daemon() { - let (daemon, client) = bring_up().await; - let cwd = TempDir::new().unwrap(); - let session = client - .spawn_session(&create_req(cwd.path(), "sleep 30")) - .await - .expect("spawn"); + // The subject is the shipped `maestro` binary — arg parsing, config + // precedence, output. It needs a daemon to talk to, not tmux, so this runs + // against an in-memory one. + let mock = maestro_testing::MockDaemon::start().await; + let id = uuid::Uuid::new_v4(); + mock.add_session(maestro_testing::session(id, "/home/mock/work", "sleep 30")); - let daemon_url = daemon.http_base(); let home = TempDir::new().unwrap(); let cfg_path = home.path().join("config.toml"); let out = run_cli( @@ -460,18 +459,16 @@ async fn cli_list_reads_from_daemon() { &[ ("HOME", home.path().to_str().unwrap()), ("MAESTRO_CONFIG", cfg_path.to_str().unwrap()), - ("MAESTRO_DAEMON_URL", &daemon_url), + ("MAESTRO_DAEMON_URL", &mock.http_base()), ], ); let stdout = String::from_utf8_lossy(&out.stdout); let stderr = String::from_utf8_lossy(&out.stderr); assert!(out.status.success(), "list failed: {stderr}\n{stdout}"); assert!( - stdout.contains(&session.id.to_string()), + stdout.contains(&id.to_string()), "session not listed: {stdout}" ); - - client.kill_session(session.id).await.ok(); } #[tokio::test(flavor = "multi_thread")] diff --git a/crates/maestro-e2e/tests/tui_integration.rs b/crates/maestro-e2e/tests/tui_integration.rs deleted file mode 100644 index aeb7bcb..0000000 --- a/crates/maestro-e2e/tests/tui_integration.rs +++ /dev/null @@ -1,129 +0,0 @@ -//! TUI integration tests (NAP-282). -//! -//! Exercises `CatalogStore` and the per-session `SessionTask` against a real -//! daemon subprocess. We don't spin the full ratatui frontend -//! here — those interactions are best tested with `TestBackend` unit tests in -//! `crates/maestro-tui/src/*.rs`. This file fills the gap between those -//! and the headless e2e suite by verifying that the TUI's wire layer plays -//! nicely with a real daemon. -//! -//! Requires `tmux` and `git` on `$PATH`. - -#![allow(clippy::result_large_err)] - -use std::path::PathBuf; -use std::time::{Duration, Instant}; - -use maestro_e2e::Daemon; -use maestro_protocol::CreateSessionRequest; -use maestro_tui::catalog::CatalogStore; - -async fn wait_until(label: &str, deadline: Duration, mut check: F) -where - F: FnMut() -> Fut, - Fut: std::future::Future, -{ - let start = Instant::now(); - loop { - if check().await { - return; - } - if start.elapsed() >= deadline { - panic!("timeout waiting for {label}"); - } - tokio::time::sleep(Duration::from_millis(25)).await; - } -} - -#[tokio::test(flavor = "multi_thread")] -async fn catalog_bootstrap_against_real_daemon() { - let daemon = Daemon::spawn().await; - let ws_url = daemon.client().events_url().unwrap(); - - let store = CatalogStore::default(); - let bg = store.clone(); - tokio::spawn(async move { bg.run_events_loop(ws_url).await }); - - tokio::time::timeout(Duration::from_secs(3), store.wait_for_initial_snapshot()) - .await - .expect("initial snapshot"); - - assert!(store.snapshot().sessions.is_empty()); -} - -#[tokio::test(flavor = "multi_thread")] -async fn post_session_propagates_through_events_ws() { - let daemon = Daemon::spawn().await; - let ws_url = daemon.client().events_url().unwrap(); - - let store = CatalogStore::default(); - let bg = store.clone(); - tokio::spawn(async move { bg.run_events_loop(ws_url).await }); - - tokio::time::timeout(Duration::from_secs(3), store.wait_for_initial_snapshot()) - .await - .expect("initial snapshot"); - - let cwd = PathBuf::from(daemon.home()); - let req = CreateSessionRequest { - cwd, - command: "bash".into(), - env: Default::default(), - metadata: Default::default(), - worktree: None, - }; - daemon - .client() - .create_session(&req) - .await - .expect("create session"); - - wait_until( - "session to surface in catalog", - Duration::from_secs(2), - || async { store.snapshot().sessions.len() == 1 }, - ) - .await; - - let snap = store.snapshot(); - assert_eq!(snap.sessions.len(), 1); - assert_eq!(snap.sessions[0].command, "bash"); -} - -#[tokio::test(flavor = "multi_thread")] -async fn delete_session_propagates_removal() { - let daemon = Daemon::spawn().await; - let ws_url = daemon.client().events_url().unwrap(); - - let store = CatalogStore::default(); - let bg = store.clone(); - tokio::spawn(async move { bg.run_events_loop(ws_url).await }); - - tokio::time::timeout(Duration::from_secs(3), store.wait_for_initial_snapshot()) - .await - .expect("initial snapshot"); - - let cwd = PathBuf::from(daemon.home()); - let req = CreateSessionRequest { - cwd, - command: "bash".into(), - env: Default::default(), - metadata: Default::default(), - worktree: None, - }; - let client = daemon.client(); - client.create_session(&req).await.expect("create session"); - - wait_until("creation", Duration::from_secs(2), || async { - store.snapshot().sessions.len() == 1 - }) - .await; - - let sid = store.snapshot().sessions[0].id; - client.kill_session(sid).await.expect("kill session"); - - wait_until("removal", Duration::from_secs(2), || async { - store.snapshot().sessions.is_empty() - }) - .await; -} diff --git a/crates/maestro-testing/Cargo.toml b/crates/maestro-testing/Cargo.toml new file mode 100644 index 0000000..d651285 --- /dev/null +++ b/crates/maestro-testing/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "maestro-testing" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +publish.workspace = true +description = "In-memory Maestro daemon for tests: the real /api/* surface without tmux" + +[dependencies] +axum.workspace = true +chrono.workspace = true +futures-util.workspace = true +maestro-protocol = { workspace = true, features = ["client"] } +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +uuid.workspace = true diff --git a/crates/maestro-testing/src/lib.rs b/crates/maestro-testing/src/lib.rs new file mode 100644 index 0000000..cce52c5 --- /dev/null +++ b/crates/maestro-testing/src/lib.rs @@ -0,0 +1,364 @@ +//! An in-memory Maestro daemon for tests. +//! +//! Serves the real `/api/*` surface from `maestro-protocol` types, backed by a +//! `Vec` instead of tmux. Use it whenever the subject under test is a +//! client — request building, error decoding, the catalog reducer, the proxy — +//! rather than session management itself. Tests that genuinely need tmux belong +//! in `maestro-e2e` against a real daemon. + +#![deny(missing_debug_implementations)] + +use std::collections::HashMap; +use std::net::{Ipv4Addr, SocketAddr}; +use std::sync::{Arc, Mutex}; + +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::extract::{Path, Query, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::routing::{any, delete, get}; +use axum::{Json, Router}; +use chrono::Utc; +use futures_util::{SinkExt, StreamExt}; +use maestro_protocol::{ + routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonClient, DaemonInfo, DirListing, + GitRoot, RecentDirs, Session, SessionStatus, PROTOCOL_VERSION, +}; +use tokio::net::TcpListener; +use tokio::sync::{broadcast, oneshot}; +use uuid::Uuid; + +/// Build a plausible `Session` for a test to hand to [`MockDaemon::add_session`]. +pub fn session(id: Uuid, cwd: &str, command: &str) -> Session { + Session { + id, + cwd: cwd.into(), + command: command.into(), + env: HashMap::new(), + metadata: HashMap::new(), + status: SessionStatus::Running, + created_at: Utc::now(), + foreground: None, + activity: Default::default(), + last_activity_at: Utc::now(), + current_cwd: None, + } +} + +#[derive(Debug, Default)] +struct Inner { + sessions: Vec, + info: Option, + /// When set, every route answers with this instead of its normal response. + forced_error: Option<(u16, ApiError)>, +} + +#[derive(Clone)] +struct AppState { + inner: Arc>, + events: broadcast::Sender, +} + +impl std::fmt::Debug for AppState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AppState").finish_non_exhaustive() + } +} + +/// A running mock daemon. Shuts down when dropped. +#[derive(Debug)] +pub struct MockDaemon { + addr: SocketAddr, + state: AppState, + _shutdown: oneshot::Sender<()>, +} + +impl MockDaemon { + pub async fn start() -> Self { + let state = AppState { + inner: Arc::new(Mutex::new(Inner::default())), + events: broadcast::channel(64).0, + }; + let app = Router::new() + .route(routes::SESSIONS, get(list_sessions).post(create_session)) + .route(routes::patterns::SESSION, delete(kill_session)) + .route(routes::INFO, get(daemon_info)) + .route(routes::DIRS, get(list_dir)) + .route(routes::RECENT_DIRS, get(recent_dirs)) + .route(routes::GITROOT, get(git_root)) + .route(routes::EVENTS, any(events_ws)) + .route(routes::patterns::ATTACH, any(attach_ws)) + .with_state(state.clone()); + + let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let (tx, rx) = oneshot::channel::<()>(); + tokio::spawn(async move { + let _ = axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = rx.await; + }) + .await; + }); + + Self { + addr, + state, + _shutdown: tx, + } + } + + pub fn addr(&self) -> SocketAddr { + self.addr + } + + pub fn http_base(&self) -> String { + format!("http://{}", self.addr) + } + + /// A client bound to this daemon. + pub fn client(&self) -> DaemonClient { + DaemonClient::new(self.http_base()) + } + + /// Add a session and announce it, as a real spawn would. + pub fn add_session(&self, session: Session) { + self.state + .inner + .lock() + .expect("poisoned") + .sessions + .push(session.clone()); + let _ = self + .state + .events + .send(CatalogEvent::SessionAdded { session }); + } + + /// Remove a session and announce it. + pub fn remove_session(&self, id: Uuid) { + self.state + .inner + .lock() + .expect("poisoned") + .sessions + .retain(|s| s.id != id); + let _ = self.state.events.send(CatalogEvent::SessionRemoved { id }); + } + + pub fn set_info(&self, info: DaemonInfo) { + self.state.inner.lock().expect("poisoned").info = Some(info); + } + + /// Make every route answer with this error, so clients' failure paths — and + /// their [`ApiError`] decoding — can be exercised. + pub fn force_error(&self, status: u16, code: &str, message: &str) { + self.state.inner.lock().expect("poisoned").forced_error = + Some((status, ApiError::new(code, message))); + } + + pub fn clear_error(&self) { + self.state.inner.lock().expect("poisoned").forced_error = None; + } + + pub fn session_count(&self) -> usize { + self.state.inner.lock().expect("poisoned").sessions.len() + } +} + +fn forced(state: &AppState) -> Option { + let guard = state.inner.lock().expect("poisoned"); + guard.forced_error.as_ref().map(|(status, body)| { + ( + StatusCode::from_u16(*status).expect("valid status"), + Json(body.clone()), + ) + .into_response() + }) +} + +async fn list_sessions(State(s): State) -> Response { + if let Some(r) = forced(&s) { + return r; + } + let sessions = s.inner.lock().expect("poisoned").sessions.clone(); + Json(sessions).into_response() +} + +async fn create_session( + State(s): State, + Json(req): Json, +) -> Response { + if let Some(r) = forced(&s) { + return r; + } + let mut created = session(Uuid::new_v4(), &req.cwd.to_string_lossy(), &req.command); + created.env = req.env; + created.metadata = req.metadata; + s.inner + .lock() + .expect("poisoned") + .sessions + .push(created.clone()); + let _ = s.events.send(CatalogEvent::SessionAdded { + session: created.clone(), + }); + Json(created).into_response() +} + +async fn kill_session(State(s): State, Path(id): Path) -> Response { + if let Some(r) = forced(&s) { + return r; + } + let existed = { + let mut guard = s.inner.lock().expect("poisoned"); + let before = guard.sessions.len(); + guard.sessions.retain(|x| x.id != id); + guard.sessions.len() != before + }; + if !existed { + return ( + StatusCode::NOT_FOUND, + Json(ApiError::new("not_found", "no such session on this daemon")), + ) + .into_response(); + } + let _ = s.events.send(CatalogEvent::SessionRemoved { id }); + StatusCode::NO_CONTENT.into_response() +} + +async fn daemon_info(State(s): State) -> Response { + if let Some(r) = forced(&s) { + return r; + } + let info = s + .inner + .lock() + .expect("poisoned") + .info + .clone() + .unwrap_or_else(|| DaemonInfo { + hostname: "mock-host".into(), + version: PROTOCOL_VERSION.into(), + home_dir: "/home/mock".into(), + presets: Vec::new(), + }); + Json(info).into_response() +} + +async fn list_dir(State(s): State, Query(q): Query) -> Response { + if let Some(r) = forced(&s) { + return r; + } + Json(DirListing { + path: q.path.into(), + entries: Vec::new(), + truncated: false, + }) + .into_response() +} + +async fn recent_dirs(State(s): State) -> Response { + if let Some(r) = forced(&s) { + return r; + } + Json(RecentDirs { + dirs: Vec::new(), + truncated: false, + }) + .into_response() +} + +async fn git_root(State(s): State, Query(_q): Query) -> Response { + if let Some(r) = forced(&s) { + return r; + } + Json(GitRoot { root: None }).into_response() +} + +/// Snapshot on connect, then live deltas — the same contract the daemon offers. +async fn events_ws(State(s): State, ws: WebSocketUpgrade) -> Response { + ws.on_upgrade(move |socket| events_loop(s, socket)) +} + +async fn events_loop(s: AppState, socket: WebSocket) { + let (mut sink, mut stream) = socket.split(); + let mut rx = s.events.subscribe(); + let snapshot = CatalogEvent::Snapshot { + sessions: s.inner.lock().expect("poisoned").sessions.clone(), + }; + if sink + .send(Message::Text( + serde_json::to_string(&snapshot).expect("encode"), + )) + .await + .is_err() + { + return; + } + loop { + tokio::select! { + ev = rx.recv() => { + let Ok(ev) = ev else { break }; + let text = serde_json::to_string(&ev).expect("encode"); + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + frame = stream.next() => { + match frame { + None | Some(Err(_)) | Some(Ok(Message::Close(_))) => break, + Some(Ok(_)) => {} + } + } + } + } +} + +/// Echoes whatever it is sent, so a client's PTY plumbing can be exercised. +async fn attach_ws( + State(s): State, + Path(id): Path, + ws: WebSocketUpgrade, +) -> Response { + let known = s + .inner + .lock() + .expect("poisoned") + .sessions + .iter() + .any(|x| x.id == id); + if !known { + return ( + StatusCode::NOT_FOUND, + Json(ApiError::new("not_found", "no such session on this daemon")), + ) + .into_response(); + } + ws.on_upgrade(|mut socket| async move { + while let Some(Ok(msg)) = socket.next().await { + match msg { + Message::Binary(b) => { + if socket.send(Message::Binary(b)).await.is_err() { + break; + } + } + Message::Text(t) => { + if socket.send(Message::Text(t)).await.is_err() { + break; + } + } + Message::Close(_) => break, + _ => {} + } + } + }) +} + +#[derive(Debug, serde::Deserialize)] +struct PathQuery { + #[serde(default)] + path: String, +} diff --git a/crates/maestro-tui/Cargo.toml b/crates/maestro-tui/Cargo.toml index 0891a7e..dd99eff 100644 --- a/crates/maestro-tui/Cargo.toml +++ b/crates/maestro-tui/Cargo.toml @@ -23,6 +23,7 @@ uuid.workspace = true vt100.workspace = true [dev-dependencies] +maestro-testing.workspace = true axum.workspace = true chrono.workspace = true tokio = { workspace = true, features = ["sync", "time", "macros", "rt-multi-thread", "net"] } diff --git a/crates/maestro-tui/tests/catalog_ws.rs b/crates/maestro-tui/tests/catalog_ws.rs index f914a81..766ff95 100644 --- a/crates/maestro-tui/tests/catalog_ws.rs +++ b/crates/maestro-tui/tests/catalog_ws.rs @@ -1,84 +1,42 @@ -//! WS round-trip test for `CatalogStore::run_events_loop`. Spins a tiny axum -//! server that emulates the daemon's `/api/events` contract: send a `Snapshot` -//! on connect, then a delta. +//! `CatalogStore::run_events_loop` against a daemon serving the real +//! `/api/events` contract, rather than a scripted socket. use std::time::Duration; -use axum::extract::ws::{Message, WebSocketUpgrade}; -use axum::response::Response; -use axum::routing::any; -use axum::Router; -use chrono::Utc; -use maestro_protocol::{CatalogEvent, Session, SessionStatus}; +use maestro_testing::{session, MockDaemon}; use maestro_tui::catalog::CatalogStore; use uuid::Uuid; -fn session(id: u128) -> Session { - Session { - id: Uuid::from_u128(id), - command: "bash".into(), - cwd: std::path::PathBuf::from("/tmp"), - env: Default::default(), - metadata: Default::default(), - status: SessionStatus::Running, - created_at: Utc::now(), - foreground: None, - activity: Default::default(), - last_activity_at: chrono::Utc::now(), - - current_cwd: None, - } -} - -async fn events_handler(ws: WebSocketUpgrade) -> Response { - ws.on_upgrade(|mut socket| async move { - let snapshot = CatalogEvent::Snapshot { sessions: vec![] }; - let text = serde_json::to_string(&snapshot).unwrap(); - if socket.send(Message::Text(text)).await.is_err() { - return; - } - // Small delay so the client has the snapshot applied before the delta. - tokio::time::sleep(Duration::from_millis(50)).await; - let added = CatalogEvent::SessionAdded { - session: session(1), - }; - let text = serde_json::to_string(&added).unwrap(); - let _ = socket.send(Message::Text(text)).await; - // Keep the connection alive so the client doesn't enter reconnect. - tokio::time::sleep(Duration::from_secs(2)).await; - }) -} - #[tokio::test(flavor = "multi_thread")] -async fn run_events_loop_applies_snapshot_then_delta() { - let app = Router::new().route("/api/events", any(events_handler)); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - axum::serve(listener, app).await.unwrap(); - }); +async fn applies_the_snapshot_then_live_deltas() { + let mock = MockDaemon::start().await; + mock.add_session(session(Uuid::from_u128(1), "/home/mock", "bash")); + let ws_url = mock.client().events_url().expect("events url"); let store = CatalogStore::default(); let bg = store.clone(); - let ws_url = format!("ws://{addr}/api/events"); tokio::spawn(async move { bg.run_events_loop(ws_url).await }); - // Bootstrap should land within ~250 ms. - tokio::time::timeout( - Duration::from_millis(500), - store.wait_for_initial_snapshot(), - ) - .await - .expect("initial snapshot never arrived"); + tokio::time::timeout(Duration::from_secs(3), store.wait_for_initial_snapshot()) + .await + .expect("initial snapshot"); + assert_eq!(store.snapshot().sessions.len(), 1); - assert!(store.snapshot().sessions.is_empty()); + mock.add_session(session(Uuid::from_u128(2), "/home/mock", "claude")); + wait_until(|| store.snapshot().sessions.len() == 2).await; - // Delta arrives shortly after. Poll for up to 500 ms. - for _ in 0..50 { - if store.snapshot().sessions.len() == 1 { + mock.remove_session(Uuid::from_u128(1)); + wait_until(|| store.snapshot().sessions.len() == 1).await; + assert_eq!(store.snapshot().sessions[0].id, Uuid::from_u128(2)); +} + +async fn wait_until(mut cond: impl FnMut() -> bool) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(3); + while tokio::time::Instant::now() < deadline { + if cond() { return; } tokio::time::sleep(Duration::from_millis(10)).await; } - panic!("delta never propagated; got {:?}", store.snapshot()); + panic!("timed out waiting for the catalog to converge"); } diff --git a/crates/maestro-web/Cargo.toml b/crates/maestro-web/Cargo.toml index 0c325f9..93833e2 100644 --- a/crates/maestro-web/Cargo.toml +++ b/crates/maestro-web/Cargo.toml @@ -33,6 +33,8 @@ tracing.workspace = true tracing-subscriber.workspace = true [dev-dependencies] +maestro-testing.workspace = true +uuid.workspace = true reqwest.workspace = true serde_json.workspace = true tempfile.workspace = true diff --git a/crates/maestro-web/tests/proxy.rs b/crates/maestro-web/tests/proxy.rs index 407780a..91af862 100644 --- a/crates/maestro-web/tests/proxy.rs +++ b/crates/maestro-web/tests/proxy.rs @@ -1,65 +1,18 @@ //! Integration tests for the maestro-web reverse-proxy. //! -//! Spins up a tiny axum "daemon" on a random port, points maestro-web at it, -//! and exercises both the HTTP and WS proxy paths end-to-end. +//! Points maestro-web at a `MockDaemon` and exercises both the HTTP and WS +//! proxy paths end-to-end. +use maestro_testing::MockDaemon; use std::net::{Ipv4Addr, SocketAddr}; use std::time::Duration; -use axum::extract::ws::{Message as AxMessage, WebSocket, WebSocketUpgrade}; -use axum::response::IntoResponse; -use axum::routing::{any, get}; -use axum::{Json, Router}; use futures_util::{SinkExt, StreamExt}; use maestro_web::config::WebConfig; use tokio::net::TcpListener; use tokio::sync::oneshot; use tokio_tungstenite::tungstenite::Message as WsMessage; -async fn spawn_mock_daemon() -> (SocketAddr, oneshot::Sender<()>) { - async fn list_sessions() -> Json { - Json(serde_json::json!([{"id": "session-1"}])) - } - async fn daemon_info() -> Json { - Json(serde_json::json!({"hostname": "mock-host"})) - } - async fn echo_ws(ws: WebSocketUpgrade) -> impl IntoResponse { - ws.on_upgrade(echo_handler) - } - async fn echo_handler(mut socket: WebSocket) { - while let Some(Ok(msg)) = socket.next().await { - match msg { - AxMessage::Text(t) => { - let _ = socket.send(AxMessage::Text(format!("echo:{t}"))).await; - } - AxMessage::Binary(b) => { - let _ = socket.send(AxMessage::Binary(b)).await; - } - AxMessage::Close(_) => break, - _ => {} - } - } - } - - let app = Router::new() - .route("/api/sessions", get(list_sessions)) - .route("/api/info", get(daemon_info)) - .route("/api/attach/:id", any(echo_ws)); - let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) - .await - .unwrap(); - let addr = listener.local_addr().unwrap(); - let (tx, rx) = oneshot::channel::<()>(); - tokio::spawn(async move { - let _ = axum::serve(listener, app) - .with_graceful_shutdown(async move { - let _ = rx.await; - }) - .await; - }); - (addr, tx) -} - async fn spawn_web(daemon_url: String) -> (SocketAddr, oneshot::Sender<()>) { let cfg = WebConfig { bind_addr: SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), @@ -81,7 +34,8 @@ async fn spawn_web(daemon_url: String) -> (SocketAddr, oneshot::Sender<()>) { #[tokio::test] async fn http_get_api_path_forwards_to_daemon() { - let (daemon_addr, _d) = spawn_mock_daemon().await; + let daemon = MockDaemon::start().await; + let daemon_addr = daemon.addr(); let (web_addr, _w) = spawn_web(format!("http://{daemon_addr}")).await; let body: serde_json::Value = reqwest::get(format!("http://{web_addr}/api/sessions")) @@ -90,12 +44,13 @@ async fn http_get_api_path_forwards_to_daemon() { .json() .await .unwrap(); - assert_eq!(body[0]["id"], "session-1"); + assert!(body.as_array().expect("array").is_empty()); } #[tokio::test] async fn http_get_unrelated_path_falls_back_to_spa_index() { - let (daemon_addr, _d) = spawn_mock_daemon().await; + let daemon = MockDaemon::start().await; + let daemon_addr = daemon.addr(); let (web_addr, _w) = spawn_web(format!("http://{daemon_addr}")).await; // The bundle is empty in test builds — the placeholder index.html serves. @@ -113,22 +68,55 @@ async fn http_get_unrelated_path_falls_back_to_spa_index() { #[tokio::test] async fn ws_api_attach_forwards_to_daemon_and_echoes() { - let (daemon_addr, _d) = spawn_mock_daemon().await; + let daemon = MockDaemon::start().await; + let id = uuid::Uuid::from_u128(7); + daemon.add_session(maestro_testing::session(id, "/home/mock", "bash")); + let daemon_addr = daemon.addr(); let (web_addr, _w) = spawn_web(format!("http://{daemon_addr}")).await; - let url = format!("ws://{web_addr}/api/attach/abc"); + let url = format!("ws://{web_addr}/api/attach/{id}"); let (mut ws, _) = tokio_tungstenite::connect_async(url).await.unwrap(); - ws.send(WsMessage::Text("hello".into())).await.unwrap(); + ws.send(WsMessage::Binary(b"hello".to_vec())).await.unwrap(); let echoed = tokio::time::timeout(Duration::from_secs(1), ws.next()) .await .unwrap(); match echoed { - Some(Ok(WsMessage::Text(t))) => assert_eq!(t, "echo:hello"), + Some(Ok(WsMessage::Binary(b))) => assert_eq!(b, b"hello"), other => panic!("unexpected frame: {other:?}"), } } +/// The client is upgraded before the upstream connection is attempted, so an +/// upstream rejection has to come back as a close code. This is the end-to-end +/// counterpart to `upstream_close_reason`, which had only unit tests: the +/// daemon's own message must reach the browser rather than a generic failure. +#[tokio::test] +async fn ws_attach_to_an_unknown_session_closes_with_the_daemon_reason() { + let daemon = MockDaemon::start().await; + let (web_addr, _w) = spawn_web(format!("http://{}", daemon.addr())).await; + + let url = format!("ws://{web_addr}/api/attach/{}", uuid::Uuid::new_v4()); + let (mut ws, _) = tokio_tungstenite::connect_async(url) + .await + .expect("upgrade"); + + let frame = tokio::time::timeout(Duration::from_secs(2), ws.next()) + .await + .expect("close frame"); + match frame { + Some(Ok(WsMessage::Close(Some(cf)))) => { + assert_eq!(u16::from(cf.code), 4502); + assert!( + cf.reason.contains("no such session"), + "expected the daemon's message, got {:?}", + cf.reason + ); + } + other => panic!("expected a Close frame with a reason, got {other:?}"), + } +} + #[tokio::test] async fn proxy_returns_502_when_daemon_unreachable() { // Bind a port, immediately drop the listener, hand the address to the