fabro/lib/apps/fabro-server/tests/it/api/tcp.rs
Bryan Helmkamp a36bea15d2
Remove the engine flag: every run is a Petri run
Delete `Engine`, `RunEngine`, `[workflow] engine`, `[server.execution]
engine`, `FABRO_SERVER_ENGINE` and `fabro server start --engine`. The run
spec records what Petri admitted as `admission: PetriAdmission`; the
create handler always admits through `Runtime::check`; `execute_run`
always launches the Petri worker (or executes in process under the test
override); the CLI runner takes only the Petri worker path, and its
legacy control arm, artifact uploader, signal pause handlers and
credential helpers go with it. The CLI's `attach` and `events` read the
run stream only.

Two gaps this surfaced are closed here: the check adapter binds the
server's run variables as Petri compile variables (`{{ vars.* }}` in a
prompt no longer fails admission), and deleting a run removes its Petri
records, lease, platform records, projection and stream.

Tests: the config engine tests are replaced (an engine key is unknown),
the API round-trip test covers `PetriAdmission`, the server and CLI
Petri scenarios drop their engine settings, and the API tests that read
legacy event names now read the run stream or the session events. The
remaining red tests are fixtures and scenarios of the legacy executor
and the legacy event store (`fabro-store` `slate` and `run_state`,
`fabro-types` legacy `run.created` JSON, the server's handler-registry
scenarios, the CLI dry-run snapshots), which the next steps of the F4.3
series delete or port.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-18 09:58:21 -04:00

224 lines
6.7 KiB
Rust

