scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(test): restructure fabro-server it/ into api/ and scenario/ subdirs

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-05 14:58:24 -04:00
parent 0bdcb4df77
commit 5f6a5ced6b
No known key found for this signature in database
13 changed files with 916 additions and 1026 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

116
lib/crates/fabro-server/src/server.rs
View file

@ -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::<RunId>().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::<RunId>().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::<RunId>().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();

909
lib/crates/fabro-server/tests/it/api.rs
View file

@ -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<dyn Interviewer>) -> 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<String> = 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::<serde_json::Value>(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("<div id=\"root\"></div>"));
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}");
}
}
}

3
lib/crates/fabro-server/tests/it/api/mod.rs Normal file
View file

@ -0,0 +1,3 @@
#[cfg(target_os = "linux")]
mod mtls;
mod routing;

249
lib/crates/fabro-server/tests/it/api/mtls.rs Normal file
View file

@ -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]"
);
}

75
lib/crates/fabro-server/tests/it/api/routing.rs Normal file
View file

@ -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("<div id=\"root\"></div>"));
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}");
}
}

115
lib/crates/fabro-server/tests/it/helpers.rs
View file

@ -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<AppState> {
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<AppState>) -> 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:?}");
}

1
lib/crates/fabro-server/tests/it/main.rs
View file

@ -2,3 +2,4 @@ mod api;
mod helpers;
mod openapi_conformance;
mod pagination;
mod scenario;

69
lib/crates/fabro-server/tests/it/scenario/dry_run.rs Normal file
View file

@ -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);
}

184
lib/crates/fabro-server/tests/it/scenario/lifecycle.rs Normal file
View file

@ -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<dyn Interviewer>) -> 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");
}

5
lib/crates/fabro-server/tests/it/scenario/mod.rs Normal file
View file

@ -0,0 +1,5 @@
mod dry_run;
mod lifecycle;
mod run_completion;
mod sse;
mod usage;

57
lib/crates/fabro-server/tests/it/scenario/run_completion.rs Normal file
View file

@ -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}"
);
}
}

118
lib/crates/fabro-server/tests/it/scenario/sse.rs Normal file
View file

@ -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<String> = 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::<serde_json::Value>(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");
}
}
}

41
lib/crates/fabro-server/tests/it/scenario/usage.rs Normal file
View file

@ -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);
}