use std::sync::Arc;
use std::time::Duration;
use ironflow_api::routes::{RouterConfig, create_router};
use ironflow_api::state::AppState;
use ironflow_auth::jwt::{AccessToken, JwtConfig};
use ironflow_auth::password;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::engine::Engine;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
use ironflow_engine::notify::Event;
use ironflow_sdk::IronflowClient;
use ironflow_sdk::client::{ClientConfig, ListStepsParams};
use ironflow_sdk::types::PlanWorkflowRequest;
use ironflow_store::entities::{
NewStep, NewUser, RunStatus, StepKind, StepStatus, StepUpdate, TriggerKind, step_trace_id,
};
use ironflow_store::memory::InMemoryStore;
use ironflow_store::store::Store;
use serde_json::json;
use tokio::net::TcpListener;
use tokio::sync::broadcast;
use uuid::Uuid;
struct DeployWorkflow;
impl WorkflowHandler for DeployWorkflow {
fn name(&self) -> &str {
"deploy"
}
fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
Box::pin(async move { Ok(()) })
}
}
struct BuildWorkflow;
impl WorkflowHandler for BuildWorkflow {
fn name(&self) -> &str {
"build"
}
fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
Box::pin(async move { Ok(()) })
}
}
struct PlannedWorkflow;
impl WorkflowHandler for PlannedWorkflow {
fn name(&self) -> &str {
"planned"
}
fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
Box::pin(async move {
ctx.shell("compile", ShellConfig::new("echo compile"))
.await?;
ctx.shell("package", ShellConfig::new("echo package"))
.await?;
Ok(())
})
}
}
fn jwt_config() -> Arc<JwtConfig> {
Arc::new(JwtConfig {
secret: "integration-test-secret".to_string(),
access_token_ttl_secs: 900,
refresh_token_ttl_secs: 604800,
cookie_domain: None,
cookie_secure: false,
})
}
async fn spawn_server() -> (String, String) {
spawn_server_from(Arc::new(InMemoryStore::new())).await
}
async fn spawn_server_from(store: Arc<dyn Store>) -> (String, String) {
let provider = Arc::new(ClaudeCodeProvider::new());
let mut engine = Engine::new(store.clone(), provider);
engine.register(DeployWorkflow).unwrap();
engine.register(BuildWorkflow).unwrap();
engine.register(PlannedWorkflow).unwrap();
let jwt_cfg = jwt_config();
let (event_sender, _) = broadcast::channel::<Event>(16);
let hash = password::hash("test-password").unwrap();
let user = store
.create_user(NewUser {
email: "integration@test.local".to_string(),
username: "integration-test".to_string(),
password_hash: hash,
is_admin: Some(true),
})
.await
.unwrap();
let state = AppState::new(
store,
Arc::new(engine),
jwt_cfg.clone(),
"test-worker-token".to_string(),
event_sender,
);
let config = RouterConfig {
rate_limit_auth: None,
rate_limit_general: None,
..RouterConfig::default()
};
let router = create_router(state, config);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, router).await.unwrap();
});
let base_url = format!("http://{addr}");
let token = AccessToken::for_user(user.id, "integration-test", true, &jwt_cfg).unwrap();
(base_url, token.0)
}
fn make_client(base_url: &str, token: &str) -> IronflowClient {
let config = ClientConfig {
base_url: base_url.to_string(),
api_key: token.to_string(),
timeout: Duration::from_secs(10),
};
IronflowClient::from_config(config)
}
#[tokio::test]
async fn health_check() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
client.health_check().await.unwrap();
}
#[tokio::test]
async fn list_runs_returns_empty() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let response = client.list_runs().await.unwrap();
assert!(response.data.is_empty());
}
#[tokio::test]
async fn list_runs_with_filter() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let filter = ironflow_sdk::client::ListRunsFilter {
workflow: Some("deploy"),
page: Some(1),
per_page: Some(10),
..Default::default()
};
let response = client.list_runs_filtered(&filter).await.unwrap();
assert!(response.data.is_empty());
}
#[tokio::test]
async fn list_workflows_returns_registered() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let response = client.list_workflows().await.unwrap();
let names: Vec<&str> = response.data.iter().map(|w| w.name.as_str()).collect();
assert!(names.contains(&"deploy"));
assert!(names.contains(&"build"));
}
#[tokio::test]
async fn get_workflow_found() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let response = client.get_workflow("deploy").await.unwrap();
assert_eq!(response.data.name, "deploy");
}
#[tokio::test]
async fn plan_workflow_returns_steps() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request: PlanWorkflowRequest = PlanWorkflowRequest::builder()
.try_into()
.expect("plan request builds");
let response = client.plan_workflow("planned", &request).await.unwrap();
assert_eq!(response.data.workflow, "planned");
let names: Vec<&str> = response
.data
.steps
.iter()
.map(|s| s.name.as_str())
.collect();
assert_eq!(names, vec!["compile", "package"]);
assert!(!response.data.truncated);
}
#[tokio::test]
async fn get_workflow_not_found() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let err = client.get_workflow("nonexistent").await.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(404));
}
#[tokio::test]
async fn get_stats() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let response = client.get_stats().await.unwrap();
assert_eq!(response.data.total_runs, 0);
}
#[tokio::test]
async fn get_me() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let response = client.me().await.unwrap();
assert_eq!(response.data.username, "integration-test");
assert!(response.data.is_admin);
}
#[tokio::test]
async fn get_run_not_found() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let fake_id = Uuid::now_v7();
let err = client.get_run(fake_id).await.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(404));
}
#[tokio::test]
async fn create_and_get_run() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let mut payload = serde_json::Map::new();
payload.insert("env".to_string(), serde_json::json!("staging"));
let request = ironflow_sdk::types::CreateRunRequest {
workflow: "deploy".to_string(),
payload: Some(payload),
labels: None,
scheduled_at: None,
max_retries: Some(2),
max_cost_usd: None,
concurrency_key: None,
priority: None,
concurrency_limits: Vec::new(),
worker_tags: Vec::new(),
};
let created = client.create_run(&request).await.unwrap();
assert_eq!(created.data.workflow_name, "deploy");
assert_eq!(created.data.max_retries, 2);
let fetched = client.get_run(created.data.id).await.unwrap();
assert_eq!(fetched.data.run.id, created.data.id);
}
#[tokio::test]
async fn create_run_with_max_cost_usd() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
workflow: "deploy".to_string(),
payload: None,
labels: None,
scheduled_at: None,
max_retries: None,
max_cost_usd: Some(2.5),
concurrency_key: None,
priority: None,
concurrency_limits: Vec::new(),
worker_tags: Vec::new(),
};
let created = client.create_run(&request).await.unwrap();
assert_eq!(created.data.max_cost_usd, Some(2.5));
}
#[tokio::test]
async fn create_run_rejects_negative_max_cost_usd() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
workflow: "deploy".to_string(),
payload: None,
labels: None,
scheduled_at: None,
max_retries: None,
max_cost_usd: Some(-1.0),
concurrency_key: None,
priority: None,
concurrency_limits: Vec::new(),
worker_tags: Vec::new(),
};
let err = client.create_run(&request).await.unwrap_err();
assert_eq!(err.status(), Some(400));
}
#[tokio::test]
async fn create_run_unknown_workflow() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
workflow: "does-not-exist".to_string(),
payload: None,
labels: None,
scheduled_at: None,
max_retries: None,
max_cost_usd: None,
concurrency_key: None,
priority: None,
concurrency_limits: Vec::new(),
worker_tags: Vec::new(),
};
let err = client.create_run(&request).await.unwrap_err();
assert!(err.is_api_error());
}
#[tokio::test]
async fn create_run_with_concurrency_limits_and_filter_by_group() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
concurrency_limits: vec![ironflow_sdk::types::ConcurrencyLimit {
group: "repo:acme".to_string(),
limit: 2,
}],
..deploy_request()
};
let created = client.create_run(&request).await.unwrap();
let limits = &created.data.concurrency_limits;
assert_eq!(limits.len(), 1);
assert_eq!(limits[0].group, "repo:acme");
assert_eq!(limits[0].limit, 2);
client.create_run(&deploy_request()).await.unwrap();
let filter = ironflow_sdk::client::ListRunsFilter {
concurrency_group: Some("repo:acme"),
..Default::default()
};
let response = client.list_runs_filtered(&filter).await.unwrap();
assert_eq!(response.data.len(), 1);
assert_eq!(response.data[0].id, created.data.id);
}
#[tokio::test]
async fn create_run_with_priority_and_filter_by_priority() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
priority: Some(80),
..deploy_request()
};
let created = client.create_run(&request).await.unwrap();
assert_eq!(created.data.priority, Some(80));
let default = client.create_run(&deploy_request()).await.unwrap();
assert_eq!(default.data.priority, Some(0));
let filter = ironflow_sdk::client::ListRunsFilter {
priority: Some(80),
..Default::default()
};
let response = client.list_runs_filtered(&filter).await.unwrap();
assert_eq!(response.data.len(), 1);
assert_eq!(response.data[0].id, created.data.id);
}
#[tokio::test]
async fn create_run_rejects_an_out_of_range_priority() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
priority: Some(101),
..deploy_request()
};
let err = client.create_run(&request).await.unwrap_err();
assert!(err.is_api_error());
}
#[tokio::test]
async fn create_run_rejects_a_zero_concurrency_limit() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
concurrency_limits: vec![ironflow_sdk::types::ConcurrencyLimit {
group: "repo:acme".to_string(),
limit: 0,
}],
..deploy_request()
};
let err = client.create_run(&request).await.unwrap_err();
assert_eq!(err.status(), Some(400));
}
#[tokio::test]
async fn create_run_with_worker_tags_returns_them_sorted() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
worker_tags: vec!["region:eu".to_string(), "gpu".to_string()],
..deploy_request()
};
let created = client.create_run(&request).await.unwrap();
let expected = vec!["gpu".to_string(), "region:eu".to_string()];
assert_eq!(created.data.worker_tags, expected);
let fetched = client.get_run(created.data.id).await.unwrap();
assert_eq!(fetched.data.run.worker_tags, expected);
}
#[tokio::test]
async fn create_run_rejects_an_invalid_worker_tag() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let request = ironflow_sdk::types::CreateRunRequest {
worker_tags: vec!["not a tag".to_string()],
..deploy_request()
};
let err = client.create_run(&request).await.unwrap_err();
assert_eq!(err.status(), Some(400));
}
#[tokio::test]
async fn unauthorized_without_token() {
let (base_url, _) = spawn_server().await;
let client = make_client(&base_url, "invalid-token");
let err = client.list_runs().await.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(401));
}
fn deploy_request() -> ironflow_sdk::types::CreateRunRequest {
let mut payload = serde_json::Map::new();
payload.insert("env".to_string(), serde_json::json!("prod"));
ironflow_sdk::types::CreateRunRequest {
workflow: "deploy".to_string(),
payload: Some(payload),
labels: None,
scheduled_at: None,
max_retries: None,
max_cost_usd: None,
concurrency_key: None,
priority: None,
concurrency_limits: Vec::new(),
worker_tags: Vec::new(),
}
}
#[tokio::test]
async fn create_run_idempotent_binds_the_key() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let created = client
.create_run_idempotent(&deploy_request(), "github:abc-123")
.await
.unwrap();
assert_eq!(
created.data.idempotency_key.as_deref(),
Some("github:abc-123")
);
}
#[tokio::test]
async fn create_run_idempotent_replays_the_same_run() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let first = client
.create_run_idempotent(&deploy_request(), "github:abc-123")
.await
.unwrap();
let second = client
.create_run_idempotent(&deploy_request(), "github:abc-123")
.await
.unwrap();
assert_eq!(first.data.id, second.data.id);
}
#[tokio::test]
async fn create_run_idempotent_conflicts_on_a_different_payload() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
client
.create_run_idempotent(&deploy_request(), "github:abc-123")
.await
.unwrap();
let mut other = deploy_request();
let mut payload = serde_json::Map::new();
payload.insert("env".to_string(), serde_json::json!("staging"));
other.payload = Some(payload);
let err = client
.create_run_idempotent(&other, "github:abc-123")
.await
.unwrap_err();
assert_eq!(err.status(), Some(409));
}
#[tokio::test]
async fn create_run_idempotent_rejects_a_malformed_key() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let err = client
.create_run_idempotent(&deploy_request(), "")
.await
.unwrap_err();
assert_eq!(err.status(), Some(400));
}
#[tokio::test]
async fn create_run_without_a_key_still_duplicates() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let first = client.create_run(&deploy_request()).await.unwrap();
let second = client.create_run(&deploy_request()).await.unwrap();
assert_ne!(first.data.id, second.data.id);
assert!(first.data.idempotency_key.is_none());
}
#[tokio::test]
async fn replay_completed_run_creates_new_run() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
let original = client.create_run(&deploy_request()).await.unwrap();
let original_id = original.data.id;
store
.update_run_status(original_id, RunStatus::Running)
.await
.unwrap();
store
.update_run_status(original_id, RunStatus::Completed)
.await
.unwrap();
let replayed = client.replay_run(original_id).await.unwrap();
let new_id = replayed.data.id;
assert_ne!(new_id, original_id);
assert_eq!(replayed.data.workflow_name, "deploy");
let stored = store.get_run(new_id).await.unwrap().unwrap();
assert_eq!(
stored.trigger,
TriggerKind::Replay {
original_run_id: original_id,
}
);
assert_eq!(stored.status.state, RunStatus::Pending);
let fetched = client.get_run(new_id).await.unwrap();
assert_eq!(fetched.data.run.id, new_id);
let original_after = store.get_run(original_id).await.unwrap().unwrap();
assert_eq!(original_after.status.state, RunStatus::Completed);
}
#[tokio::test]
async fn replay_pending_run_returns_400() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
let original = client.create_run(&deploy_request()).await.unwrap();
let err = client.replay_run(original.data.id).await.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(400));
}
#[tokio::test]
async fn replay_run_not_found() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let err = client.replay_run(Uuid::now_v7()).await.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(404));
}
async fn run_awaiting_input(store: &Arc<dyn Store>, client: &IronflowClient) -> (Uuid, Uuid) {
let run = client.create_run(&deploy_request()).await.unwrap();
let run_id = run.data.id;
store
.update_run_status(run_id, RunStatus::Running)
.await
.unwrap();
store
.update_run_status(run_id, RunStatus::AwaitingApproval)
.await
.unwrap();
let step = store
.create_step(NewStep {
run_id,
trace_id: step_trace_id(run_id, "clarify", 0),
name: "clarify".to_string(),
kind: StepKind::HumanInput,
position: 0,
input: Some(json!({
"message": "Answer the questions",
"schema": {
"type": "object",
"required": ["answers"],
"properties": {"answers": {"type": "array", "items": {"type": "string"}}},
},
})),
is_error_handler: false,
})
.await
.unwrap();
for status in [StepStatus::Running, StepStatus::AwaitingApproval] {
store
.update_step(
step.id,
StepUpdate {
status: Some(status),
..StepUpdate::default()
},
)
.await
.unwrap();
}
(run_id, step.id)
}
#[tokio::test]
async fn submit_input_posts_the_answer_to_the_step() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
let (run_id, step_id) = run_awaiting_input(&store, &client).await;
let answer = json!({"answers": ["staging"]});
let resp = client.submit_input(run_id, step_id, &answer).await.unwrap();
assert_eq!(resp.data.id, run_id);
let step = store.get_step(step_id).await.unwrap().unwrap();
assert_eq!(step.status.state, StepStatus::Completed);
assert_eq!(step.output, Some(answer.clone()));
let err = client
.submit_input(run_id, step_id, &answer)
.await
.unwrap_err();
assert_eq!(err.status(), Some(409));
}
#[tokio::test]
async fn submit_input_with_an_invalid_answer_returns_422() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
let (run_id, step_id) = run_awaiting_input(&store, &client).await;
let err = client
.submit_input(run_id, step_id, &json!({"answers": 3}))
.await
.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(422));
assert_eq!(err.code(), Some("INVALID_INPUT"));
}
#[tokio::test]
async fn reject_input_records_the_reason() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
let (run_id, step_id) = run_awaiting_input(&store, &client).await;
client
.reject_input(run_id, step_id, Some("out of scope"))
.await
.unwrap();
let step = store.get_step(step_id).await.unwrap().unwrap();
assert_eq!(step.status.state, StepStatus::Rejected);
assert_eq!(step.error.as_deref(), Some("out of scope"));
}
async fn create_role_step(store: &Arc<dyn Store>, client: &IronflowClient, role: &str) -> Uuid {
let run = client.create_run(&deploy_request()).await.unwrap();
let run_id = run.data.id;
store
.create_step(NewStep {
run_id,
trace_id: step_trace_id(run_id, "review", 0),
name: "review".to_string(),
kind: StepKind::Agent,
position: 0,
input: Some(json!({"type": "agent", "prompt": "p", "role": role})),
is_error_handler: false,
})
.await
.unwrap()
.id
}
#[tokio::test]
async fn list_steps_returns_the_steps_of_a_role() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
let step_id = create_role_step(&store, &client, "Reviewer").await;
create_role_step(&store, &client, "Writer").await;
let response = client
.list_steps(&ListStepsParams {
agent_role: "Reviewer",
page: None,
per_page: None,
})
.await
.unwrap();
assert_eq!(response.data.len(), 1);
assert_eq!(response.data[0].id, step_id);
assert_eq!(response.data[0].agent_role.as_deref(), Some("Reviewer"));
}
#[tokio::test]
async fn list_steps_rejects_a_blank_role() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let err = client
.list_steps(&ListStepsParams {
agent_role: " ",
page: None,
per_page: None,
})
.await
.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(400));
}
#[tokio::test]
async fn agent_role_stats_returns_one_entry_per_role() {
let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
let (base_url, token) = spawn_server_from(store.clone()).await;
let client = make_client(&base_url, &token);
create_role_step(&store, &client, "Reviewer").await;
create_role_step(&store, &client, "Reviewer").await;
let response = client.agent_role_stats(Some("24h")).await.unwrap();
assert_eq!(response.data.roles.len(), 1);
assert_eq!(response.data.roles[0].agent_role, "Reviewer");
assert_eq!(response.data.roles[0].executions, 2);
assert_eq!(response.data.roles[0].failed, 0);
}
#[tokio::test]
async fn agent_role_stats_rejects_an_unknown_period() {
let (base_url, token) = spawn_server().await;
let client = make_client(&base_url, &token);
let err = client.agent_role_stats(Some("1y")).await.unwrap_err();
assert!(err.is_api_error());
assert_eq!(err.status(), Some(400));
}