#![expect(
clippy::disallowed_methods,
reason = "integration tests stage fixtures with sync std::fs; test infrastructure, not Tokio-hot path"
)]
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;
use axum::http::StatusCode;
use fabro_config::bind::Bind;
use fabro_config::{RuntimeDirectory, ServerSettingsBuilder};
use fabro_server::jwt_auth::{AuthMode, resolve_auth_mode_with_lookup};
use fabro_server::serve::{ServeArgs, serve_command};
use fabro_server::server::{RouterOptions, build_router_with_options};
use fabro_server::test_support::{TEST_DEV_TOKEN, TEST_SESSION_SECRET, test_app_state};
use fabro_util::terminal::Styles;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::task::JoinHandle;
use tokio::time::sleep;
use crate::helpers::{api, reqwest_status};
async fn start_tcp_server(auth_mode: AuthMode) -> SocketAddr {
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("test TCP listener should bind");
let addr = listener
.local_addr()
.expect("test TCP listener should have a local address");
let state = test_app_state();
let router = build_router_with_options(state, &auth_mode, RouterOptions::default());
tokio::spawn(async move {
let _ = axum::serve(listener, router).await;
});
addr
}
#[cfg(unix)]
fn build_unix_client(path: &Path) -> fabro_http::HttpClient {
fabro_http::HttpClientBuilder::new()
.unix_socket(path)
.no_proxy()
.build()
.expect("Unix test client should build")
}
fn write_test_config(tempdir: &TempDir, settings: &str) -> PathBuf {
let config_path = tempdir.path().join("settings.toml");
std::fs::write(&config_path, settings).expect("test settings should write");
std::fs::write(
RuntimeDirectory::new(tempdir.path()).env_path(),
format!("FABRO_DEV_TOKEN={TEST_DEV_TOKEN}\nSESSION_SECRET={TEST_SESSION_SECRET}\n"),
)
.expect("test env file should write");
config_path
}
async fn spawn_served_listener(
settings: impl AsRef<str>,
) -> (JoinHandle<anyhow::Result<()>>, Bind, TempDir) {
let tempdir = tempfile::tempdir().expect("temporary server directory should create");
let config_path = write_test_config(&tempdir, settings.as_ref());
let styles: &'static Styles = Box::leak(Box::new(Styles::new(false)));
let (tx, rx) = tokio::sync::oneshot::channel();
let mut tx = Some(tx);
let storage_dir = tempdir.path().to_path_buf();
let handle = tokio::spawn(async move {
Box::pin(serve_command(
ServeArgs {
bind: None,
web: false,
no_web: true,
model: None,
provider: None,
environment: None,
max_concurrent_runs: None,
config: Some(config_path),
#[cfg(debug_assertions)]
watch_web: false,
},
styles,
Some(storage_dir),
None,
move |bind| {
let sender = tx.take().expect("server should only report readiness once");
sender.send(bind.clone()).ok();
Ok(())
},
))
.await
});
let Ok(bind) = rx.await else {
let result = handle
.await
.expect("server task should not panic before reporting readiness");
panic!("server should report its bind address: {result:?}");
};
(handle, bind, tempdir)
}
async fn wait_for_health(client: &fabro_http::HttpClient, url: &str) {
for _ in 0..50 {
if let Ok(response) = client.get(url).send().await {
let status = response.status();
if status == 200 {
return;
}
}
sleep(Duration::from_millis(10)).await;
}
panic!("timed out waiting for health endpoint at {url}");
}
#[tokio::test]
async fn tcp_accepts_plain_http_requests() {
let (handle, bind, _tempdir) = spawn_served_listener(
r#"
_version = 1
[server.listen]
type = "tcp"
address = "127.0.0.1:0"
[server.auth]
methods = ["dev-token"]
"#,
)
.await;
let addr = match bind {
Bind::Tcp(addr) => addr,
Bind::Unix(path) => panic!("expected TCP bind, got unix socket at {}", path.display()),
};
let client = fabro_http::test_http_client().unwrap();
wait_for_health(&client, &format!("http://127.0.0.1:{}/health", addr.port())).await;
let response = client
.get(format!("http://127.0.0.1:{}{}", addr.port(), api("/runs")))
.bearer_auth(TEST_DEV_TOKEN)
.send()
.await
.expect("plain HTTP request should succeed");
reqwest_status(response, StatusCode::OK, "GET /api/v1/runs").await;
handle.abort();
}
#[tokio::test]
async fn tcp_dev_token_auth_uses_bearer_auth() {
let resolved = ServerSettingsBuilder::from_toml(
r#"
_version = 1
[server.auth]
methods = ["dev-token"]
"#,
)
.expect("test settings should resolve")
.server;
let auth_mode = resolve_auth_mode_with_lookup(&resolved, |name| match name {
"SESSION_SECRET" => Some(TEST_SESSION_SECRET.to_string()),
"FABRO_DEV_TOKEN" => Some(TEST_DEV_TOKEN.to_string()),
_ => None,
})
.expect("auth mode should resolve");
let addr = start_tcp_server(auth_mode).await;
let client = fabro_http::test_http_client().unwrap();
let url = format!("http://127.0.0.1:{}{}", addr.port(), api("/runs"));
let unauthorized = client.get(&url).send().await.unwrap();
reqwest_status(unauthorized, StatusCode::UNAUTHORIZED, "GET /api/v1/runs").await;
let authorized = client
.get(url)
.bearer_auth(TEST_DEV_TOKEN)
.send()
.await
.unwrap();
reqwest_status(authorized, StatusCode::OK, "GET /api/v1/runs").await;
}
#[cfg(unix)]
#[tokio::test]
async fn unix_socket_accepts_plain_http_requests() {
let socket_dir = tempfile::tempdir().unwrap();
let socket_path = socket_dir.path().join("fabro.sock");
let (handle, bind, _tempdir) = spawn_served_listener(format!(
r#"
_version = 1
[server.listen]
type = "unix"
path = "{}"
[server.auth]
methods = ["dev-token"]
"#,
socket_path.display()
))
.await;
let path = match bind {
Bind::Unix(path) => path,
Bind::Tcp(addr) => panic!("expected Unix bind, got TCP address {addr}"),
};
let client = build_unix_client(&path);
wait_for_health(&client, "http://fabro/health").await;
let response = client
.get(format!("http://fabro{}", api("/runs")))
.bearer_auth(TEST_DEV_TOKEN)
.send()
.await
.expect("Unix-socket HTTP request should succeed");
reqwest_status(response, StatusCode::OK, "GET /api/v1/runs").await;
handle.abort();
}