From 5f6a5ced6b1bce7725a90108ddd295b06f00ed5f Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 5 Apr 2026 14:58:24 -0400 Subject: [PATCH] refactor(test): restructure fabro-server it/ into api/ and scenario/ subdirs Move integration tests from monolithic api.rs into api/ (single-endpoint contract tests) and scenario/ (multi-API-step flows), mirroring the CLI's cmd/ vs scenario/ pattern. Move 3 scheduler-dependent unit tests from server.rs into it/scenario/ where they get the correct nextest timeout (kind=test override). Deduplicate shared helpers into helpers.rs. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/crates/fabro-server/src/server.rs | 116 --- lib/crates/fabro-server/tests/it/api.rs | 909 ------------------ lib/crates/fabro-server/tests/it/api/mod.rs | 3 + lib/crates/fabro-server/tests/it/api/mtls.rs | 249 +++++ .../fabro-server/tests/it/api/routing.rs | 75 ++ lib/crates/fabro-server/tests/it/helpers.rs | 115 ++- lib/crates/fabro-server/tests/it/main.rs | 1 + .../fabro-server/tests/it/scenario/dry_run.rs | 69 ++ .../tests/it/scenario/lifecycle.rs | 184 ++++ .../fabro-server/tests/it/scenario/mod.rs | 5 + .../tests/it/scenario/run_completion.rs | 57 ++ .../fabro-server/tests/it/scenario/sse.rs | 118 +++ .../fabro-server/tests/it/scenario/usage.rs | 41 + 13 files changed, 916 insertions(+), 1026 deletions(-) delete mode 100644 lib/crates/fabro-server/tests/it/api.rs create mode 100644 lib/crates/fabro-server/tests/it/api/mod.rs create mode 100644 lib/crates/fabro-server/tests/it/api/mtls.rs create mode 100644 lib/crates/fabro-server/tests/it/api/routing.rs create mode 100644 lib/crates/fabro-server/tests/it/scenario/dry_run.rs create mode 100644 lib/crates/fabro-server/tests/it/scenario/lifecycle.rs create mode 100644 lib/crates/fabro-server/tests/it/scenario/mod.rs create mode 100644 lib/crates/fabro-server/tests/it/scenario/run_completion.rs create mode 100644 lib/crates/fabro-server/tests/it/scenario/sse.rs create mode 100644 lib/crates/fabro-server/tests/it/scenario/usage.rs diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index bae981174..95d83ae8f 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -2313,9 +2313,6 @@ mod tests { start -> exit }"#; - const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10); - const POLL_ATTEMPTS: usize = 500; - fn dry_run_settings() -> Settings { Settings { dry_run: Some(true), @@ -3135,72 +3132,6 @@ mod tests { assert_eq!(response.status(), StatusCode::NOT_FOUND); } - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn attach_run_events_returns_sse_stream() { - let state = create_app_state(); - let app = test_app_with_scheduler(state); - - let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await; - let run_id = run_id_str.parse::().unwrap(); - - // Wait for scheduler to promote run. - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/attach"))) - .body(Body::empty()) - .unwrap(); - - let response = app.oneshot(req).await.unwrap(); - let status = response.status(); - assert!( - status == StatusCode::OK || status == StatusCode::GONE, - "unexpected status: {status}" - ); - - if status == StatusCode::OK { - let content_type = response - .headers() - .get("content-type") - .expect("content-type header should be present") - .to_str() - .unwrap(); - assert!( - content_type.contains("text/event-stream"), - "expected text/event-stream, got: {content_type}" - ); - } - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn run_completes_and_status_is_completed() { - let state = create_app_state_with_options(dry_run_settings(), 5); - let app = test_app_with_scheduler(state); - - let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await; - let run_id = run_id_str.parse::().unwrap(); - - // Poll until run completes - let mut status = String::new(); - for _ in 0..POLL_ATTEMPTS { - tokio::time::sleep(POLL_INTERVAL).await; - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let body = body_json(response.into_body()).await; - status = body["status"].as_str().unwrap().to_string(); - if status == "succeeded" || status == "failed" { - break; - } - } - assert_eq!(status, "succeeded"); - } - #[tokio::test] async fn get_graph_returns_svg() { let state = create_app_state(); @@ -3373,53 +3304,6 @@ mod tests { assert!(body["by_model"].as_array().unwrap().is_empty()); } - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn aggregate_usage_increments_after_run_completes() { - let state = create_app_state_with_options(dry_run_settings(), 5); - let app = test_app_with_scheduler(state); - - let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await; - let run_id = run_id_str.parse::().unwrap(); - - // Poll until run completes - let mut status = String::new(); - for _ in 0..POLL_ATTEMPTS { - tokio::time::sleep(POLL_INTERVAL).await; - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - status = body["status"].as_str().unwrap().to_string(); - if status == "succeeded" || status == "failed" { - break; - } - } - assert_eq!(status, "succeeded"); - - let mut total_runs = 0; - for _ in 0..POLL_ATTEMPTS { - let req = Request::builder() - .method("GET") - .uri(api("/usage")) - .body(Body::empty()) - .unwrap(); - - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - - let body = body_json(response.into_body()).await; - total_runs = body["totals"]["runs"].as_i64().unwrap(); - if total_runs == 1 { - break; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - assert_eq!(total_runs, 1); - } - #[tokio::test] async fn post_runs_returns_submitted_status() { let state = create_app_state(); diff --git a/lib/crates/fabro-server/tests/it/api.rs b/lib/crates/fabro-server/tests/it/api.rs deleted file mode 100644 index e14cdab23..000000000 --- a/lib/crates/fabro-server/tests/it/api.rs +++ /dev/null @@ -1,909 +0,0 @@ -// =========================================================================== -// mTLS end-to-end tests -// =========================================================================== - -#![allow(clippy::absolute_paths)] - -fn api(path: &str) -> String { - format!("/api/v1{path}") -} - -// Skip on macOS: LibreSSL generates certs with extensions rustls rejects -#[cfg(target_os = "linux")] -mod mtls_e2e { - use super::api; - use std::path::{Path, PathBuf}; - use std::sync::Arc; - - use fabro_server::jwt_auth::{AuthMode, AuthStrategy}; - use fabro_server::server::{build_router, create_app_state}; - use fabro_server::server_config::TlsSettings; - use fabro_server::tls::{ClientAuth, build_rustls_config}; - use tokio::net::TcpListener; - - fn fixture_path(name: &str) -> PathBuf { - Path::new(env!("CARGO_MANIFEST_DIR")) - .join("tests/fixtures/mtls") - .join(name) - } - - fn fixture_pki() -> PkiPaths { - PkiPaths { - ca_cert: fixture_path("ca.crt"), - server_cert: fixture_path("server.crt"), - server_key: fixture_path("server.key"), - client_cert: fixture_path("client.crt"), - client_key: fixture_path("client.key"), - } - } - - struct PkiPaths { - ca_cert: PathBuf, - server_cert: PathBuf, - server_key: PathBuf, - client_cert: PathBuf, - client_key: PathBuf, - } - - /// Start a TLS server on a random port, returning the bound address. - async fn start_tls_server( - tls_settings: &TlsSettings, - client_auth: ClientAuth, - auth_mode: AuthMode, - ) -> std::net::SocketAddr { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - - let rustls_config = build_rustls_config(tls_settings, client_auth); - let tls_acceptor = tokio_rustls::TlsAcceptor::from(rustls_config); - - let state = create_app_state(); - let router = build_router(state, auth_mode); - - tokio::spawn(async move { - let _ = fabro_server::tls::serve_tls(listener, tls_acceptor, router).await; - }); - - addr - } - - /// Build a reqwest client with the given CA cert and optional client identity. - fn build_client( - ca_cert_path: &Path, - client_cert_path: Option<&Path>, - client_key_path: Option<&Path>, - ) -> reqwest::Client { - let ca_pem = std::fs::read(ca_cert_path).unwrap(); - let ca_cert = reqwest::tls::Certificate::from_pem(&ca_pem).unwrap(); - - let mut builder = reqwest::Client::builder() - .add_root_certificate(ca_cert) - .no_proxy() - .use_rustls_tls(); - - if let (Some(cert_path), Some(key_path)) = (client_cert_path, client_key_path) { - let cert_pem = std::fs::read(cert_path).unwrap(); - let key_pem = std::fs::read(key_path).unwrap(); - let mut identity_pem = cert_pem; - identity_pem.extend_from_slice(&key_pem); - let identity = reqwest::tls::Identity::from_pem(&identity_pem).unwrap(); - builder = builder.identity(identity); - } - - builder.build().unwrap() - } - - fn install_crypto_provider() { - let _ = rustls::crypto::ring::default_provider().install_default(); - } - - #[tokio::test] - async fn mtls_accepts_valid_client_cert() { - install_crypto_provider(); - let pki = fixture_pki(); - - let tls_settings = TlsSettings { - cert: pki.server_cert.clone(), - key: pki.server_key.clone(), - ca: pki.ca_cert.clone(), - }; - - let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]); - let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await; - - let client = build_client(&pki.ca_cert, Some(&pki.client_cert), Some(&pki.client_key)); - - let response = client - .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) - .send() - .await - .expect("request with valid client cert should succeed"); - - assert_eq!(response.status(), 200); - } - - #[tokio::test] - async fn mtls_only_rejects_wrong_ca_client_cert() { - install_crypto_provider(); - let pki = fixture_pki(); - - let tls_settings = TlsSettings { - cert: pki.server_cert.clone(), - key: pki.server_key.clone(), - ca: pki.ca_cert.clone(), - }; - - let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]); - let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await; - - let wrong_client_cert = fixture_path("wrong-client.crt"); - let wrong_client_key = fixture_path("wrong-client.key"); - - // Client trusts the REAL server CA, but presents a cert from the WRONG CA - let client = build_client( - &pki.ca_cert, - Some(&wrong_client_cert), - Some(&wrong_client_key), - ); - - let result = client - .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) - .send() - .await; - - // Server should reject the TLS handshake — the wrong CA client cert - // will cause a connection error (not an HTTP error) - assert!( - result.is_err(), - "request with wrong-CA client cert should fail at TLS level, but got: {:?}", - result.unwrap().status() - ); - } - - #[tokio::test] - async fn mtls_only_rejects_no_client_cert() { - install_crypto_provider(); - let pki = fixture_pki(); - - let tls_settings = TlsSettings { - cert: pki.server_cert.clone(), - key: pki.server_key.clone(), - ca: pki.ca_cert.clone(), - }; - - // mTLS is the ONLY strategy → client cert is required at TLS level - let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]); - let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await; - - // Client trusts the server CA but presents NO client cert - let client = build_client(&pki.ca_cert, None, None); - - let result = client - .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) - .send() - .await; - - // Server requires client cert → TLS handshake should fail - assert!( - result.is_err(), - "request without client cert should fail when mTLS is the only strategy, but got: {:?}", - result.unwrap().status() - ); - } - - fn fixture_jwt_keypair() -> (jsonwebtoken::EncodingKey, jsonwebtoken::DecodingKey) { - let private_pem = std::fs::read(fixture_path("jwt-ed25519-private.pem")).unwrap(); - let public_pem = std::fs::read(fixture_path("jwt-ed25519-public.pem")).unwrap(); - - let encoding = - jsonwebtoken::EncodingKey::from_ed_pem(&private_pem).expect("invalid private key"); - let decoding = - jsonwebtoken::DecodingKey::from_ed_pem(&public_pem).expect("invalid public key"); - (encoding, decoding) - } - - fn sign_jwt(key: &jsonwebtoken::EncodingKey, sub: &str) -> String { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs(); - let claims = serde_json::json!({ - "iss": "fabro-web", - "iat": now, - "exp": now + 60, - "sub": sub, - }); - let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::EdDSA); - jsonwebtoken::encode(&header, &claims, key).expect("failed to sign token") - } - - #[tokio::test] - async fn mtls_and_jwt_accepts_valid_jwt_without_client_cert() { - install_crypto_provider(); - let pki = fixture_pki(); - - let tls_settings = TlsSettings { - cert: pki.server_cert.clone(), - key: pki.server_key.clone(), - ca: pki.ca_cert.clone(), - }; - - let (encoding_key, decoding_key) = fixture_jwt_keypair(); - - // Both mTLS and JWT strategies; mTLS is optional since JWT is also present - let auth_mode = AuthMode::Strategies(vec![ - AuthStrategy::Mtls, - AuthStrategy::Jwt { - key: Arc::new(decoding_key), - validation: Arc::new(fabro_server::jwt_auth::jwt_validation()), - allowed_usernames: vec!["brynary".to_string()], - }, - ]); - let addr = start_tls_server(&tls_settings, ClientAuth::Optional, auth_mode).await; - - // Client trusts the server CA but presents NO client cert - let client = build_client(&pki.ca_cert, None, None); - - let token = sign_jwt(&encoding_key, "https://github.com/brynary"); - - let response = client - .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) - .bearer_auth(&token) - .send() - .await - .expect("request with valid JWT and no client cert should succeed"); - - assert_eq!( - response.status(), - 200, - "valid JWT should be accepted when strategies = [mtls, jwt]" - ); - } -} - -// =========================================================================== -// Full HTTP server lifecycle (TS Scenario 4) -// =========================================================================== - -mod server_lifecycle { - use super::api; - use std::sync::Arc; - use std::time::Duration; - - use axum::body::Body; - use axum::http::{Request, StatusCode}; - use fabro_interview::Interviewer; - use fabro_server::server::{build_router, create_app_state_with_registry_factory}; - use fabro_workflow::handler::HandlerRegistry; - use fabro_workflow::handler::agent::AgentHandler; - use fabro_workflow::handler::exit::ExitHandler; - use fabro_workflow::handler::human::HumanHandler; - use fabro_workflow::handler::start::StartHandler; - use tower::ServiceExt; - - fn gate_registry(interviewer: Arc) -> HandlerRegistry { - let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None))); - registry.register("start", Box::new(StartHandler)); - registry.register("exit", Box::new(ExitHandler)); - registry.register("agent", Box::new(AgentHandler::new(None))); - registry.register("human", Box::new(HumanHandler::new(interviewer))); - registry - } - - async fn body_json(body: Body) -> serde_json::Value { - let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); - serde_json::from_slice(&bytes).unwrap() - } - - const POLL_INTERVAL: Duration = Duration::from_millis(10); - const POLL_ATTEMPTS: usize = 500; - - async fn run_json(app: &axum::Router, run_id: &str) -> serde_json::Value { - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - body_json(response.into_body()).await - } - - async fn wait_for_question_id(app: &axum::Router, run_id: &str) -> String { - for _ in 0..POLL_ATTEMPTS { - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/questions"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - let arr = body["data"].as_array().unwrap(); - if let Some(question_id) = arr - .first() - .and_then(|item| item["id"].as_str()) - .map(ToOwned::to_owned) - { - return question_id; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - panic!("question should have appeared"); - } - - async fn wait_for_run_status(app: &axum::Router, run_id: &str, expected: &[&str]) -> String { - for _ in 0..POLL_ATTEMPTS { - let body = run_json(app, run_id).await; - let status = body["status"].as_str().unwrap().to_string(); - if expected.iter().any(|candidate| *candidate == status) { - return status; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - panic!("run {run_id} did not reach any of {expected:?}"); - } - - async fn wait_for_run_status_not_in( - app: &axum::Router, - run_id: &str, - unexpected: &[&str], - ) -> String { - for _ in 0..POLL_ATTEMPTS { - let body = run_json(app, run_id).await; - let status = body["status"].as_str().unwrap().to_string(); - if unexpected.iter().all(|candidate| *candidate != status) { - return status; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - panic!("run {run_id} stayed in {unexpected:?}"); - } - - const GATE_DOT: &str = r#"digraph GateTest { - graph [goal="Test gate"] - start [shape=Mdiamond] - exit [shape=Msquare] - work [shape=box, prompt="Do work"] - gate [shape=hexagon, type="human", label="Approve?"] - done [shape=box, prompt="Finish"] - revise [shape=box, prompt="Revise"] - - start -> work -> gate - gate -> done [label="[A] Approve"] - gate -> revise [label="[R] Revise"] - done -> exit - revise -> gate - }"#; - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn full_http_lifecycle_approve_and_complete() { - let state = create_app_state_with_registry_factory(gate_registry); - fabro_server::server::spawn_scheduler(Arc::clone(&state)); - let app = build_router( - Arc::clone(&state), - fabro_server::jwt_auth::AuthMode::Disabled, - ); - - // 1. Create run - let req = Request::builder() - .method("POST") - .uri(api("/runs")) - .header("content-type", "application/json") - .body(Body::from( - serde_json::to_string(&serde_json::json!({"dot_source": GATE_DOT})).unwrap(), - )) - .unwrap(); - - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::CREATED); - let body = body_json(response.into_body()).await; - let run_id = body["id"].as_str().unwrap().to_string(); - - // 1b. Start the run - let req = Request::builder() - .method("POST") - .uri(api(&format!("/runs/{run_id}/start"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - - // 2. Poll for question to appear (run goes start -> work -> gate, then blocks) - let question_id = wait_for_question_id(&app, &run_id).await; - - // 3. Submit answer selecting first option (Approve) - let req = Request::builder() - .method("POST") - .uri(api(&format!( - "/runs/{run_id}/questions/{question_id}/answer" - ))) - .header("content-type", "application/json") - .body(Body::from( - serde_json::to_string(&serde_json::json!({"value": "A"})).unwrap(), - )) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::NO_CONTENT); - - // 4. Poll until the run reaches a terminal success or failure state. - let final_status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; - assert_eq!(final_status, "succeeded"); - - // 5. Verify no pending questions - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/questions"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - assert!( - body["data"].as_array().unwrap().is_empty(), - "no pending questions after completion" - ); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn full_http_lifecycle_cancel() { - let state = create_app_state_with_registry_factory(gate_registry); - fabro_server::server::spawn_scheduler(Arc::clone(&state)); - let app = build_router( - Arc::clone(&state), - fabro_server::jwt_auth::AuthMode::Disabled, - ); - - // Create and start a run that will block at the human gate - let req = Request::builder() - .method("POST") - .uri(api("/runs")) - .header("content-type", "application/json") - .body(Body::from( - serde_json::to_string(&serde_json::json!({"dot_source": GATE_DOT})).unwrap(), - )) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - let run_id = body["id"].as_str().unwrap().to_string(); - - let req = Request::builder() - .method("POST") - .uri(api(&format!("/runs/{run_id}/start"))) - .body(Body::empty()) - .unwrap(); - app.clone().oneshot(req).await.unwrap(); - - // Subscribe as soon as the scheduler has created the live event stream. - // Waiting past "starting" races with stage events because `/events` - // subscribes to future broadcast messages only; it does not replay. - wait_for_run_status_not_in(&app, &run_id, &["queued"]).await; - - // Cancel it - let req = Request::builder() - .method("POST") - .uri(api(&format!("/runs/{run_id}/cancel"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let body = body_json(response.into_body()).await; - assert_eq!(body["status"], "cancelled"); - - // Verify the durable store view converges to cancelled failure. - let status = wait_for_run_status(&app, &run_id, &["failed"]).await; - assert_eq!(status, "failed"); - let body = run_json(&app, &run_id).await; - assert_eq!(body["status_reason"], "cancelled"); - } -} - -// =========================================================================== -// SSE event stream content parsing (TS Scenario 8) -// =========================================================================== - -mod sse_events { - use super::api; - use std::sync::Arc; - use std::time::Duration; - - use axum::body::Body; - use axum::http::{Request, StatusCode}; - use fabro_server::server::{build_router, create_app_state_with_options}; - use fabro_types::Settings; - use http_body_util::BodyExt; - use tower::ServiceExt; - - const SIMPLE_DOT: &str = r#"digraph SSETest { - graph [goal="Test SSE"] - start [shape=Mdiamond] - work [shape=box, prompt="Do work"] - exit [shape=Msquare] - start -> work -> exit - }"#; - - const POLL_INTERVAL: Duration = Duration::from_millis(10); - const POLL_ATTEMPTS: usize = 500; - - fn dry_run_settings() -> Settings { - Settings { - dry_run: Some(true), - ..Default::default() - } - } - - async fn body_json(body: Body) -> serde_json::Value { - let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); - serde_json::from_slice(&bytes).unwrap() - } - - async fn run_status(app: &axum::Router, run_id: &str) -> String { - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let body = body_json(response.into_body()).await; - body["status"].as_str().unwrap().to_string() - } - - async fn wait_for_run_status_not_in( - app: &axum::Router, - run_id: &str, - unexpected: &[&str], - ) -> String { - for _ in 0..POLL_ATTEMPTS { - let status = run_status(app, run_id).await; - if unexpected.iter().all(|candidate| *candidate != status) { - return status; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - panic!("run {run_id} stayed in {unexpected:?}"); - } - - async fn wait_for_checkpoint(app: &axum::Router, run_id: &str) -> serde_json::Value { - for _ in 0..POLL_ATTEMPTS { - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/checkpoint"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - if response.status() == StatusCode::OK { - return body_json(response.into_body()).await; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - panic!("checkpoint did not become available for {run_id}"); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn sse_stream_contains_expected_event_types() { - let state = create_app_state_with_options(dry_run_settings(), 5); - fabro_server::server::spawn_scheduler(Arc::clone(&state)); - let app = build_router( - Arc::clone(&state), - fabro_server::jwt_auth::AuthMode::Disabled, - ); - - // Start run - let req = Request::builder() - .method("POST") - .uri(api("/runs")) - .header("content-type", "application/json") - .body(Body::from( - serde_json::to_string(&serde_json::json!({"dot_source": SIMPLE_DOT})).unwrap(), - )) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::CREATED); - let body = body_json(response.into_body()).await; - let run_id = body["id"].as_str().unwrap().to_string(); - - wait_for_run_status_not_in(&app, &run_id, &["queued", "starting"]).await; - - // Get SSE stream - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}/attach"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - // May be 200 (stream open) or 410 (run completed before connect) - let sse_status = response.status(); - assert!( - sse_status == StatusCode::OK || sse_status == StatusCode::GONE, - "expected 200 or 410, got: {sse_status}" - ); - if sse_status == StatusCode::GONE { - return; - } - - let content_type = response - .headers() - .get("content-type") - .unwrap() - .to_str() - .unwrap(); - assert!(content_type.contains("text/event-stream")); - - // Collect SSE frames with a timeout - let mut body = response.into_body(); - let mut sse_data = String::new(); - while let Ok(Some(Ok(frame))) = - tokio::time::timeout(Duration::from_secs(2), body.frame()).await - { - if let Some(data) = frame.data_ref() { - sse_data.push_str(&String::from_utf8_lossy(data)); - } - } - - // Parse SSE data lines and extract event types - let mut event_types: Vec = Vec::new(); - for line in sse_data.lines() { - if let Some(json_str) = line.strip_prefix("data:") { - let json_str = json_str.trim(); - if let Ok(event) = serde_json::from_str::(json_str) { - if let Some(event_name) = event["payload"]["event"].as_str() { - event_types.push(event_name.to_string()); - } - } - } - } - - // Because we subscribe while the run is only guaranteed to be past - // "queued", a live stream should include at least one stage event. - // A 410 response above still covers the case where the run completed - // before we managed to attach. - if !event_types.is_empty() { - assert!( - event_types - .iter() - .any(|t| t == "stage.started" || t == "stage.completed"), - "should contain stage events, got: {event_types:?}" - ); - } - - // Pipeline is complete (SSE stream ended), verify checkpoint - let cp_body = wait_for_checkpoint(&app, &run_id).await; - // If run completed, checkpoint should have completed_nodes - if !cp_body.is_null() { - let completed = cp_body["completed_nodes"].as_array(); - if let Some(nodes) = completed { - let names: Vec<&str> = nodes.iter().filter_map(|v| v.as_str()).collect(); - assert!(names.contains(&"work"), "work should be in completed_nodes"); - } - } - } -} - -// =========================================================================== -// Serve command: dry-run registry factory builds a working router -// =========================================================================== - -mod serve_dry_run { - use super::api; - use std::sync::Arc; - use std::time::Duration; - - use axum::body::Body; - use axum::http::{Request, StatusCode}; - use fabro_server::server::{build_router, create_app_state_with_options}; - use tower::ServiceExt; - - const MINIMAL_DOT: &str = r#"digraph Test { - graph [goal="Test"] - start [shape=Mdiamond] - exit [shape=Msquare] - start -> exit - }"#; - - /// Build the router exactly as `serve_command` does in dry-run mode. - fn dry_run_app() -> axum::Router { - let state = create_app_state_with_options( - fabro_types::Settings { - dry_run: Some(true), - ..Default::default() - }, - 5, - ); - fabro_server::server::spawn_scheduler(Arc::clone(&state)); - build_router(state, fabro_server::jwt_auth::AuthMode::Disabled) - } - - async fn body_json(body: Body) -> serde_json::Value { - let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); - serde_json::from_slice(&bytes).unwrap() - } - - const POLL_INTERVAL: Duration = Duration::from_millis(10); - const POLL_ATTEMPTS: usize = 500; - - async fn wait_for_run_status(app: &axum::Router, run_id: &str, expected: &[&str]) -> String { - for _ in 0..POLL_ATTEMPTS { - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}"))) - .body(Body::empty()) - .unwrap(); - - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let body = body_json(response.into_body()).await; - let status = body["status"].as_str().unwrap().to_string(); - if expected.iter().any(|candidate| *candidate == status) { - return status; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - panic!("run {run_id} did not reach any of {expected:?}"); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn dry_run_serve_starts_and_runs_workflow() { - let app = dry_run_app(); - - // POST /runs to create a run - let req = Request::builder() - .method("POST") - .uri(api("/runs")) - .header("content-type", "application/json") - .body(Body::from( - serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(), - )) - .unwrap(); - - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::CREATED); - - let body = body_json(response.into_body()).await; - let run_id = body["id"].as_str().unwrap().to_string(); - assert!(!run_id.is_empty()); - - // POST /runs/{id}/start to queue it - let req = Request::builder() - .method("POST") - .uri(api(&format!("/runs/{run_id}/start"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - - let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; - assert_eq!(status, "succeeded"); - } - - #[tokio::test] - async fn test_model_known_via_full_router() { - let app = dry_run_app(); - - let req = Request::builder() - .method("POST") - .uri(api("/models/claude-opus-4-6/test")) - .header("content-type", "application/json") - .body(Body::empty()) - .unwrap(); - - let response = app.oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - - let body = body_json(response.into_body()).await; - assert_eq!(body["model_id"], "claude-opus-4-6"); - // No API keys in test env, so status will be "error" - assert!(body["status"] == "ok" || body["status"] == "error"); - } - - #[tokio::test] - async fn test_model_unknown_via_full_router() { - let app = dry_run_app(); - - let req = Request::builder() - .method("POST") - .uri(api("/models/nonexistent-model-xyz/test")) - .header("content-type", "application/json") - .body(Body::empty()) - .unwrap(); - - let response = app.oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::NOT_FOUND); - } - - #[tokio::test] - async fn dry_run_serve_rejects_invalid_dot() { - let app = dry_run_app(); - - let req = Request::builder() - .method("POST") - .uri(api("/runs")) - .header("content-type", "application/json") - .body(Body::from( - serde_json::to_string(&serde_json::json!({"dot_source": "not valid dot"})).unwrap(), - )) - .unwrap(); - - let response = app.oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::BAD_REQUEST); - } -} - -mod route_prefixes { - use axum::body::Body; - use axum::http::{Method, Request, StatusCode}; - use fabro_server::server::{build_router, create_app_state}; - use tower::ServiceExt; - - async fn body_json(body: Body) -> serde_json::Value { - let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); - serde_json::from_slice(&bytes).unwrap() - } - - #[tokio::test] - async fn old_unversioned_routes_return_404() { - let app = build_router( - create_app_state(), - fabro_server::jwt_auth::AuthMode::Disabled, - ); - - let cases = [(Method::POST, "/completions")]; - - for (method, path) in cases { - let req = Request::builder() - .method(method.clone()) - .uri(path) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::NOT_FOUND, "{method} {path}"); - } - } - - #[tokio::test] - async fn root_and_health_stay_at_root() { - let app = build_router( - create_app_state(), - fabro_server::jwt_auth::AuthMode::Disabled, - ); - - let root_req = Request::builder() - .method("GET") - .uri("/") - .body(Body::empty()) - .unwrap(); - let root_response = app.clone().oneshot(root_req).await.unwrap(); - assert_eq!(root_response.status(), StatusCode::OK); - let root_body = axum::body::to_bytes(root_response.into_body(), usize::MAX) - .await - .unwrap(); - let root_html = String::from_utf8(root_body.to_vec()).unwrap(); - assert!(root_html.contains("
")); - - let health_req = Request::builder() - .method("GET") - .uri("/health") - .body(Body::empty()) - .unwrap(); - let health_response = app.oneshot(health_req).await.unwrap(); - assert_eq!(health_response.status(), StatusCode::OK); - let health_body = body_json(health_response.into_body()).await; - assert_eq!(health_body["status"], "ok"); - } - - #[tokio::test] - async fn moved_routes_not_at_root_of_api_prefix() { - let app = build_router( - create_app_state(), - fabro_server::jwt_auth::AuthMode::Disabled, - ); - - for path in ["/api/v1/health", "/api/v1/"] { - let req = Request::builder() - .method("GET") - .uri(path) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::NOT_FOUND, "GET {path}"); - } - } -} diff --git a/lib/crates/fabro-server/tests/it/api/mod.rs b/lib/crates/fabro-server/tests/it/api/mod.rs new file mode 100644 index 000000000..d07c2fefc --- /dev/null +++ b/lib/crates/fabro-server/tests/it/api/mod.rs @@ -0,0 +1,3 @@ +#[cfg(target_os = "linux")] +mod mtls; +mod routing; diff --git a/lib/crates/fabro-server/tests/it/api/mtls.rs b/lib/crates/fabro-server/tests/it/api/mtls.rs new file mode 100644 index 000000000..9ae4ade9f --- /dev/null +++ b/lib/crates/fabro-server/tests/it/api/mtls.rs @@ -0,0 +1,249 @@ +#![allow(clippy::absolute_paths)] + +use crate::helpers::api; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use fabro_server::jwt_auth::{AuthMode, AuthStrategy}; +use fabro_server::server::{build_router, create_app_state}; +use fabro_server::server_config::TlsSettings; +use fabro_server::tls::{ClientAuth, build_rustls_config}; +use tokio::net::TcpListener; + +fn fixture_path(name: &str) -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("tests/fixtures/mtls") + .join(name) +} + +fn fixture_pki() -> PkiPaths { + PkiPaths { + ca_cert: fixture_path("ca.crt"), + server_cert: fixture_path("server.crt"), + server_key: fixture_path("server.key"), + client_cert: fixture_path("client.crt"), + client_key: fixture_path("client.key"), + } +} + +struct PkiPaths { + ca_cert: PathBuf, + server_cert: PathBuf, + server_key: PathBuf, + client_cert: PathBuf, + client_key: PathBuf, +} + +/// Start a TLS server on a random port, returning the bound address. +async fn start_tls_server( + tls_settings: &TlsSettings, + client_auth: ClientAuth, + auth_mode: AuthMode, +) -> std::net::SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let rustls_config = build_rustls_config(tls_settings, client_auth); + let tls_acceptor = tokio_rustls::TlsAcceptor::from(rustls_config); + + let state = create_app_state(); + let router = build_router(state, auth_mode); + + tokio::spawn(async move { + let _ = fabro_server::tls::serve_tls(listener, tls_acceptor, router).await; + }); + + addr +} + +/// Build a reqwest client with the given CA cert and optional client identity. +fn build_client( + ca_cert_path: &Path, + client_cert_path: Option<&Path>, + client_key_path: Option<&Path>, +) -> reqwest::Client { + let ca_pem = std::fs::read(ca_cert_path).unwrap(); + let ca_cert = reqwest::tls::Certificate::from_pem(&ca_pem).unwrap(); + + let mut builder = reqwest::Client::builder() + .add_root_certificate(ca_cert) + .no_proxy() + .use_rustls_tls(); + + if let (Some(cert_path), Some(key_path)) = (client_cert_path, client_key_path) { + let cert_pem = std::fs::read(cert_path).unwrap(); + let key_pem = std::fs::read(key_path).unwrap(); + let mut identity_pem = cert_pem; + identity_pem.extend_from_slice(&key_pem); + let identity = reqwest::tls::Identity::from_pem(&identity_pem).unwrap(); + builder = builder.identity(identity); + } + + builder.build().unwrap() +} + +fn install_crypto_provider() { + let _ = rustls::crypto::ring::default_provider().install_default(); +} + +#[tokio::test] +async fn mtls_accepts_valid_client_cert() { + install_crypto_provider(); + let pki = fixture_pki(); + + let tls_settings = TlsSettings { + cert: pki.server_cert.clone(), + key: pki.server_key.clone(), + ca: pki.ca_cert.clone(), + }; + + let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]); + let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await; + + let client = build_client(&pki.ca_cert, Some(&pki.client_cert), Some(&pki.client_key)); + + let response = client + .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) + .send() + .await + .expect("request with valid client cert should succeed"); + + assert_eq!(response.status(), 200); +} + +#[tokio::test] +async fn mtls_only_rejects_wrong_ca_client_cert() { + install_crypto_provider(); + let pki = fixture_pki(); + + let tls_settings = TlsSettings { + cert: pki.server_cert.clone(), + key: pki.server_key.clone(), + ca: pki.ca_cert.clone(), + }; + + let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]); + let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await; + + let wrong_client_cert = fixture_path("wrong-client.crt"); + let wrong_client_key = fixture_path("wrong-client.key"); + + // Client trusts the REAL server CA, but presents a cert from the WRONG CA + let client = build_client( + &pki.ca_cert, + Some(&wrong_client_cert), + Some(&wrong_client_key), + ); + + let result = client + .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) + .send() + .await; + + // Server should reject the TLS handshake — the wrong CA client cert + // will cause a connection error (not an HTTP error) + assert!( + result.is_err(), + "request with wrong-CA client cert should fail at TLS level, but got: {:?}", + result.unwrap().status() + ); +} + +#[tokio::test] +async fn mtls_only_rejects_no_client_cert() { + install_crypto_provider(); + let pki = fixture_pki(); + + let tls_settings = TlsSettings { + cert: pki.server_cert.clone(), + key: pki.server_key.clone(), + ca: pki.ca_cert.clone(), + }; + + // mTLS is the ONLY strategy -> client cert is required at TLS level + let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]); + let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await; + + // Client trusts the server CA but presents NO client cert + let client = build_client(&pki.ca_cert, None, None); + + let result = client + .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) + .send() + .await; + + // Server requires client cert -> TLS handshake should fail + assert!( + result.is_err(), + "request without client cert should fail when mTLS is the only strategy, but got: {:?}", + result.unwrap().status() + ); +} + +fn fixture_jwt_keypair() -> (jsonwebtoken::EncodingKey, jsonwebtoken::DecodingKey) { + let private_pem = std::fs::read(fixture_path("jwt-ed25519-private.pem")).unwrap(); + let public_pem = std::fs::read(fixture_path("jwt-ed25519-public.pem")).unwrap(); + + let encoding = + jsonwebtoken::EncodingKey::from_ed_pem(&private_pem).expect("invalid private key"); + let decoding = jsonwebtoken::DecodingKey::from_ed_pem(&public_pem).expect("invalid public key"); + (encoding, decoding) +} + +fn sign_jwt(key: &jsonwebtoken::EncodingKey, sub: &str) -> String { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let claims = serde_json::json!({ + "iss": "fabro-web", + "iat": now, + "exp": now + 60, + "sub": sub, + }); + let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::EdDSA); + jsonwebtoken::encode(&header, &claims, key).expect("failed to sign token") +} + +#[tokio::test] +async fn mtls_and_jwt_accepts_valid_jwt_without_client_cert() { + install_crypto_provider(); + let pki = fixture_pki(); + + let tls_settings = TlsSettings { + cert: pki.server_cert.clone(), + key: pki.server_key.clone(), + ca: pki.ca_cert.clone(), + }; + + let (encoding_key, decoding_key) = fixture_jwt_keypair(); + + // Both mTLS and JWT strategies; mTLS is optional since JWT is also present + let auth_mode = AuthMode::Strategies(vec![ + AuthStrategy::Mtls, + AuthStrategy::Jwt { + key: Arc::new(decoding_key), + validation: Arc::new(fabro_server::jwt_auth::jwt_validation()), + allowed_usernames: vec!["brynary".to_string()], + }, + ]); + let addr = start_tls_server(&tls_settings, ClientAuth::Optional, auth_mode).await; + + // Client trusts the server CA but presents NO client cert + let client = build_client(&pki.ca_cert, None, None); + + let token = sign_jwt(&encoding_key, "https://github.com/brynary"); + + let response = client + .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs"))) + .bearer_auth(&token) + .send() + .await + .expect("request with valid JWT and no client cert should succeed"); + + assert_eq!( + response.status(), + 200, + "valid JWT should be accepted when strategies = [mtls, jwt]" + ); +} diff --git a/lib/crates/fabro-server/tests/it/api/routing.rs b/lib/crates/fabro-server/tests/it/api/routing.rs new file mode 100644 index 000000000..f1631ac6e --- /dev/null +++ b/lib/crates/fabro-server/tests/it/api/routing.rs @@ -0,0 +1,75 @@ +use axum::body::Body; +use axum::http::{Method, Request, StatusCode}; +use fabro_server::server::{build_router, create_app_state}; +use tower::ServiceExt; + +use crate::helpers::body_json; + +#[tokio::test] +async fn old_unversioned_routes_return_404() { + let app = build_router( + create_app_state(), + fabro_server::jwt_auth::AuthMode::Disabled, + ); + + let cases = [(Method::POST, "/completions")]; + + for (method, path) in cases { + let req = Request::builder() + .method(method.clone()) + .uri(path) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND, "{method} {path}"); + } +} + +#[tokio::test] +async fn root_and_health_stay_at_root() { + let app = build_router( + create_app_state(), + fabro_server::jwt_auth::AuthMode::Disabled, + ); + + let root_req = Request::builder() + .method("GET") + .uri("/") + .body(Body::empty()) + .unwrap(); + let root_response = app.clone().oneshot(root_req).await.unwrap(); + assert_eq!(root_response.status(), StatusCode::OK); + let root_body = axum::body::to_bytes(root_response.into_body(), usize::MAX) + .await + .unwrap(); + let root_html = String::from_utf8(root_body.to_vec()).unwrap(); + assert!(root_html.contains("
")); + + let health_req = Request::builder() + .method("GET") + .uri("/health") + .body(Body::empty()) + .unwrap(); + let health_response = app.oneshot(health_req).await.unwrap(); + assert_eq!(health_response.status(), StatusCode::OK); + let health_body = body_json(health_response.into_body()).await; + assert_eq!(health_body["status"], "ok"); +} + +#[tokio::test] +async fn moved_routes_not_at_root_of_api_prefix() { + let app = build_router( + create_app_state(), + fabro_server::jwt_auth::AuthMode::Disabled, + ); + + for path in ["/api/v1/health", "/api/v1/"] { + let req = Request::builder() + .method("GET") + .uri(path) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND, "GET {path}"); + } +} diff --git a/lib/crates/fabro-server/tests/it/helpers.rs b/lib/crates/fabro-server/tests/it/helpers.rs index 03573071a..282c73e98 100644 --- a/lib/crates/fabro-server/tests/it/helpers.rs +++ b/lib/crates/fabro-server/tests/it/helpers.rs @@ -1,7 +1,120 @@ use std::sync::Arc; +use std::time::Duration; -use fabro_server::server::{AppState, create_app_state}; +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use fabro_server::jwt_auth::AuthMode; +use fabro_server::server::{ + AppState, build_router, create_app_state, create_app_state_with_options, spawn_scheduler, +}; +use fabro_types::Settings; +use tower::ServiceExt; + +pub(crate) const MINIMAL_DOT: &str = r#"digraph Test { + graph [goal="Test"] + start [shape=Mdiamond] + exit [shape=Msquare] + start -> exit +}"#; + +pub(crate) const POLL_INTERVAL: Duration = Duration::from_millis(10); +pub(crate) const POLL_ATTEMPTS: usize = 500; pub(crate) fn test_app_state() -> Arc { create_app_state() } + +pub(crate) fn dry_run_settings() -> Settings { + Settings { + dry_run: Some(true), + ..Default::default() + } +} + +pub(crate) fn dry_run_app() -> axum::Router { + let state = create_app_state_with_options(dry_run_settings(), 5); + spawn_scheduler(Arc::clone(&state)); + build_router(state, AuthMode::Disabled) +} + +pub(crate) fn test_app_with_scheduler(state: Arc) -> axum::Router { + spawn_scheduler(Arc::clone(&state)); + build_router(state, AuthMode::Disabled) +} + +pub(crate) fn api(path: &str) -> String { + format!("/api/v1{path}") +} + +pub(crate) async fn body_json(body: Body) -> serde_json::Value { + let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + serde_json::from_slice(&bytes).unwrap() +} + +/// Create a run via POST /runs, then start it via POST /runs/{id}/start. +/// Returns the run_id string. +pub(crate) async fn create_and_start_run(app: &axum::Router, dot_source: &str) -> String { + let req = Request::builder() + .method("POST") + .uri(api("/runs")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&serde_json::json!({"dot_source": dot_source})).unwrap(), + )) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + let body = body_json(response.into_body()).await; + let run_id = body["id"].as_str().unwrap().to_string(); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/start"))) + .body(Body::empty()) + .unwrap(); + app.clone().oneshot(req).await.unwrap(); + + run_id +} + +pub(crate) async fn run_json(app: &axum::Router, run_id: &str) -> serde_json::Value { + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + body_json(response.into_body()).await +} + +pub(crate) async fn wait_for_run_status( + app: &axum::Router, + run_id: &str, + expected: &[&str], +) -> String { + for _ in 0..POLL_ATTEMPTS { + let body = run_json(app, run_id).await; + let status = body["status"].as_str().unwrap().to_string(); + if expected.iter().any(|candidate| *candidate == status) { + return status; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + panic!("run {run_id} did not reach any of {expected:?}"); +} + +pub(crate) async fn wait_for_run_status_not_in( + app: &axum::Router, + run_id: &str, + unexpected: &[&str], +) -> String { + for _ in 0..POLL_ATTEMPTS { + let body = run_json(app, run_id).await; + let status = body["status"].as_str().unwrap().to_string(); + if unexpected.iter().all(|candidate| *candidate != status) { + return status; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + panic!("run {run_id} stayed in {unexpected:?}"); +} diff --git a/lib/crates/fabro-server/tests/it/main.rs b/lib/crates/fabro-server/tests/it/main.rs index d96c912e1..c5919aa46 100644 --- a/lib/crates/fabro-server/tests/it/main.rs +++ b/lib/crates/fabro-server/tests/it/main.rs @@ -2,3 +2,4 @@ mod api; mod helpers; mod openapi_conformance; mod pagination; +mod scenario; diff --git a/lib/crates/fabro-server/tests/it/scenario/dry_run.rs b/lib/crates/fabro-server/tests/it/scenario/dry_run.rs new file mode 100644 index 000000000..2807e4465 --- /dev/null +++ b/lib/crates/fabro-server/tests/it/scenario/dry_run.rs @@ -0,0 +1,69 @@ +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use tower::ServiceExt; + +use crate::helpers::{ + MINIMAL_DOT, api, body_json, create_and_start_run, dry_run_app, wait_for_run_status, +}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn dry_run_serve_starts_and_runs_workflow() { + let app = dry_run_app(); + + let run_id = create_and_start_run(&app, MINIMAL_DOT).await; + + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!(status, "succeeded"); +} + +#[tokio::test] +async fn test_model_known_via_full_router() { + let app = dry_run_app(); + + let req = Request::builder() + .method("POST") + .uri(api("/models/claude-opus-4-6/test")) + .header("content-type", "application/json") + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response.into_body()).await; + assert_eq!(body["model_id"], "claude-opus-4-6"); + // No API keys in test env, so status will be "error" + assert!(body["status"] == "ok" || body["status"] == "error"); +} + +#[tokio::test] +async fn test_model_unknown_via_full_router() { + let app = dry_run_app(); + + let req = Request::builder() + .method("POST") + .uri(api("/models/nonexistent-model-xyz/test")) + .header("content-type", "application/json") + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn dry_run_serve_rejects_invalid_dot() { + let app = dry_run_app(); + + let req = Request::builder() + .method("POST") + .uri(api("/runs")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&serde_json::json!({"dot_source": "not valid dot"})).unwrap(), + )) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); +} diff --git a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs new file mode 100644 index 000000000..b55bcdff7 --- /dev/null +++ b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs @@ -0,0 +1,184 @@ +use std::sync::Arc; + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use fabro_interview::Interviewer; +use fabro_server::server::{build_router, create_app_state_with_registry_factory}; +use fabro_workflow::handler::HandlerRegistry; +use fabro_workflow::handler::agent::AgentHandler; +use fabro_workflow::handler::exit::ExitHandler; +use fabro_workflow::handler::human::HumanHandler; +use fabro_workflow::handler::start::StartHandler; +use tower::ServiceExt; + +use crate::helpers::{ + POLL_ATTEMPTS, POLL_INTERVAL, api, body_json, run_json, wait_for_run_status, + wait_for_run_status_not_in, +}; + +fn gate_registry(interviewer: Arc) -> HandlerRegistry { + let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None))); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("agent", Box::new(AgentHandler::new(None))); + registry.register("human", Box::new(HumanHandler::new(interviewer))); + registry +} + +async fn wait_for_question_id(app: &axum::Router, run_id: &str) -> String { + for _ in 0..POLL_ATTEMPTS { + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/questions"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + let body = body_json(response.into_body()).await; + let arr = body["data"].as_array().unwrap(); + if let Some(question_id) = arr + .first() + .and_then(|item| item["id"].as_str()) + .map(ToOwned::to_owned) + { + return question_id; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + panic!("question should have appeared"); +} + +const GATE_DOT: &str = r#"digraph GateTest { + graph [goal="Test gate"] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, prompt="Do work"] + gate [shape=hexagon, type="human", label="Approve?"] + done [shape=box, prompt="Finish"] + revise [shape=box, prompt="Revise"] + + start -> work -> gate + gate -> done [label="[A] Approve"] + gate -> revise [label="[R] Revise"] + done -> exit + revise -> gate +}"#; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn full_http_lifecycle_approve_and_complete() { + let state = create_app_state_with_registry_factory(gate_registry); + fabro_server::server::spawn_scheduler(Arc::clone(&state)); + let app = build_router( + Arc::clone(&state), + fabro_server::jwt_auth::AuthMode::Disabled, + ); + + // 1. Create run + let req = Request::builder() + .method("POST") + .uri(api("/runs")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&serde_json::json!({"dot_source": GATE_DOT})).unwrap(), + )) + .unwrap(); + + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::CREATED); + let body = body_json(response.into_body()).await; + let run_id = body["id"].as_str().unwrap().to_string(); + + // 1b. Start the run + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/start"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + // 2. Poll for question to appear (run goes start -> work -> gate, then blocks) + let question_id = wait_for_question_id(&app, &run_id).await; + + // 3. Submit answer selecting first option (Approve) + let req = Request::builder() + .method("POST") + .uri(api(&format!( + "/runs/{run_id}/questions/{question_id}/answer" + ))) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&serde_json::json!({"value": "A"})).unwrap(), + )) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::NO_CONTENT); + + // 4. Poll until the run reaches a terminal success or failure state. + let final_status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!(final_status, "succeeded"); + + // 5. Verify no pending questions + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/questions"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + let body = body_json(response.into_body()).await; + assert!( + body["data"].as_array().unwrap().is_empty(), + "no pending questions after completion" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn full_http_lifecycle_cancel() { + let state = create_app_state_with_registry_factory(gate_registry); + fabro_server::server::spawn_scheduler(Arc::clone(&state)); + let app = build_router( + Arc::clone(&state), + fabro_server::jwt_auth::AuthMode::Disabled, + ); + + // Create and start a run that will block at the human gate + let req = Request::builder() + .method("POST") + .uri(api("/runs")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&serde_json::json!({"dot_source": GATE_DOT})).unwrap(), + )) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + let body = body_json(response.into_body()).await; + let run_id = body["id"].as_str().unwrap().to_string(); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/start"))) + .body(Body::empty()) + .unwrap(); + app.clone().oneshot(req).await.unwrap(); + + // Subscribe as soon as the scheduler has created the live event stream. + // Waiting past "starting" races with stage events because `/events` + // subscribes to future broadcast messages only; it does not replay. + wait_for_run_status_not_in(&app, &run_id, &["queued"]).await; + + // Cancel it + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/cancel"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = body_json(response.into_body()).await; + assert_eq!(body["status"], "cancelled"); + + // Verify the durable store view converges to cancelled failure. + let status = wait_for_run_status(&app, &run_id, &["failed"]).await; + assert_eq!(status, "failed"); + let body = run_json(&app, &run_id).await; + assert_eq!(body["status_reason"], "cancelled"); +} diff --git a/lib/crates/fabro-server/tests/it/scenario/mod.rs b/lib/crates/fabro-server/tests/it/scenario/mod.rs new file mode 100644 index 000000000..43889b335 --- /dev/null +++ b/lib/crates/fabro-server/tests/it/scenario/mod.rs @@ -0,0 +1,5 @@ +mod dry_run; +mod lifecycle; +mod run_completion; +mod sse; +mod usage; diff --git a/lib/crates/fabro-server/tests/it/scenario/run_completion.rs b/lib/crates/fabro-server/tests/it/scenario/run_completion.rs new file mode 100644 index 000000000..bc3dfc43e --- /dev/null +++ b/lib/crates/fabro-server/tests/it/scenario/run_completion.rs @@ -0,0 +1,57 @@ +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use fabro_server::server::create_app_state_with_options; +use tower::ServiceExt; + +use crate::helpers::{ + MINIMAL_DOT, api, create_and_start_run, dry_run_settings, test_app_with_scheduler, + wait_for_run_status, +}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn run_completes_and_status_is_completed() { + let state = create_app_state_with_options(dry_run_settings(), 5); + let app = test_app_with_scheduler(state); + + let run_id = create_and_start_run(&app, MINIMAL_DOT).await; + + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!(status, "succeeded"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn attach_run_events_returns_sse_stream() { + let state = create_app_state_with_options(dry_run_settings(), 5); + let app = test_app_with_scheduler(state); + + let run_id = create_and_start_run(&app, MINIMAL_DOT).await; + + // Wait for scheduler to promote run. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/attach"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + let status = response.status(); + assert!( + status == StatusCode::OK || status == StatusCode::GONE, + "unexpected status: {status}" + ); + + if status == StatusCode::OK { + let content_type = response + .headers() + .get("content-type") + .expect("content-type header should be present") + .to_str() + .unwrap(); + assert!( + content_type.contains("text/event-stream"), + "expected text/event-stream, got: {content_type}" + ); + } +} diff --git a/lib/crates/fabro-server/tests/it/scenario/sse.rs b/lib/crates/fabro-server/tests/it/scenario/sse.rs new file mode 100644 index 000000000..fcffc41bd --- /dev/null +++ b/lib/crates/fabro-server/tests/it/scenario/sse.rs @@ -0,0 +1,118 @@ +use std::time::Duration; + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use fabro_server::server::create_app_state_with_options; +use http_body_util::BodyExt; +use tower::ServiceExt; + +use crate::helpers::{ + POLL_ATTEMPTS, POLL_INTERVAL, api, body_json, create_and_start_run, dry_run_settings, + test_app_with_scheduler, wait_for_run_status_not_in, +}; + +const SIMPLE_DOT: &str = r#"digraph SSETest { + graph [goal="Test SSE"] + start [shape=Mdiamond] + work [shape=box, prompt="Do work"] + exit [shape=Msquare] + start -> work -> exit +}"#; + +async fn wait_for_checkpoint(app: &axum::Router, run_id: &str) -> serde_json::Value { + for _ in 0..POLL_ATTEMPTS { + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/checkpoint"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + if response.status() == StatusCode::OK { + return body_json(response.into_body()).await; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + panic!("checkpoint did not become available for {run_id}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn sse_stream_contains_expected_event_types() { + let state = create_app_state_with_options(dry_run_settings(), 5); + let app = test_app_with_scheduler(state); + + let run_id = create_and_start_run(&app, SIMPLE_DOT).await; + + wait_for_run_status_not_in(&app, &run_id, &["queued", "starting"]).await; + + // Get SSE stream + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/attach"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + // May be 200 (stream open) or 410 (run completed before connect) + let sse_status = response.status(); + assert!( + sse_status == StatusCode::OK || sse_status == StatusCode::GONE, + "expected 200 or 410, got: {sse_status}" + ); + if sse_status == StatusCode::GONE { + return; + } + + let content_type = response + .headers() + .get("content-type") + .unwrap() + .to_str() + .unwrap(); + assert!(content_type.contains("text/event-stream")); + + // Collect SSE frames with a timeout + let mut body = response.into_body(); + let mut sse_data = String::new(); + while let Ok(Some(Ok(frame))) = tokio::time::timeout(Duration::from_secs(2), body.frame()).await + { + if let Some(data) = frame.data_ref() { + sse_data.push_str(&String::from_utf8_lossy(data)); + } + } + + // Parse SSE data lines and extract event types + let mut event_types: Vec = Vec::new(); + for line in sse_data.lines() { + if let Some(json_str) = line.strip_prefix("data:") { + let json_str = json_str.trim(); + if let Ok(event) = serde_json::from_str::(json_str) { + if let Some(event_name) = event["payload"]["event"].as_str() { + event_types.push(event_name.to_string()); + } + } + } + } + + // Because we subscribe while the run is only guaranteed to be past + // "queued", a live stream should include at least one stage event. + // A 410 response above still covers the case where the run completed + // before we managed to attach. + if !event_types.is_empty() { + assert!( + event_types + .iter() + .any(|t| t == "stage.started" || t == "stage.completed"), + "should contain stage events, got: {event_types:?}" + ); + } + + // Pipeline is complete (SSE stream ended), verify checkpoint + let cp_body = wait_for_checkpoint(&app, &run_id).await; + // If run completed, checkpoint should have completed_nodes + if !cp_body.is_null() { + let completed = cp_body["completed_nodes"].as_array(); + if let Some(nodes) = completed { + let names: Vec<&str> = nodes.iter().filter_map(|v| v.as_str()).collect(); + assert!(names.contains(&"work"), "work should be in completed_nodes"); + } + } +} diff --git a/lib/crates/fabro-server/tests/it/scenario/usage.rs b/lib/crates/fabro-server/tests/it/scenario/usage.rs new file mode 100644 index 000000000..1d478cde7 --- /dev/null +++ b/lib/crates/fabro-server/tests/it/scenario/usage.rs @@ -0,0 +1,41 @@ +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use fabro_server::server::create_app_state_with_options; +use tower::ServiceExt; + +use crate::helpers::{ + MINIMAL_DOT, POLL_ATTEMPTS, POLL_INTERVAL, api, body_json, create_and_start_run, + dry_run_settings, test_app_with_scheduler, wait_for_run_status, +}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn aggregate_usage_increments_after_run_completes() { + let state = create_app_state_with_options(dry_run_settings(), 5); + let app = test_app_with_scheduler(state); + + let run_id = create_and_start_run(&app, MINIMAL_DOT).await; + + // Poll until run completes + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!(status, "succeeded"); + + let mut total_runs = 0; + for _ in 0..POLL_ATTEMPTS { + let req = Request::builder() + .method("GET") + .uri(api("/usage")) + .body(Body::empty()) + .unwrap(); + + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response.into_body()).await; + total_runs = body["totals"]["runs"].as_i64().unwrap(); + if total_runs == 1 { + break; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + assert_eq!(total_runs, 1); +}