naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

test(maestro): rework e2e harness for a single daemon

Browse Source 
Drops the registry subprocess and the T2 broker scenarios; the harness
spawns one daemon and the client talks straight to its /api/* surface.
This commit is contained in:
Miguel Palhas
2026-08-22 10:21:27 +01:00
parent 1bb464be4b
commit de097cf17f
6 changed files with 125 additions and 815 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-e2e/src/client.rs
+11 -86
View File
@@ -1,15 +1,13 @@
//! Thin HTTP+WS client used by tests to drive the registry.
//!
//! Cookies are managed automatically by `reqwest`'s built-in cookie store.
//! Thin HTTP+WS client used by tests to drive a daemon's `/api/*` surface.
#![allow(clippy::result_large_err)]
use std::sync::Arc;
use std::time::Duration;
use futures_util::{SinkExt, StreamExt};
use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, RegistryEvent, Session};
use maestro_protocol::{CatalogEvent, CreateSessionRequest, Session};
use reqwest::Client as HttpClient;
use std::sync::Arc;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::Message as WsMessage;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
@@ -28,7 +26,7 @@ pub enum ClientError {
Status { status: u16, body: String },
}
/// Sticky-cookie HTTP client + a few WS-aware helpers.
/// HTTP client plus a few WS-aware helpers, bound to one daemon.
#[derive(Debug, Clone)]
pub struct TestClient {
http_base: String,
@@ -39,7 +37,6 @@ pub struct TestClient {
impl TestClient {
pub fn new(http_base: &str, ws_base: &str) -> Self {
let http = HttpClient::builder()
.cookie_store(true)
.timeout(Duration::from_secs(5))
.build()
.expect("reqwest client");
@@ -50,23 +47,6 @@ impl TestClient {
}
}
/// Hit any GET endpoint so the registry issues the auth cookie that all
/// subsequent calls reuse.
pub async fn bootstrap(&self) -> Result<(), ClientError> {
let _ = self.list_daemons().await?;
Ok(())
}
pub async fn list_daemons(&self) -> Result<Vec<DaemonInfo>, ClientError> {
let r = self
.http
.get(format!("{}/api/daemons", self.http_base))
.send()
.await?;
check_ok(&r)?;
Ok(r.json().await?)
}
pub async fn list_sessions(&self) -> Result<Vec<Session>, ClientError> {
let r = self
.http
@@ -105,50 +85,23 @@ impl TestClient {
Ok(())
}
pub async fn connect_info(&self, id: Uuid) -> Result<ConnectInfo, ClientError> {
let r = self
.http
.get(format!("{}/api/sessions/{id}/connect-info", self.http_base))
.send()
.await?;
check_ok(&r)?;
Ok(r.json().await?)
}
/// Connect to the daemon's per-session WS and return the stream. The
/// caller drives the conversation.
pub async fn attach(
&self,
info: &ConnectInfo,
id: Uuid,
) -> Result<WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>, ClientError> {
let url = format!(
"{}/sessions/{id}/stream?token={}",
info.daemon_url.trim_end_matches('/'),
info.jwt
);
let url = format!("{}/api/attach/{id}", self.ws_base);
let request = url.into_client_request()?;
let (ws, _resp) = connect_async(request).await?;
Ok(ws)
}
/// Subscribe to the registry's `/api/events` WS. Returns a tokio task
/// handle that drives the WS and pushes parsed events into an Arc<Mutex<Vec>>.
/// Subscribe to the daemon's `/api/events` WS. Returns a handle whose
/// background task pushes parsed events into a shared vector.
pub async fn subscribe_events(&self) -> EventStream {
// Forward the latest cookie value with the upgrade — `reqwest` and
// `tungstenite` don't share cookie jars, so we copy the cookie out by
// hand. The registry will accept us either way (missing cookie just
// gets a fresh one), but reusing the same identity keeps log noise
// down.
let cookie_header = self.grab_cookie().await.unwrap_or_default();
let url = format!("{}/api/events", self.ws_base);
let mut request = url.into_client_request().expect("ws req");
if !cookie_header.is_empty() {
request.headers_mut().insert(
reqwest::header::COOKIE,
cookie_header.parse().expect("cookie hv"),
);
}
let request = url.into_client_request().expect("ws req");
let (ws, _resp) = connect_async(request).await.expect("events ws");
let events = Arc::new(tokio::sync::Mutex::new(Vec::new()));
@@ -158,7 +111,7 @@ impl TestClient {
while let Some(msg) = stream.next().await {
match msg {
Ok(WsMessage::Text(t)) => {
if let Ok(ev) = serde_json::from_str::<RegistryEvent>(&t) {
if let Ok(ev) = serde_json::from_str::<CatalogEvent>(&t) {
stream_events.lock().await.push(ev);
}
}
@@ -170,34 +123,6 @@ impl TestClient {
});
EventStream { events, task }
}
async fn grab_cookie(&self) -> Option<String> {
// Force a request to make sure the jar has a cookie, then read it back
// via a debug header roundtrip. reqwest's CookieStore isn't directly
// exposed; the simplest path is to ask the registry for an auth
// cookie and capture the Set-Cookie from the response.
let r = self
.http
.get(format!("{}/api/daemons", self.http_base))
.send()
.await
.ok()?;
// The cookie store has the cookie now, but reqwest doesn't surface
// it. We rebuild from Set-Cookie if present.
r.headers()
.get_all(reqwest::header::SET_COOKIE)
.iter()
.find_map(|hv| hv.to_str().ok().and_then(extract_cookie_pair))
}
}
fn extract_cookie_pair(set_cookie: &str) -> Option<String> {
// Take everything before the first ';', which is `name=value`.
set_cookie
.split(';')
.next()
.map(str::trim)
.map(String::from)
}
fn check_ok(r: &reqwest::Response) -> Result<(), ClientError> {
@@ -214,12 +139,12 @@ fn check_ok(r: &reqwest::Response) -> Result<(), ClientError> {
/// Handle to a background task draining the events WS into a vector.
#[derive(Debug)]
pub struct EventStream {
events: Arc<tokio::sync::Mutex<Vec<RegistryEvent>>>,
events: Arc<tokio::sync::Mutex<Vec<CatalogEvent>>>,
task: tokio::task::JoinHandle<()>,
}
impl EventStream {
pub async fn snapshot(&self) -> Vec<RegistryEvent> {
pub async fn snapshot(&self) -> Vec<CatalogEvent> {
self.events.lock().await.clone()
}
}
crates/maestro-e2e/src/daemon.rs
+39 -131
View File
@@ -1,4 +1,4 @@
//! Subprocess wrapper around `maestrod`.
//! Subprocess wrapper around `maestro daemon`.
use std::net::{Ipv4Addr, SocketAddr};
use std::path::PathBuf;
@@ -10,68 +10,49 @@ use tempfile::TempDir;
use crate::build::workspace_target_bin;
use crate::util::{free_port, wait_for_async};
/// A live `maestrod` running in a child process. Kill the child on drop.
/// A live daemon running in a child process. Kill the child on drop.
///
/// `state_dir`, `home`, and `tmux_tmpdir` are kept as `TempDir`s so they
/// outlive the process and get cleaned up automatically.
/// `home` and `tmux_tmpdir` are kept as `TempDir`s so they outlive the process
/// and get cleaned up automatically. State lives under `home` via
/// `XDG_STATE_HOME`.
#[derive(Debug)]
pub struct Daemon {
pub daemon_id: String,
pub bind_addr: SocketAddr,
pub daemon_url: String,
child: Option<Child>,
home: TempDir,
state: TempDir,
tmux: TempDir,
registry_url: String,
token: String,
}
/// Optional knobs for `Daemon::spawn_with` — perimeter advertisement for
/// NAP-247 broker tests.
#[derive(Debug, Default, Clone)]
pub struct DaemonOptions {
pub perimeter_id: Option<String>,
pub private_addr: Option<String>,
}
impl Daemon {
/// Spawn a fresh daemon pointing at `registry_base_url` (the registry's
/// HTTP base, e.g. `http://127.0.0.1:1234`). The daemon derives its WS
/// endpoint internally via `derive_daemon_ws_url`.
pub async fn spawn(daemon_id: &str, registry_base_url: &str, token: &str) -> Self {
Self::spawn_with(
daemon_id,
registry_base_url,
token,
DaemonOptions::default(),
)
.await
/// Spawn a fresh daemon on a free localhost port.
pub async fn spawn() -> Self {
Self::spawn_at(SocketAddr::from((Ipv4Addr::LOCALHOST, free_port()))).await
}
/// Spawn with NAP-247 options (perimeter advertisement).
pub async fn spawn_with(
daemon_id: &str,
registry_base_url: &str,
token: &str,
opts: DaemonOptions,
) -> Self {
let port = free_port();
let bind = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
Self::spawn_with_port_and_opts(daemon_id, registry_base_url, token, bind, opts).await
/// Spawn on a caller-supplied bind address.
pub async fn spawn_at(bind: SocketAddr) -> Self {
let home = TempDir::new().expect("home tmpdir");
let tmux = TempDir::new().expect("tmux tmpdir");
let child = launch(bind, home.path(), tmux.path());
let me = Self {
bind_addr: bind,
child: Some(child),
home,
tmux,
};
me.wait_until_ready().await;
me
}
/// Spawn on a caller-supplied bind address. Used by NAP-247 broker tests
/// where the daemon's advertised `private_addr` must equal the actual
/// bind address so the broker can dial it back.
pub async fn spawn_at(
daemon_id: &str,
registry_base_url: &str,
token: &str,
bind: SocketAddr,
opts: DaemonOptions,
) -> Self {
Self::spawn_with_port_and_opts(daemon_id, registry_base_url, token, bind, opts).await
/// HTTP base URL of this daemon's API.
pub fn http_base(&self) -> String {
format!("http://{}", self.bind_addr)
}
/// WS base URL of this daemon's API.
pub fn ws_base(&self) -> String {
format!("ws://{}", self.bind_addr)
}
/// Stop the underlying process, but keep its state dirs around so we can
@@ -85,70 +66,18 @@ impl Daemon {
/// Re-spawn the daemon on the same port + state. The previous tmux
/// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
/// its sessions. Restarts inherit the original options via the on-disk
/// config file written at `spawn_with_port_and_opts` time.
/// its sessions.
pub async fn restart(&mut self) {
self.stop();
let child = launch(
&self.daemon_id,
self.bind_addr,
&self.daemon_url,
&self.registry_url,
&self.token,
self.home.path(),
self.state.path(),
self.tmux.path(),
&DaemonOptions::default(),
);
self.child = Some(child);
self.wait_until_attach_ready().await;
self.child = Some(launch(self.bind_addr, self.home.path(), self.tmux.path()));
self.wait_until_ready().await;
}
async fn spawn_with_port_and_opts(
daemon_id: &str,
registry_url: &str,
token: &str,
bind: SocketAddr,
opts: DaemonOptions,
) -> Self {
let home = TempDir::new().expect("home tmpdir");
let state = TempDir::new().expect("state tmpdir");
let tmux = TempDir::new().expect("tmux tmpdir");
let daemon_url = format!("ws://{bind}");
let child = launch(
daemon_id,
bind,
&daemon_url,
registry_url,
token,
home.path(),
state.path(),
tmux.path(),
&opts,
);
let me = Self {
daemon_id: daemon_id.to_string(),
bind_addr: bind,
daemon_url,
child: Some(child),
home,
state,
tmux,
registry_url: registry_url.to_string(),
token: token.to_string(),
};
me.wait_until_attach_ready().await;
me
}
async fn wait_until_attach_ready(&self) {
async fn wait_until_ready(&self) {
let addr = self.bind_addr;
wait_for_async(
"daemon attach port",
Duration::from_secs(5),
|| async move { tokio::net::TcpStream::connect(addr).await.is_ok() },
)
wait_for_async("daemon api port", Duration::from_secs(5), || async move {
tokio::net::TcpStream::connect(addr).await.is_ok()
})
.await;
}
@@ -158,7 +87,7 @@ impl Daemon {
pub fn state_dir(&self) -> PathBuf {
// Matches maestro_daemon::default_state_dir which uses XDG_STATE_HOME.
self.home.path().join(".local/state/maestrod")
self.home.path().join(".local/state/maestro")
}
}
@@ -171,18 +100,7 @@ impl Drop for Daemon {
}
}
#[allow(clippy::too_many_arguments)]
fn launch(
daemon_id: &str,
bind: SocketAddr,
daemon_url: &str,
registry_url: &str,
token: &str,
home: &std::path::Path,
_state: &std::path::Path,
tmux_tmpdir: &std::path::Path,
opts: &DaemonOptions,
) -> Child {
fn launch(bind: SocketAddr, home: &std::path::Path, tmux_tmpdir: &std::path::Path) -> Child {
let bin = workspace_target_bin("maestro", "maestro");
let mut cmd = Command::new(&bin);
cmd.arg("daemon");
@@ -195,21 +113,11 @@ fn launch(
.env("XDG_CONFIG_HOME", home.join(".config"))
.env("XDG_STATE_HOME", home.join(".local/state"))
.env("TMUX_TMPDIR", tmux_tmpdir)
.env("MAESTRO_TOKEN", token)
.env("MAESTRO_DAEMON_ID", daemon_id)
.env("MAESTRO_REGISTRY_URL", registry_url)
.env("MAESTRO_DAEMON_BIND_ADDR", bind.to_string())
.env("MAESTRO_DAEMON_URL", daemon_url)
.env("RUST_LOG", "warn")
.stdout(Stdio::null())
.stderr(Stdio::inherit())
.arg("run");
if let Some(p) = &opts.perimeter_id {
cmd.env("MAESTRO_DAEMON_PERIMETER_ID", p);
}
if let Some(a) = &opts.private_addr {
cmd.env("MAESTRO_DAEMON_PRIVATE_ADDR", a);
}
cmd.spawn()
.unwrap_or_else(|e| panic!("spawn {} failed: {e}", bin.display()))
}
crates/maestro-e2e/src/lib.rs
+4 -6
View File
@@ -1,8 +1,8 @@
//! Cross-process integration test harness for Maestro.
//!
//! Each test spawns a real `maestro-registry` and one or more `maestrod`
//! subprocesses, isolates them in temp dirs (`$HOME`, `TMUX_TMPDIR`, state
//! dir), and drives them through their public HTTP/WS APIs.
//! Each test spawns a real `maestro daemon` subprocess, isolates it in temp
//! dirs (`$HOME`, `TMUX_TMPDIR`, state dir), and drives it through its public
//! HTTP/WS API.
//!
//! See `tests/` for the actual scenarios.
@@ -11,10 +11,8 @@
pub mod build;
pub mod client;
pub mod daemon;
pub mod registry;
pub mod util;
pub use build::workspace_target_bin;
pub use client::TestClient;
pub use daemon::{Daemon, DaemonOptions};
pub use registry::Registry;
pub use daemon::Daemon;
crates/maestro-e2e/tests/e2e.rs
+58 -554
View File
@@ -1,9 +1,8 @@
//! Cross-process end-to-end tests for the headless Maestro slice.
//!
//! Each test brings up a real `maestro-registry` plus 1-2 real `maestrod`
//! subprocesses, isolates them in temp dirs, and drives them through the
//! public HTTP+WS surface using the same client APIs a real CLI/web client
//! would.
//! Each test brings up a real daemon subprocess, isolated in temp dirs, and
//! drives it through the public HTTP+WS surface using the same client APIs a
//! real CLI/web client would.
//!
//! These tests require `tmux` and `git` on `$PATH`.
@@ -15,45 +14,19 @@ use std::time::{Duration, Instant};
use futures_util::{SinkExt, StreamExt};
use maestro_e2e::util::wait_for_async;
use maestro_e2e::{Daemon, DaemonOptions, Registry, TestClient};
use maestro_e2e::{Daemon, TestClient};
use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
use tempfile::TempDir;
use tokio_tungstenite::tungstenite::Message as WsMessage;
const TOKEN: &str = "e2e-shared-token";
async fn bring_up(daemon_ids: &[&str]) -> (Registry, Vec<Daemon>, TestClient) {
let registry = Registry::spawn(TOKEN).await;
let mut daemons = Vec::with_capacity(daemon_ids.len());
for id in daemon_ids {
let d = Daemon::spawn(id, &registry.http_base(), TOKEN).await;
daemons.push(d);
}
let client = TestClient::new(&registry.http_base(), &registry.ws_base());
client.bootstrap().await.expect("bootstrap");
// Wait until both daemons have connected and announced themselves.
let expected: Vec<String> = daemon_ids.iter().map(|s| s.to_string()).collect();
wait_for_async("daemons to register", Duration::from_secs(10), || {
let client = client.clone();
let expected = expected.clone();
async move {
match client.list_daemons().await {
Ok(ds) => expected
.iter()
.all(|id| ds.iter().any(|d| &d.daemon_id == id)),
Err(_) => false,
}
}
})
.await;
(registry, daemons, client)
async fn bring_up() -> (Daemon, TestClient) {
let daemon = Daemon::spawn().await;
let client = TestClient::new(&daemon.http_base(), &daemon.ws_base());
(daemon, client)
}
fn create_req(daemon_id: &str, cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
CreateSessionRequest {
daemon_id: daemon_id.into(),
cwd: cwd.to_path_buf(),
command: command.into(),
env: HashMap::new(),
@@ -63,211 +36,39 @@ fn create_req(daemon_id: &str, cwd: &std::path::Path, command: &str) -> CreateSe
}
#[tokio::test(flavor = "multi_thread")]
async fn lists_sessions_from_both_daemons() {
let (_reg, _daemons, client) = bring_up(&["alpha", "beta"]).await;
async fn lists_spawned_sessions() {
let (_daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let a = client
.spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
.spawn_session(&create_req(cwd.path(), "sleep 30"))
.await
.expect("spawn alpha");
.expect("spawn a");
let b = client
.spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
.spawn_session(&create_req(cwd.path(), "sleep 30"))
.await
.expect("spawn beta");
.expect("spawn b");
let sessions = client.list_sessions().await.expect("list");
assert!(sessions
.iter()
.any(|s| s.id == a.id && s.daemon_id == "alpha"));
assert!(sessions
.iter()
.any(|s| s.id == b.id && s.daemon_id == "beta"));
assert!(sessions.iter().any(|s| s.id == a.id));
assert!(sessions.iter().any(|s| s.id == b.id));
client.kill_session(a.id).await.expect("kill a");
client.kill_session(b.id).await.expect("kill b");
}
#[tokio::test(flavor = "multi_thread")]
async fn daemon_local_api_mirrors_peer_catalog() {
// NAP-246: a daemon's GET /api/daemons + /api/sessions should reflect
// peer state (not just self), because the registry fans out catalog
// events to every connected daemon.
let (_reg, daemons, client) = bring_up(&["alpha", "beta"]).await;
let cwd = TempDir::new().unwrap();
// Create a session on beta via the registry path.
let b = client
.spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
.await
.expect("spawn beta");
// Now query alpha's local HTTP API and confirm it sees beta + beta's
// session. The mirror is eventually consistent, so poll.
let alpha_addr = daemons[0].bind_addr;
let alpha_base = format!("http://{alpha_addr}");
wait_for_async(
"alpha mirrors beta in /api/daemons",
Duration::from_secs(5),
|| {
let base = alpha_base.clone();
async move {
let resp = match reqwest::get(format!("{base}/api/daemons")).await {
Ok(r) => r,
Err(_) => return false,
};
let list: Vec<maestro_protocol::DaemonInfo> = match resp.json().await {
Ok(v) => v,
Err(_) => return false,
};
list.iter().any(|d| d.daemon_id == "beta")
}
},
)
.await;
wait_for_async(
"alpha mirrors beta's session in /api/sessions",
Duration::from_secs(5),
|| {
let base = alpha_base.clone();
let want = b.id;
async move {
let resp = match reqwest::get(format!("{base}/api/sessions")).await {
Ok(r) => r,
Err(_) => return false,
};
let list: Vec<maestro_protocol::Session> = match resp.json().await {
Ok(v) => v,
Err(_) => return false,
};
list.iter().any(|s| s.id == want && s.daemon_id == "beta")
}
},
)
.await;
// Both self and peer come back.
let list: Vec<maestro_protocol::DaemonInfo> = reqwest::get(format!("{alpha_base}/api/daemons"))
.await
.unwrap()
.json()
.await
.unwrap();
let ids: Vec<&str> = list.iter().map(|d| d.daemon_id.as_str()).collect();
assert!(ids.contains(&"alpha"));
assert!(ids.contains(&"beta"));
// NAP-328: home_dir is advertised end-to-end — for a peer relayed through
// the registry (beta, via Hello → register → catalog mirror), so the web
// spawn form can prefill the working directory.
let beta = list.iter().find(|d| d.daemon_id == "beta").unwrap();
assert!(
!beta.home_dir.is_empty(),
"peer home_dir should propagate through the registry: {beta:?}"
);
client.kill_session(b.id).await.expect("kill b");
}
#[tokio::test(flavor = "multi_thread")]
async fn home_daemon_brokers_cross_daemon_create_and_kill() {
// NAP-327: POST /api/sessions to alpha (home) with daemon_id=beta must
// broker the create to beta over the registry control link and return
// beta's session — and a kill via alpha must reach beta too. This is the
// path maestro-web uses (it only talks to its local daemon).
let (_reg, daemons, registry_client) = bring_up(&["alpha", "beta"]).await;
let cwd = TempDir::new().unwrap();
// A client pointed at alpha's *local* API (not the registry).
let alpha_base = format!("http://{}", daemons[0].bind_addr);
let alpha = TestClient::new(&alpha_base, &alpha_base);
// Create a beta-owned session by asking alpha.
let s = alpha
.spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
.await
.expect("alpha brokers create to beta");
assert_eq!(s.daemon_id, "beta", "session should be owned by beta");
// The registry catalog (and thus every client) sees it on beta.
wait_for_async(
"beta session visible in catalog",
Duration::from_secs(5),
|| {
let client = registry_client.clone();
let want = s.id;
async move {
client
.list_sessions()
.await
.map(|ss| ss.iter().any(|x| x.id == want && x.daemon_id == "beta"))
.unwrap_or(false)
}
},
)
.await;
// alpha's mirror must reflect beta's session before it can broker the kill.
wait_for_async(
"alpha mirrors beta's session",
Duration::from_secs(5),
|| {
let base = alpha_base.clone();
let want = s.id;
async move {
let resp = match reqwest::get(format!("{base}/api/sessions")).await {
Ok(r) => r,
Err(_) => return false,
};
let list: Vec<maestro_protocol::Session> = match resp.json().await {
Ok(v) => v,
Err(_) => return false,
};
list.iter().any(|x| x.id == want)
}
},
)
.await;
// Kill the beta session via alpha — brokered to beta.
alpha
.kill_session(s.id)
.await
.expect("alpha brokers kill to beta");
wait_for_async(
"beta session gone from catalog",
Duration::from_secs(5),
|| {
let client = registry_client.clone();
let gone = s.id;
async move {
client
.list_sessions()
.await
.map(|ss| !ss.iter().any(|x| x.id == gone))
.unwrap_or(false)
}
},
)
.await;
}
#[tokio::test(flavor = "multi_thread")]
async fn multi_attach_sees_same_output() {
let (_reg, _daemons, client) = bring_up(&["solo"]).await;
let (_daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let s = client
.spawn_session(&create_req("solo", cwd.path(), "bash --noprofile --norc"))
.spawn_session(&create_req(cwd.path(), "bash --noprofile --norc"))
.await
.expect("spawn");
let info = client.connect_info(s.id).await.expect("connect-info");
let mut a = client.attach(&info, s.id).await.expect("attach a");
let mut b = client.attach(&info, s.id).await.expect("attach b");
let mut a = client.attach(s.id).await.expect("attach a");
let mut b = client.attach(s.id).await.expect("attach b");
// Drive input through client A; both attachments should observe it via
// the daemon's PTY broadcast.
@@ -314,13 +115,12 @@ async fn multi_attach_sees_same_output() {
#[tokio::test(flavor = "multi_thread")]
async fn cleanup_on_kill_removes_worktree() {
let (_reg, _daemons, client) = bring_up(&["host"]).await;
let (_daemon, client) = bring_up().await;
let base = TempDir::new().unwrap();
init_repo(base.path());
let req = CreateSessionRequest {
daemon_id: "host".into(),
cwd: base.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -362,52 +162,20 @@ async fn cleanup_on_kill_removes_worktree() {
#[tokio::test(flavor = "multi_thread")]
async fn daemon_restart_recovers_session() {
let (_reg, mut daemons, client) = bring_up(&["restarter"]).await;
let (mut daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let s = client
.spawn_session(&create_req("restarter", cwd.path(), "sleep 300"))
.spawn_session(&create_req(cwd.path(), "sleep 300"))
.await
.expect("spawn");
assert_eq!(s.daemon_id, "restarter");
// Stop the daemon and wait for the registry to notice the disconnect.
// The catalog drops sessions for a daemon that's no longer connected.
daemons[0].stop();
wait_for_async(
"registry drops disconnected daemon",
Duration::from_secs(10),
|| {
let client = client.clone();
async move {
client
.list_daemons()
.await
.map(|ds| ds.iter().all(|d| d.daemon_id != "restarter"))
.unwrap_or(false)
}
},
)
.await;
// Bring the daemon back. It should recover its tmux session and the
// registry catalog should repopulate it.
daemons[0].restart().await;
wait_for_async("restarter re-registers", Duration::from_secs(10), || {
let client = client.clone();
async move {
client
.list_daemons()
.await
.map(|ds| ds.iter().any(|d| d.daemon_id == "restarter"))
.unwrap_or(false)
}
})
.await;
// Restart against the same tmux server + state dir: the daemon must adopt
// the still-running tmux session on boot.
daemon.restart().await;
wait_for_async(
"session reappears in catalog",
"session reappears after restart",
Duration::from_secs(10),
|| {
let client = client.clone();
@@ -428,7 +196,7 @@ async fn daemon_restart_recovers_session() {
#[tokio::test(flavor = "multi_thread")]
async fn concurrent_spawns_each_get_unique_ids() {
let (_reg, _daemons, client) = bring_up(&["parallel"]).await;
let (_daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let mut handles = Vec::new();
@@ -437,7 +205,7 @@ async fn concurrent_spawns_each_get_unique_ids() {
let path = cwd.path().to_path_buf();
handles.push(tokio::spawn(async move {
client
.spawn_session(&create_req("parallel", &path, "sleep 30"))
.spawn_session(&create_req(&path, "sleep 30"))
.await
.expect("spawn")
}));
@@ -461,8 +229,8 @@ async fn spawn_records_history_and_infers_gitroot_for_form() {
// the daemon's recent-dirs endpoint surfaces on the spawn form's next open
// (#27's recents picker), and gitroot infers the worktree base_repo from the
// same dir. Exercised end to end against real tmux + git.
let (_reg, daemons, client) = bring_up(&["host"]).await;
let daemon_base = format!("http://{}", daemons[0].bind_addr);
let (daemon, client) = bring_up().await;
let daemon_base = daemon.http_base();
// A git repo as the working dir, so gitroot has a repo to infer.
let repo = TempDir::new().unwrap();
@@ -471,7 +239,7 @@ async fn spawn_records_history_and_infers_gitroot_for_form() {
// Spawn in that dir (the command stands in for a preset tile selection).
let s = client
.spawn_session(&create_req("host", repo.path(), "sleep 30"))
.spawn_session(&create_req(repo.path(), "sleep 30"))
.await
.expect("spawn");
@@ -482,7 +250,7 @@ async fn spawn_records_history_and_infers_gitroot_for_form() {
"recent-dirs surfaces the spawned cwd",
Duration::from_secs(5),
|| {
let url = format!("{daemon_base}/api/daemons/host/recent-dirs");
let url = format!("{daemon_base}/api/recent-dirs");
let want = cwd.clone();
async move {
let resp = match reqwest::get(&url).await {
@@ -505,7 +273,7 @@ async fn spawn_records_history_and_infers_gitroot_for_form() {
// gitroot infers the worktree base_repo from the same dir (#25/#26): the dir
// is itself the repo toplevel (it holds `.git`).
let gr: maestro_protocol::GitRoot = reqwest::Client::new()
.get(format!("{daemon_base}/api/daemons/host/gitroot"))
.get(format!("{daemon_base}/api/gitroot"))
.query(&[("path", cwd.as_str())])
.send()
.await
@@ -530,15 +298,15 @@ async fn activity_reflects_foreground_and_input_blocking() {
// (the distinction the /proc/<pid>/syscall check buys us). Both report the
// right foreground. Exercised against real tmux.
use maestro_protocol::ActivityState;
let (_reg, _daemons, client) = bring_up(&["host"]).await;
let (_daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let cat = client
.spawn_session(&create_req("host", cwd.path(), "cat"))
.spawn_session(&create_req(cwd.path(), "cat"))
.await
.expect("spawn cat");
let slp = client
.spawn_session(&create_req("host", cwd.path(), "sleep 300"))
.spawn_session(&create_req(cwd.path(), "sleep 300"))
.await
.expect("spawn sleep");
@@ -621,244 +389,14 @@ fn git_worktree_list(dir: &std::path::Path) -> Vec<String> {
.collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn broker_t2_attach_through_home_daemon() {
// NAP-247: two daemons share perimeter "wg-test". alpha's broker resolves
// a session owned by beta, gets a pair JWT + perimeter address from the
// registry, dials beta via T2, and proxies the attach.
let registry = Registry::spawn("e2e-shared-token").await;
let beta_bind = std::net::SocketAddr::from((
std::net::Ipv4Addr::LOCALHOST,
maestro_e2e::util::free_port(),
));
let beta = Daemon::spawn_at(
"beta",
&registry.http_base(),
"e2e-shared-token",
beta_bind,
DaemonOptions {
perimeter_id: Some("wg-test".into()),
private_addr: Some(beta_bind.to_string()),
},
)
.await;
let alpha_bind = std::net::SocketAddr::from((
std::net::Ipv4Addr::LOCALHOST,
maestro_e2e::util::free_port(),
));
let alpha = Daemon::spawn_at(
"alpha",
&registry.http_base(),
"e2e-shared-token",
alpha_bind,
DaemonOptions {
perimeter_id: Some("wg-test".into()),
private_addr: Some(alpha_bind.to_string()),
},
)
.await;
assert_eq!(beta.bind_addr, beta_bind);
assert_eq!(alpha.bind_addr, alpha_bind);
let client = TestClient::new(&registry.http_base(), &registry.ws_base());
client.bootstrap().await.expect("bootstrap");
wait_for_async("both daemons connected", Duration::from_secs(10), || {
let client = client.clone();
async move {
client
.list_daemons()
.await
.map(|ds| {
ds.iter().any(|d| d.daemon_id == "alpha")
&& ds.iter().any(|d| d.daemon_id == "beta")
})
.unwrap_or(false)
}
})
.await;
let cwd = TempDir::new().unwrap();
let session = client
.spawn_session(&create_req("beta", cwd.path(), "bash --noprofile --norc"))
.await
.expect("spawn beta");
// Wait until alpha's catalog mirror sees the peer session — otherwise the
// broker can't locate the owning daemon.
let alpha_base = format!("http://{}", alpha.bind_addr);
let sid = session.id;
wait_for_async(
"alpha mirrors beta's session",
Duration::from_secs(5),
|| {
let base = alpha_base.clone();
async move {
let resp = match reqwest::get(format!("{base}/api/sessions")).await {
Ok(r) => r,
Err(_) => return false,
};
let list: Vec<maestro_protocol::Session> = match resp.json().await {
Ok(v) => v,
Err(_) => return false,
};
list.iter().any(|s| s.id == sid && s.daemon_id == "beta")
}
},
)
.await;
// Attach through alpha's broker.
let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid);
let (mut ws, _) = tokio_tungstenite::connect_async(broker_url)
.await
.expect("broker WS handshake");
ws.send(WsMessage::Binary(b"echo MAESTRO_BROKER_E2E\n".to_vec()))
.await
.expect("send keys");
let want = b"MAESTRO_BROKER_E2E";
let mut buf = Vec::new();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while std::time::Instant::now() < deadline {
tokio::select! {
msg = ws.next() => match msg {
Some(Ok(WsMessage::Binary(b))) => buf.extend(b),
Some(Ok(_)) => {}
_ => break,
},
_ = tokio::time::sleep(Duration::from_millis(50)) => {}
}
if buf.windows(want.len()).any(|w| w == want) {
break;
}
}
assert!(
buf.windows(want.len()).any(|w| w == want),
"broker WS never echoed our keystrokes: buf len {}",
buf.len()
);
let _ = ws.close(None).await;
client.kill_session(sid).await.expect("kill");
drop(alpha);
drop(beta);
drop(registry);
}
#[tokio::test(flavor = "multi_thread")]
async fn broker_attach_returns_502_on_perimeter_mismatch() {
// NAP-270: the asymmetric-VLAN boundary. alpha and beta are on DIFFERENT
// perimeters, so alpha's broker must refuse T2 (the only implemented
// transport) with a structured 502 rather than dial across perimeters.
let registry = Registry::spawn("e2e-shared-token").await;
let beta_bind = std::net::SocketAddr::from((
std::net::Ipv4Addr::LOCALHOST,
maestro_e2e::util::free_port(),
));
let beta = Daemon::spawn_at(
"beta",
&registry.http_base(),
"e2e-shared-token",
beta_bind,
DaemonOptions {
perimeter_id: Some("wg-beta".into()),
private_addr: Some(beta_bind.to_string()),
},
)
.await;
let alpha_bind = std::net::SocketAddr::from((
std::net::Ipv4Addr::LOCALHOST,
maestro_e2e::util::free_port(),
));
let alpha = Daemon::spawn_at(
"alpha",
&registry.http_base(),
"e2e-shared-token",
alpha_bind,
DaemonOptions {
// Different perimeter from beta → T2 unavailable.
perimeter_id: Some("wg-alpha".into()),
private_addr: Some(alpha_bind.to_string()),
},
)
.await;
let client = TestClient::new(&registry.http_base(), &registry.ws_base());
client.bootstrap().await.expect("bootstrap");
wait_for_async("both daemons connected", Duration::from_secs(10), || {
let client = client.clone();
async move {
client
.list_daemons()
.await
.map(|ds| ds.len() >= 2)
.unwrap_or(false)
}
})
.await;
let cwd = TempDir::new().unwrap();
let session = client
.spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
.await
.expect("spawn beta");
let sid = session.id;
// Wait until alpha's catalog mirror sees beta's session.
let alpha_base = format!("http://{}", alpha.bind_addr);
wait_for_async(
"alpha mirrors beta's session",
Duration::from_secs(5),
|| {
let base = alpha_base.clone();
async move {
let resp = match reqwest::get(format!("{base}/api/sessions")).await {
Ok(r) => r,
Err(_) => return false,
};
let list: Vec<maestro_protocol::Session> = match resp.json().await {
Ok(v) => v,
Err(_) => return false,
};
list.iter().any(|s| s.id == sid)
}
},
)
.await;
// Attach through alpha's broker — must 502 (perimeter mismatch), never
// upgrade and never dial beta.
let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid);
let err = tokio_tungstenite::connect_async(broker_url)
.await
.expect_err("cross-perimeter attach must fail the handshake");
match err {
tokio_tungstenite::tungstenite::Error::Http(resp) => {
assert_eq!(
resp.status(),
502,
"expected BAD_GATEWAY for perimeter mismatch, got {}",
resp.status()
);
}
other => panic!("expected an HTTP 502 response, got {other:?}"),
}
client.kill_session(sid).await.expect("kill");
drop(alpha);
drop(beta);
drop(registry);
}
#[tokio::test(flavor = "multi_thread")]
async fn daemon_events_ws_streams_snapshot_then_session_added() {
// NAP-248: subscribe to a daemon's WS /api/events, expect an initial
// NAP-248: subscribe to the daemon's WS /api/events, expect an initial
// Snapshot, then a SessionAdded after we spawn a session via REST.
let (_reg, daemons, client) = bring_up(&["alpha"]).await;
let (daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let alpha_addr = daemons[0].bind_addr;
let ws_url = format!("ws://{alpha_addr}/api/events");
let ws_url = format!("{}/api/events", daemon.ws_base());
let (mut ws, _) = tokio_tungstenite::connect_async(ws_url)
.await
.expect("events WS handshake");
@@ -866,23 +404,22 @@ async fn daemon_events_ws_streams_snapshot_then_session_added() {
// First frame must be a Snapshot (the WS sends it eagerly on connect).
let snapshot = next_event(&mut ws).await;
match &snapshot {
maestro_protocol::RegistryEvent::Snapshot { daemons, sessions } => {
assert!(daemons.iter().any(|d| d.daemon_id == "alpha"));
maestro_protocol::CatalogEvent::Snapshot { sessions } => {
assert!(sessions.is_empty());
}
other => panic!("expected initial Snapshot, got {other:?}"),
}
let session = client
.spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
.spawn_session(&create_req(cwd.path(), "sleep 30"))
.await
.expect("spawn alpha");
.expect("spawn");
// SessionAdded must show up on the WS within a reasonable window.
let saw_added = tokio::time::timeout(Duration::from_secs(5), async {
loop {
match next_event(&mut ws).await {
maestro_protocol::RegistryEvent::SessionAdded { session: s }
maestro_protocol::CatalogEvent::SessionAdded { session: s }
if s.id == session.id =>
{
return true;
@@ -902,7 +439,7 @@ async fn daemon_events_ws_streams_snapshot_then_session_added() {
}
// ---------------------------------------------------------------------------
// NAP-244 — CLI rewired through the home-daemon broker, with registry fallback
// CLI against the local daemon
// ---------------------------------------------------------------------------
/// Run the real `maestro` binary with an isolated env and capture its output.
@@ -921,18 +458,15 @@ fn run_cli(args: &[&str], envs: &[(&str, &str)]) -> std::process::Output {
}
#[tokio::test(flavor = "multi_thread")]
async fn cli_list_reads_from_home_daemon() {
// Broker path: point the CLI at the daemon's local API and DON'T configure
// a registry. Success proves the catalog read came from the home daemon —
// the registry path would have errored with NoRegistryUrl.
let (_reg, daemons, client) = bring_up(&["alpha"]).await;
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("alpha", cwd.path(), "sleep 30"))
.spawn_session(&create_req(cwd.path(), "sleep 30"))
.await
.expect("spawn");
let daemon_url = format!("http://{}", daemons[0].bind_addr);
let daemon_url = daemon.http_base();
let home = TempDir::new().unwrap();
let cfg_path = home.path().join("config.toml");
let out = run_cli(
@@ -940,7 +474,7 @@ async fn cli_list_reads_from_home_daemon() {
&[
("HOME", home.path().to_str().unwrap()),
("MAESTRO_CONFIG", cfg_path.to_str().unwrap()),
("MAESTRO_HOME_DAEMON_URL", &daemon_url),
("MAESTRO_DAEMON_URL", &daemon_url),
],
);
let stdout = String::from_utf8_lossy(&out.stdout);
@@ -948,65 +482,35 @@ async fn cli_list_reads_from_home_daemon() {
assert!(out.status.success(), "list failed: {stderr}\n{stdout}");
assert!(
stdout.contains(&session.id.to_string()),
"session not listed via home daemon: {stdout}"
);
assert!(
!stderr.contains("falling back"),
"unexpected registry fallback: {stderr}"
"session not listed: {stdout}"
);
client.kill_session(session.id).await.ok();
}
#[tokio::test(flavor = "multi_thread")]
async fn cli_list_falls_back_to_registry_when_daemon_down() {
// Compat path: home daemon URL points at a dead port (connection refused).
// The CLI must warn and fall back to the configured registry, still
// returning the session.
let (reg, _daemons, client) = bring_up(&["alpha"]).await;
let cwd = TempDir::new().unwrap();
let session = client
.spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
.await
.expect("spawn");
async fn cli_list_fails_when_daemon_is_down() {
let dead_daemon = format!("http://127.0.0.1:{}", maestro_e2e::util::free_port());
let registry_url = reg.http_base();
let home = TempDir::new().unwrap();
let cfg_path = home.path().join("config.toml");
let out = run_cli(
&["list", "--json"],
&["list"],
&[
("HOME", home.path().to_str().unwrap()),
("MAESTRO_CONFIG", cfg_path.to_str().unwrap()),
("MAESTRO_HOME_DAEMON_URL", &dead_daemon),
("MAESTRO_REGISTRY_URL", &registry_url),
("MAESTRO_DAEMON_URL", &dead_daemon),
],
);
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()),
"session not listed via registry fallback: {stdout}"
);
assert!(
stderr.contains("falling back"),
"expected a fallback warning on stderr: {stderr}"
);
client.kill_session(session.id).await.ok();
assert!(!out.status.success(), "expected a failure with no daemon");
}
async fn next_event<S>(ws: &mut S) -> maestro_protocol::RegistryEvent
async fn next_event<S>(ws: &mut S) -> maestro_protocol::CatalogEvent
where
S: StreamExt<Item = Result<WsMessage, tokio_tungstenite::tungstenite::Error>> + Unpin,
{
loop {
match ws.next().await {
Some(Ok(WsMessage::Text(t))) => {
return serde_json::from_str(&t).expect("RegistryEvent")
}
Some(Ok(WsMessage::Text(t))) => return serde_json::from_str(&t).expect("CatalogEvent"),
Some(Ok(_)) => continue,
other => panic!("WS closed waiting for event: {other:?}"),
}
crates/maestro-e2e/tests/tui_integration.rs
+12 -34
View File
@@ -1,11 +1,11 @@
//! TUI integration tests (NAP-282).
//!
//! Exercises `CatalogStore` and the per-session `SessionTask` against a real
//! registry + daemon subprocess pair. We don't spin the full ratatui frontend
//! 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 `maestrod`.
//! nicely with a real daemon.
//!
//! Requires `tmux` and `git` on `$PATH`.
@@ -14,26 +14,10 @@
use std::path::PathBuf;
use std::time::{Duration, Instant};
use maestro_e2e::{Daemon, Registry};
use maestro_e2e::Daemon;
use maestro_protocol::CreateSessionRequest;
use maestro_tui::catalog::CatalogStore;
const TOKEN: &str = "tui-integration-token";
async fn bring_up() -> (Registry, Daemon) {
let registry = Registry::spawn(TOKEN).await;
let daemon = Daemon::spawn("konishi", &registry.http_base(), TOKEN).await;
(registry, daemon)
}
fn http_base(daemon: &Daemon) -> String {
format!("http://{}", daemon.bind_addr)
}
fn ws_base(daemon: &Daemon) -> String {
format!("ws://{}", daemon.bind_addr)
}
async fn wait_until<F, Fut>(label: &str, deadline: Duration, mut check: F)
where
F: FnMut() -> Fut,
@@ -53,8 +37,8 @@ where
#[tokio::test(flavor = "multi_thread")]
async fn catalog_bootstrap_against_real_daemon() {
let (_registry, daemon) = bring_up().await;
let ws_url = format!("{}/api/events", ws_base(&daemon));
let daemon = Daemon::spawn().await;
let ws_url = format!("{}/api/events", daemon.ws_base());
let store = CatalogStore::default();
let bg = store.clone();
@@ -64,17 +48,14 @@ async fn catalog_bootstrap_against_real_daemon() {
.await
.expect("initial snapshot");
let snap = store.snapshot();
assert_eq!(snap.daemons.len(), 1, "self-daemon snapshot");
assert_eq!(snap.daemons[0].daemon_id, "konishi");
assert!(snap.sessions.is_empty());
assert!(store.snapshot().sessions.is_empty());
}
#[tokio::test(flavor = "multi_thread")]
async fn post_session_propagates_through_events_ws() {
let (_registry, daemon) = bring_up().await;
let ws_url = format!("{}/api/events", ws_base(&daemon));
let http_base = http_base(&daemon);
let daemon = Daemon::spawn().await;
let ws_url = format!("{}/api/events", daemon.ws_base());
let http_base = daemon.http_base();
let store = CatalogStore::default();
let bg = store.clone();
@@ -86,7 +67,6 @@ async fn post_session_propagates_through_events_ws() {
let cwd = PathBuf::from(daemon.home());
let req = CreateSessionRequest {
daemon_id: "konishi".into(),
cwd,
command: "bash".into(),
env: Default::default(),
@@ -110,15 +90,14 @@ async fn post_session_propagates_through_events_ws() {
let snap = store.snapshot();
assert_eq!(snap.sessions.len(), 1);
assert_eq!(snap.sessions[0].daemon_id, "konishi");
assert_eq!(snap.sessions[0].command, "bash");
}
#[tokio::test(flavor = "multi_thread")]
async fn delete_session_propagates_removal() {
let (_registry, daemon) = bring_up().await;
let ws_url = format!("{}/api/events", ws_base(&daemon));
let http_base = http_base(&daemon);
let daemon = Daemon::spawn().await;
let ws_url = format!("{}/api/events", daemon.ws_base());
let http_base = daemon.http_base();
let store = CatalogStore::default();
let bg = store.clone();
@@ -130,7 +109,6 @@ async fn delete_session_propagates_removal() {
let cwd = PathBuf::from(daemon.home());
let req = CreateSessionRequest {
daemon_id: "konishi".into(),
cwd,
command: "bash".into(),
env: Default::default(),
crates/maestro-tui/src/sidebar.rs
+1 -4
View File
@@ -119,10 +119,7 @@ mod tests {
#[test]
fn renders_session_names() {
let sb = Sidebar {
sessions: vec![
session_named(1, "claude"),
session_named(2, "bash"),
],
sessions: vec![session_named(1, "claude"), session_named(2, "bash")],
highlight: None,
attached: Default::default(),
};