#![cfg(feature = "server")]
use axon::axon_server::{build_router, ServerConfig};
use axum::body::Body;
use axum::http::{Request, StatusCode};
use http_body_util::BodyExt;
use tower::ServiceExt;
fn server_cfg() -> ServerConfig {
ServerConfig {
host: "127.0.0.1".into(),
port: 0,
channel: "memory".into(),
auth_token: String::new(),
log_level: "INFO".into(),
log_format: "json".into(),
log_file: None,
database_url: None,
config_path: None,
strict_type_driven_transport: false,
default_backend: None,
schemas_dir: None,
}
}
async fn deploy(app: axum::Router, src: &str) {
let body = serde_json::json!({
"source": src,
"source_file": "test.axon",
"backend": "stub",
});
let req = Request::builder()
.method("POST")
.uri("/v1/deploy")
.header("content-type", "application/json")
.body(Body::from(body.to_string()))
.unwrap();
let resp = app.oneshot(req).await.unwrap();
let status = resp.status();
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap_or_default();
assert_eq!(status, StatusCode::OK, "deploy failed: {json}");
assert_eq!(
json.get("success").and_then(|v| v.as_bool()),
Some(true),
"deploy success=false: {json}"
);
}
const POST_FLOW: &str =
"flow Charge() -> String { let result = \"ok\" return result }\n\
axonendpoint ChargeEndpoint { public: true method: POST path: \"/charge\" execute: Charge }";
const GET_FLOW: &str =
"flow Ping() -> String { let result = \"pong\" return result }\n\
axonendpoint PingEndpoint { public: true method: GET path: \"/ping\" execute: Ping }";
const SECOND_PATH_FLOW: &str =
"flow Charge() -> String { let result = \"ok\" return result }\n\
flow Refund() -> String { let result = \"refunded\" return result }\n\
axonendpoint ChargeEndpoint { public: true method: POST path: \"/charge\" execute: Charge }\n\
axonendpoint RefundEndpoint { public: true method: POST path: \"/refund\" execute: Refund }";
fn post_with(
path: &str,
body: &serde_json::Value,
idempotency_key: Option<&str>,
auth: Option<&str>,
) -> Request<Body> {
let mut builder = Request::builder()
.method("POST")
.uri(path)
.header("content-type", "application/json");
if let Some(k) = idempotency_key {
builder = builder.header("idempotency-key", k);
}
if let Some(a) = auth {
builder = builder.header("authorization", a);
}
builder
.body(Body::from(serde_json::to_vec(body).unwrap()))
.unwrap()
}
#[tokio::test]
async fn first_request_caches_and_replay_returns_cached_body() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let body = serde_json::json!({"amount": 100});
let req1 = post_with("/charge", &body, Some("idem-001"), Some("Bearer tA"));
let resp1 = app.clone().oneshot(req1).await.unwrap();
let status1 = resp1.status();
let replayed_hdr_1 = resp1.headers().get("idempotency-status").cloned();
let bytes1 = resp1.into_body().collect().await.unwrap().to_bytes();
assert_eq!(status1, StatusCode::OK);
assert!(
replayed_hdr_1.is_none(),
"first response must NOT carry Idempotency-Status header"
);
let req2 = post_with("/charge", &body, Some("idem-001"), Some("Bearer tA"));
let resp2 = app.oneshot(req2).await.unwrap();
let status2 = resp2.status();
let replayed_hdr_2 = resp2
.headers()
.get("idempotency-status")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let bytes2 = resp2.into_body().collect().await.unwrap().to_bytes();
assert_eq!(status2, StatusCode::OK);
assert_eq!(
replayed_hdr_2.as_deref(),
Some("replayed"),
"replay response must carry Idempotency-Status: replayed"
);
assert_eq!(
bytes1, bytes2,
"D7 invariant: same key + same body ⟹ byte-identical response"
);
}
#[tokio::test]
async fn same_key_different_body_returns_422_conflict() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let req1 = post_with(
"/charge",
&serde_json::json!({"amount": 100}),
Some("idem-conflict"),
Some("Bearer tA"),
);
let resp1 = app.clone().oneshot(req1).await.unwrap();
assert_eq!(resp1.status(), StatusCode::OK);
let req2 = post_with(
"/charge",
&serde_json::json!({"amount": 999}),
Some("idem-conflict"),
Some("Bearer tA"),
);
let resp2 = app.oneshot(req2).await.unwrap();
assert_eq!(resp2.status(), StatusCode::UNPROCESSABLE_ENTITY);
let bytes = resp2.into_body().collect().await.unwrap().to_bytes();
let payload: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(
payload["error"],
"idempotency_key_reused_with_different_request"
);
assert_eq!(payload["d_letter"], "D7");
assert_eq!(payload["idempotency_key"], "idem-conflict");
let prefix = payload["cached_body_hash_prefix"].as_str().unwrap();
assert_eq!(prefix.len(), 16);
}
#[tokio::test]
async fn no_idempotency_key_skips_cache_and_replay_header() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let body = serde_json::json!({"amount": 50});
let req1 = post_with("/charge", &body, None, Some("Bearer tA"));
let resp1 = app.clone().oneshot(req1).await.unwrap();
assert_eq!(resp1.status(), StatusCode::OK);
assert!(resp1.headers().get("idempotency-status").is_none());
let req2 = post_with("/charge", &body, None, Some("Bearer tA"));
let resp2 = app.oneshot(req2).await.unwrap();
assert!(resp2.headers().get("idempotency-status").is_none());
}
#[tokio::test]
async fn cross_tenant_same_key_is_not_a_collision() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let req_a = post_with(
"/charge",
&serde_json::json!({"amount": 1}),
Some("shared"),
Some("Bearer tA"),
);
let resp_a = app.clone().oneshot(req_a).await.unwrap();
assert_eq!(resp_a.status(), StatusCode::OK);
let req_b = post_with(
"/charge",
&serde_json::json!({"amount": 1}),
Some("shared"),
Some("Bearer tB"),
);
let resp_b = app.oneshot(req_b).await.unwrap();
let replayed = resp_b.headers().get("idempotency-status");
assert!(
replayed.is_none(),
"Tenant B's first request must NOT see Tenant A's cached response"
);
}
#[tokio::test]
async fn cross_path_same_key_is_not_a_collision() {
let app = build_router(server_cfg());
deploy(app.clone(), SECOND_PATH_FLOW).await;
let req_charge = post_with(
"/charge",
&serde_json::json!({"a": 1}),
Some("path-key"),
Some("Bearer tA"),
);
let resp_charge = app.clone().oneshot(req_charge).await.unwrap();
assert_eq!(resp_charge.status(), StatusCode::OK);
let req_refund = post_with(
"/refund",
&serde_json::json!({"a": 1}),
Some("path-key"),
Some("Bearer tA"),
);
let resp_refund = app.oneshot(req_refund).await.unwrap();
let replayed = resp_refund.headers().get("idempotency-status");
assert!(
replayed.is_none(),
"/refund must NOT replay /charge's cached response under the same key"
);
}
#[tokio::test]
async fn idempotency_key_on_get_is_ignored() {
let app = build_router(server_cfg());
deploy(app.clone(), GET_FLOW).await;
let req = Request::builder()
.method("GET")
.uri("/ping")
.header("idempotency-key", "key-on-get")
.header("authorization", "Bearer tA")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert!(resp.headers().get("idempotency-status").is_none());
}
#[tokio::test]
async fn v1_execute_legacy_unaffected_by_idempotency_gate() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let body = serde_json::json!({"flow": "Charge", "backend": "stub"});
let req = Request::builder()
.method("POST")
.uri("/v1/execute")
.header("content-type", "application/json")
.header("idempotency-key", "legacy-key")
.body(Body::from(serde_json::to_vec(&body).unwrap()))
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
assert!(resp.headers().get("idempotency-status").is_none());
}
#[tokio::test]
async fn multiple_replays_all_return_same_body() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let body = serde_json::json!({"amount": 250});
let mut bodies = Vec::new();
for _ in 0..4 {
let req = post_with("/charge", &body, Some("retry-key"), Some("Bearer tA"));
let resp = app.clone().oneshot(req).await.unwrap();
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
bodies.push(bytes);
}
for i in 1..bodies.len() {
assert_eq!(
bodies[0], bodies[i],
"replay #{i} must be byte-identical to the original"
);
}
}
#[tokio::test]
async fn put_method_also_caches() {
let src = "flow Upsert() -> String { let result = \"upserted\" return result }\n\
axonendpoint UpsertEndpoint { public: true method: PUT path: \"/upsert\" execute: Upsert }";
let app = build_router(server_cfg());
deploy(app.clone(), src).await;
let make_req = |key: &str| -> Request<Body> {
Request::builder()
.method("PUT")
.uri("/upsert")
.header("content-type", "application/json")
.header("idempotency-key", key)
.header("authorization", "Bearer tA")
.body(Body::from(serde_json::json!({"v": 1}).to_string()))
.unwrap()
};
let resp1 = app.clone().oneshot(make_req("put-key")).await.unwrap();
let bytes1 = resp1.into_body().collect().await.unwrap().to_bytes();
let resp2 = app.oneshot(make_req("put-key")).await.unwrap();
let replayed = resp2
.headers()
.get("idempotency-status")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let bytes2 = resp2.into_body().collect().await.unwrap().to_bytes();
assert_eq!(replayed.as_deref(), Some("replayed"));
assert_eq!(bytes1, bytes2);
}
#[tokio::test]
async fn conflict_payload_shape_is_locked() {
let app = build_router(server_cfg());
deploy(app.clone(), POST_FLOW).await;
let _ = app
.clone()
.oneshot(post_with(
"/charge",
&serde_json::json!({"amount": 1}),
Some("conflict-shape"),
Some("Bearer tA"),
))
.await
.unwrap();
let resp = app
.oneshot(post_with(
"/charge",
&serde_json::json!({"amount": 2}),
Some("conflict-shape"),
Some("Bearer tA"),
))
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
let p: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert!(p["error"].is_string());
assert!(p["idempotency_key"].is_string());
assert!(p["endpoint"].is_string());
assert!(p["method"].is_string());
assert!(p["path"].is_string());
assert!(p["cached_body_hash_prefix"].is_string());
assert!(p["hint"].is_string());
assert_eq!(p["d_letter"], "D7");
}
#[tokio::test]
async fn body_schema_violation_fires_before_idempotency_caching() {
let src = "type Req { amount: Integer }\n\
flow Charge() -> String { let result = \"ok\" return result }\n\
axonendpoint ChargeEndpoint { public: true method: POST path: \"/charge2\" body: Req execute: Charge }";
let app = build_router(server_cfg());
deploy(app.clone(), src).await;
let req1 = post_with(
"/charge2",
&serde_json::json!({"amount": "not-int"}),
Some("compose-key"),
Some("Bearer tA"),
);
let resp1 = app.clone().oneshot(req1).await.unwrap();
assert_eq!(resp1.status(), StatusCode::BAD_REQUEST);
let bytes1 = resp1.into_body().collect().await.unwrap().to_bytes();
let p1: serde_json::Value = serde_json::from_slice(&bytes1).unwrap();
assert_eq!(p1["error"], "body_schema_violation");
let req2 = post_with(
"/charge2",
&serde_json::json!({"amount": 42}),
Some("compose-key"),
Some("Bearer tA"),
);
let resp2 = app.oneshot(req2).await.unwrap();
let status2 = resp2.status();
let bytes2 = resp2.into_body().collect().await.unwrap().to_bytes();
let p2: serde_json::Value = serde_json::from_slice(&bytes2).unwrap_or_default();
assert_ne!(p2["error"], "body_schema_violation");
assert_ne!(p2["error"], "idempotency_key_reused_with_different_request");
assert!(
status2.is_success(),
"well-formed second request should succeed (got status {status2}, body {p2})"
);
}