#![cfg(feature = "http")]
use std::collections::BTreeMap;
use std::sync::Arc;
use camel_api::Value;
use camel_integration_test::runner::fill_bind_vars;
use camel_integration_test::{
DirectStimulus, DocumentOutcome, EndpointRef, HttpPartner, HttpRecorder, LayeredEnv,
PartnerAdapter, PartnerRouter, Provisioning, ScenarioAction, ScenarioDocument, ScenarioFailure,
ScenarioTarget, ScenarioVars, ScenarioVerdict, TransportError, ambient_std, boot_scenario,
parse_scenario_document, partner_scripts_for, run_scenario_document,
};
const ORDERS: &str = "http://127.0.0.1:0/orders";
const PUT_PUT_OK_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 200
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: put-ok
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
response:
status: 200
headers:
content-type: application/json
body: put-ok
"#;
const ESCAPE_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: POST
body: '$${not_a_var}'
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
"#;
const DELETE_UNMATCHED_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: DELETE
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 500
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: 200
partners:
http://127.0.0.1:0/orders:
- method: POST
path: /orders
response:
status: 200
body: nope
"#;
const EMPTY_PARTNERS_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: GET
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 500
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: 200
partners:
http://127.0.0.1:0/orders: []
"#;
const UNSET_VAR_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to: http://${missing}/orders
method: POST
"#;
const CRUD_CHAIN_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: POST
body:
sku: abc
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
orderId: body.id
- send:
to: 'http://${PARTNER}/orders/${orderId}'
method: GET
- receive:
from: 'http://${PARTNER}/orders/${orderId}'
deadline: 5s
- validate:
target:
lastReceived: 'http://${PARTNER}/orders/${orderId}'
expectation:
contains: ord-7
partners:
http://127.0.0.1:0/orders:
- method: POST
path: /orders
response:
status: 201
headers:
content-type: application/json
body:
id: ord-7
- method: GET
path: /orders/ord-7
response:
status: 200
headers:
content-type: application/json
body:
id: ord-7
"#;
const RECEIVE_INTERPOLATED_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: POST
- receive:
from: 'http://${PARTNER}/orders'
deadline: 5s
partners:
http://127.0.0.1:0/orders:
- method: POST
path: /orders
response:
status: 200
body: parked-ok
"#;
const PLAIN_STRING_BODY_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: exact text
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
response:
status: 200
body: "exact text"
"#;
const NULL_BODY_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: ''
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
response:
status: 200
body: null
"#;
const DELAY_RESPONSE_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 200
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: slow-ok
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
delay: 100ms
response:
status: 200
headers:
content-type: application/json
body: slow-ok
"#;
const ELAPSED_WAITED_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation:
equals:
ok: true
elapsedAtLeast: 200ms
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
delay: 300ms
response:
status: 200
headers:
content-type: application/json
body: {"ok": true}
"#;
const ELAPSED_EARLY_ARRIVAL_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- sleep:
duration: 1s
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation:
equals:
ok: true
elapsedAtLeast: 500ms
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
response:
status: 200
headers:
content-type: application/json
body: {"ok": true}
"#;
const ELAPSED_TOO_EARLY_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation:
equals:
ok: true
elapsedAtLeast: 10s
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
response:
status: 200
headers:
content-type: application/json
body: {"ok": true}
"#;
const FAULT_CLOSE_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
fault: close
"#;
const DELAY_BEFORE_FAULT_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
delay: 100ms
fault: close
"#;
const TIMES_TWO_FALLBACK_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 201
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: A
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 201
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: A
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: PUT
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
deadline: 5s
extract:
status: status
- validate:
target: {variable: status}
expectation: 200
- validate:
target:
lastReceived: http://127.0.0.1:0/orders
expectation: B
partners:
http://127.0.0.1:0/orders:
- method: PUT
path: /orders
times: 2
response:
status: 201
headers:
content-type: application/json
body: A
- method: PUT
path: /orders
response:
status: 200
headers:
content-type: application/json
body: B
"#;
const TWO_LAYER_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: PARTNER
method: POST
- send:
to: 'http://${PARTNER}/orders'
method: GET
- send:
to: direct:start
body: env-tier-probe
"#;
fn wired_refs(doc: &ScenarioDocument) -> Vec<EndpointRef> {
doc.scenario
.iter()
.flat_map(|action| match action {
ScenarioAction::Send { to, .. } => vec![to.clone()],
ScenarioAction::Receive { from, .. } => vec![from.clone()],
ScenarioAction::Validate {
target: ScenarioTarget::LastReceived(endpoint),
..
} => vec![endpoint.clone()],
_ => Vec::new(),
})
.collect()
}
async fn run_doc(yaml: &str) -> (DocumentOutcome, BTreeMap<String, HttpRecorder>) {
let dir = tempfile::tempdir().expect("temp dir");
let path = dir.path().join("case.test.yaml");
std::fs::write(&path, yaml).expect("write case file");
let doc = parse_scenario_document(&path).expect("document must load");
let wired = wired_refs(&doc);
let mut adapters: BTreeMap<String, Box<dyn PartnerAdapter>> = BTreeMap::new();
let mut recorders: BTreeMap<String, HttpRecorder> = BTreeMap::new();
for reference in &wired {
if reference.provisioning != Some(Provisioning::Harness)
|| !reference.endpoint.starts_with("http://")
{
continue;
}
let scripts = partner_scripts_for(&doc, &reference.endpoint);
let partner = match scripts {
Some(scripts) => HttpPartner::start(scripts).await,
None => HttpPartner::start_permissive(200).await,
}
.expect("partner must bind 127.0.0.1:0");
recorders.insert(reference.endpoint.clone(), partner.recorder());
adapters.insert(reference.endpoint.clone(), Box::new(partner));
}
let router = PartnerRouter::new(adapters);
let mut vars = ScenarioVars::new();
fill_bind_vars(&wired, &router, &mut vars);
let outcome = run_scenario_document(&doc, &router, &mut vars, None).await;
(outcome, recorders)
}
#[tokio::test]
async fn partner_direct_send_reaches_bound_address() {
let (outcome, recorders) = run_doc(PUT_PUT_OK_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"every action must pass: {outcome:?}"
);
let recorded = recorders[ORDERS].recorded_requests();
assert_eq!(recorded.len(), 1, "exactly one request must reach the wire");
assert_eq!(recorded[0].method, "PUT");
assert_eq!(recorded[0].path, "/orders");
}
#[tokio::test]
async fn escape_reaches_wire() {
let (outcome, recorders) = run_doc(ESCAPE_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"every action must pass: {outcome:?}"
);
let recorded = recorders[ORDERS].recorded_requests();
assert_eq!(recorded.len(), 1);
assert_eq!(
String::from_utf8_lossy(&recorded[0].body),
"${not_a_var}",
"the escaped leaf must reach the wire as the literal: {:?}",
recorded[0].body
);
}
#[tokio::test]
async fn unmatched_script_serves_500_empty() {
let (outcome, _recorders) = run_doc(DELETE_UNMATCHED_DOC).await;
assert_eq!(outcome.verdict, None, "the last validate must fail");
assert!(
matches!(&outcome.per_action[2], Ok(ScenarioVerdict::Pass)),
"the status validate must pass, proving the received status is 500: {outcome:?}"
);
let failure = outcome
.per_action
.last()
.and_then(|result| result.as_ref().err())
.expect("the last action must have failed");
let ScenarioFailure::ValidationMismatch { detail, .. } = failure else {
panic!("expected ValidationMismatch, got {failure:?}");
};
assert!(
detail.contains("got \"\""),
"the mismatch detail must show the empty body: {detail}"
);
}
#[tokio::test]
async fn declared_empty_partners_serves_unmatched_500() {
let (outcome, _recorders) = run_doc(EMPTY_PARTNERS_DOC).await;
assert_eq!(outcome.verdict, None, "the last validate must fail");
assert!(
matches!(&outcome.per_action[2], Ok(ScenarioVerdict::Pass)),
"the status validate must pass, proving the received status is 500: {outcome:?}"
);
let failure = outcome
.per_action
.last()
.and_then(|result| result.as_ref().err())
.expect("the last action must have failed");
let ScenarioFailure::ValidationMismatch { detail, .. } = failure else {
panic!("expected ValidationMismatch, got {failure:?}");
};
assert!(
detail.contains("got \"\""),
"the mismatch detail must show the empty body: {detail}"
);
}
#[tokio::test]
async fn unset_variable_fails_verdict() {
let (outcome, _recorders) = run_doc(UNSET_VAR_DOC).await;
assert_eq!(outcome.verdict, None);
let failure = outcome
.per_action
.last()
.and_then(|result| result.as_ref().err())
.expect("the send must fail");
assert!(
matches!(&failure, ScenarioFailure::VarUnresolved { name } if name == "missing"),
"expected VarUnresolved {{ name: \"missing\" }}, got {failure:?}"
);
}
#[tokio::test]
async fn crud_chain_interpolates_extracted_id() {
let (outcome, recorders) = run_doc(CRUD_CHAIN_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the whole chain must pass: {outcome:?}"
);
let recorded = recorders[ORDERS].recorded_requests();
assert_eq!(recorded.len(), 2, "POST then GET on the same partner");
assert_eq!(recorded[0].method, "POST");
assert_eq!(recorded[1].method, "GET");
assert_eq!(
recorded[1].path, "/orders/ord-7",
"the extracted id must be on the wire path"
);
}
#[tokio::test]
async fn receive_endpoint_interpolates() {
let (outcome, _recorders) = run_doc(RECEIVE_INTERPOLATED_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the interpolated receive must find the parked roundtrip: {outcome:?}"
);
}
#[tokio::test]
async fn delay_response_serves_e2e() {
let (outcome, recorders) = run_doc(DELAY_RESPONSE_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the delayed response must serve: {outcome:?}"
);
let recorded = recorders[ORDERS].recorded_requests();
assert_eq!(recorded.len(), 1, "exactly one request must reach the wire");
assert_eq!(recorded[0].method, "PUT");
assert_eq!(recorded[0].path, "/orders");
}
#[tokio::test]
async fn waited_arrival_passes_elapsed_bound() {
let (outcome, _recorders) = run_doc(ELAPSED_WAITED_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the delayed arrival must satisfy the elapsed bound: {outcome:?}"
);
}
#[tokio::test]
async fn early_arrival_fails_even_when_consumed_late() {
let (outcome, _recorders) = run_doc(ELAPSED_EARLY_ARRIVAL_DOC).await;
assert_eq!(
outcome.verdict, None,
"the early arrival must fail the 500ms bound: {outcome:?}"
);
let failure = outcome
.per_action
.get(3)
.and_then(|result| result.as_ref().err())
.expect("the final validate must fail");
let ScenarioFailure::ValidationMismatch { action, detail } = failure else {
panic!("expected ValidationMismatch, got {failure:?}");
};
assert_eq!(*action, 3, "the final validate is the failing action");
assert!(
detail.contains(ORDERS),
"detail must name the redacted endpoint subject: {detail}"
);
assert!(
detail.contains("500ms"),
"detail must name the elapsed bound: {detail}"
);
assert!(
detail.contains("after the scenario started"),
"detail must name the actual elapsed (humantime, any unit) \
between `arrived` and `after the scenario started`: {detail}"
);
}
#[tokio::test]
async fn too_early_arrival_fails_naming_actual() {
let (outcome, _recorders) = run_doc(ELAPSED_TOO_EARLY_DOC).await;
assert_eq!(
outcome.verdict, None,
"the 10s bound must fail an immediate arrival: {outcome:?}"
);
let failure = outcome
.per_action
.get(2)
.and_then(|result| result.as_ref().err())
.expect("the final validate must fail");
let ScenarioFailure::ValidationMismatch { action, detail } = failure else {
panic!("expected ValidationMismatch, got {failure:?}");
};
assert_eq!(*action, 2, "the final validate is the failing action");
assert!(
detail.contains(ORDERS),
"detail must name the redacted endpoint subject: {detail}"
);
assert!(
detail.contains("10s"),
"detail must name the bound: {detail}"
);
assert!(
detail.contains("after the scenario started"),
"detail must name the actual elapsed (humantime, any unit) \
between `arrived` and `after the scenario started`: {detail}"
);
}
#[tokio::test]
async fn plain_string_body_served_verbatim() {
let (outcome, _recorders) = run_doc(PLAIN_STRING_BODY_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the string body must serve verbatim: {outcome:?}"
);
}
#[tokio::test]
async fn null_body_serves_empty() {
let (outcome, _recorders) = run_doc(NULL_BODY_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the null body must serve empty: {outcome:?}"
);
}
fn assert_receive_transport_failure(outcome: &DocumentOutcome, recorder: &HttpRecorder) {
assert_eq!(outcome.verdict, None, "the receive must fail");
assert!(
matches!(&outcome.per_action[0], Ok(ScenarioVerdict::Pass)),
"the send must dial and park the roundtrip: {outcome:?}"
);
let recorded = recorder.recorded_requests();
assert_eq!(recorded.len(), 1, "exactly one request must reach the wire");
assert_eq!(recorded[0].method, "PUT");
assert_eq!(recorded[0].path, "/orders");
let failure = outcome
.per_action
.get(1)
.and_then(|result| result.as_ref().err())
.expect("the receive must fail");
let ScenarioFailure::ActionTransport { action, source } = failure else {
panic!("expected ActionTransport, got {failure:?}");
};
assert_eq!(*action, 1, "the receive is the failing action");
let TransportError::Other { message } = source else {
panic!("expected a transport failure, got {source:?}");
};
assert!(
message.contains("connection closed"),
"the transport message should name the connection closure: {message}"
);
}
#[tokio::test]
async fn fault_close_fails_receive_e2e() {
let (outcome, recorders) = run_doc(FAULT_CLOSE_DOC).await;
assert_receive_transport_failure(&outcome, &recorders[ORDERS]);
}
#[tokio::test]
async fn delay_before_fault_fails_receive_e2e() {
let (outcome, recorders) = run_doc(DELAY_BEFORE_FAULT_DOC).await;
assert_receive_transport_failure(&outcome, &recorders[ORDERS]);
}
#[tokio::test]
async fn times_two_then_fallback_e2e() {
let (outcome, recorders) = run_doc(TIMES_TWO_FALLBACK_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the times-then-fallback chain must pass: {outcome:?}"
);
let recorded = recorders[ORDERS].recorded_requests();
assert_eq!(recorded.len(), 3, "three sends must reach the wire");
for request in &recorded {
assert_eq!(request.method, "PUT");
assert_eq!(request.path, "/orders");
}
}
#[tokio::test]
async fn two_layer_bindvar_both_visible() {
let dir = tempfile::tempdir().expect("temp dir");
let root = dir.path();
std::fs::write(
root.join("Camel.toml"),
"log_level = \"info\"\n\n[components.http]\nallow_internal = true\n",
)
.expect("write Camel.toml");
std::fs::write(
root.join("routes.yaml"),
"routes:\n - id: env-tier-probe\n from: direct:start\n steps:\n - to: ${env:PARTNER}\n",
)
.expect("write routes.yaml");
let path = root.join("case.test.yaml");
std::fs::write(&path, TWO_LAYER_DOC).expect("write case file");
let doc = parse_scenario_document(&path).expect("document must load");
let partner = HttpPartner::start_permissive(200)
.await
.expect("partner must bind 127.0.0.1:0");
let recorder = partner.recorder();
let harness_provisioned = BTreeMap::from([(
"PARTNER".to_string(),
format!("http://{}", partner.bound_addr()),
)]);
let env = LayeredEnv::new(
doc.env.clone().unwrap_or_default(),
harness_provisioned,
doc.env_passthrough.clone().unwrap_or_default(),
ambient_std(),
);
let run = boot_scenario(&doc, root, &env)
.await
.expect("the full boot must succeed");
let ctx = Arc::new(tokio::sync::Mutex::new(run.ctx));
let mut adapters: BTreeMap<String, Box<dyn PartnerAdapter>> = BTreeMap::new();
adapters.insert(
"direct:start".to_string(),
Box::new(DirectStimulus::new(Arc::clone(&ctx))),
);
adapters.insert(ORDERS.to_string(), Box::new(partner));
let router = PartnerRouter::new(adapters);
let mut vars = ScenarioVars::new();
fill_bind_vars(&wired_refs(&doc), &router, &mut vars);
let outcome = run_scenario_document(&doc, &router, &mut vars, None).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"both layers must be visible under one name: {outcome:?}"
);
let recorded = recorder.recorded_requests();
assert_eq!(recorded.len(), 3, "two scenario sends plus the route dial");
assert_eq!(recorded[0].method, "POST");
assert_eq!(recorded[0].path, "/orders");
assert_eq!(recorded[1].method, "GET");
assert_eq!(recorded[1].path, "/orders");
assert_eq!(
recorded[2].path, "/",
"only the env-tier producer dial arrives with no path: {:?}",
recorded[2]
);
let mut guard = ctx.lock().await;
run.boot
.shutdown(&mut guard)
.await
.expect("clean shutdown must complete");
}
const RECEIVE_LANE_ORDERS: &str = "http://127.0.0.1:0/orders";
const SIBLING_PATH_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to: direct:orders
- send:
to: direct:billing
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: MOCK
deadline: 2s
- receive:
from: 'http://${MOCK}/billing'
deadline: 2s
"#;
const BARE_AUTHORITY_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- receive:
from: 'http://${MOCK}'
deadline: 300ms
"#;
const QUERY_MATCH_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to: direct:probe
- send:
to: direct:probe
- receive:
from: 'http://${MOCK}/api?x=1'
deadline: 2s
- receive:
from: 'http://${MOCK}/api?x=2'
deadline: 300ms
"#;
const DECLARED_KEY_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to: direct:probe
- receive:
from:
endpoint: http://127.0.0.1:0/orders
provisioning: harness
bindVar: MOCK
deadline: 2s
"#;
const SECRET_QUERY_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- receive:
from: 'http://${MOCK}?authPassword=x'
deadline: 300ms
"#;
const ROUNDTRIP_STANDALONE_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to: 'http://${MOCK}/orders'
method: POST
- receive:
from: 'http://${MOCK}/orders'
deadline: 2s
- validate:
target:
lastReceived: 'http://${MOCK}/orders'
expectation:
contains: parked-ok
partners:
http://127.0.0.1:0/orders:
- method: POST
path: /orders
response:
status: 200
body: parked-ok
"#;
const PATH_AWARE_DOC: &str = r#"
routeFiles: [routes.yaml]
scenario:
- send:
to: 'http://${MOCK}/a'
method: POST
- send:
to: 'http://${MOCK}/b'
method: POST
- receive:
from: 'http://${MOCK}/b'
deadline: 2s
- receive:
from: 'http://${MOCK}/a'
deadline: 2s
- validate:
target:
lastReceived: 'http://${MOCK}/b'
expectation:
contains: b-ok
- validate:
target:
lastReceived: 'http://${MOCK}/a'
expectation:
contains: a-ok
partners:
http://127.0.0.1:0/orders:
- method: POST
path: /a
response:
status: 200
body: a-ok
- method: POST
path: /b
response:
status: 200
body: b-ok
"#;
fn single_dial_routes(dials: &[(&str, &str)]) -> String {
let routes: Vec<String> = dials
.iter()
.map(|(stimulus, target)| {
format!(
" - id: dial-{stimulus}\n from: direct:{stimulus}\n steps:\n - to: ${{env:{target}}}\n"
)
})
.collect();
format!("routes:\n{}", routes.join(""))
}
async fn run_doc_one_partner(yaml: &str) -> (DocumentOutcome, HttpRecorder) {
let dir = tempfile::tempdir().expect("temp dir");
let path = dir.path().join("case.test.yaml");
std::fs::write(&path, yaml).expect("write case file");
let doc = parse_scenario_document(&path).expect("document must load");
let scripts = partner_scripts_for(&doc, RECEIVE_LANE_ORDERS);
let partner = match scripts {
Some(scripts) => HttpPartner::start(scripts).await,
None => HttpPartner::start_permissive(200).await,
}
.expect("partner must bind 127.0.0.1:0");
let bound = partner.bound_addr().to_string();
let recorder = partner.recorder();
let mut adapters: BTreeMap<String, Box<dyn PartnerAdapter>> = BTreeMap::new();
adapters.insert(RECEIVE_LANE_ORDERS.to_string(), Box::new(partner));
let router = PartnerRouter::new(adapters);
router.set_secret_query_keys(vec!["authPassword".to_string()]);
let mut vars = ScenarioVars::new();
vars.set("MOCK", Value::String(bound));
let outcome = run_scenario_document(&doc, &router, &mut vars, None).await;
(outcome, recorder)
}
async fn run_doc_route_dialed(
yaml: &str,
routes_yaml: &str,
dial_paths: &[&str],
) -> (DocumentOutcome, HttpRecorder) {
let dir = tempfile::tempdir().expect("temp dir");
let root = dir.path();
std::fs::write(
root.join("Camel.toml"),
"log_level = \"info\"\n\n[components.http]\nallow_internal = true\n",
)
.expect("write Camel.toml");
std::fs::write(root.join("routes.yaml"), routes_yaml).expect("write routes.yaml");
let path = root.join("case.test.yaml");
std::fs::write(&path, yaml).expect("write case file");
let doc = parse_scenario_document(&path).expect("document must load");
let scripts = partner_scripts_for(&doc, RECEIVE_LANE_ORDERS);
let partner = match scripts {
Some(scripts) => HttpPartner::start(scripts).await,
None => HttpPartner::start_permissive(200).await,
}
.expect("partner must bind 127.0.0.1:0");
let bound = partner.bound_addr().to_string();
let recorder = partner.recorder();
let harness_provisioned: BTreeMap<String, String> = dial_paths
.iter()
.enumerate()
.map(|(n, suffix)| (format!("MOCK_DIAL_{n}"), format!("http://{bound}{suffix}")))
.collect();
let env = LayeredEnv::new(
doc.env.clone().unwrap_or_default(),
harness_provisioned,
doc.env_passthrough.clone().unwrap_or_default(),
ambient_std(),
);
let run = boot_scenario(&doc, root, &env)
.await
.expect("the full boot must succeed");
let ctx = Arc::new(tokio::sync::Mutex::new(run.ctx));
let mut adapters: BTreeMap<String, Box<dyn PartnerAdapter>> = BTreeMap::new();
for stimulus in ["direct:orders", "direct:billing", "direct:probe"] {
adapters.insert(
stimulus.to_string(),
Box::new(DirectStimulus::new(Arc::clone(&ctx))),
);
}
adapters.insert(RECEIVE_LANE_ORDERS.to_string(), Box::new(partner));
let router = PartnerRouter::new(adapters);
router.set_secret_query_keys(vec!["authPassword".to_string()]);
let mut vars = ScenarioVars::new();
fill_bind_vars(&wired_refs(&doc), &router, &mut vars);
vars.set("MOCK", Value::String(bound));
let outcome = run_scenario_document(&doc, &router, &mut vars, None).await;
let mut guard = ctx.lock().await;
run.boot
.shutdown(&mut guard)
.await
.expect("clean shutdown must complete");
(outcome, recorder)
}
#[tokio::test]
async fn dynamic_receive_sibling_path_drains_own_lane() {
let routes = single_dial_routes(&[("orders", "MOCK_DIAL_0"), ("billing", "MOCK_DIAL_1")]);
let (outcome, recorders) =
run_doc_route_dialed(SIBLING_PATH_DOC, &routes, &["/orders", "/billing"]).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the billing receive must drain the billing lane: {outcome:?}"
);
let recorded = recorders.recorded_requests();
assert_eq!(
recorded.len(),
2,
"both route dials must reach the wire: {recorded:?}"
);
assert_eq!(recorded[0].path, "/orders");
assert_eq!(recorded[1].path, "/billing");
}
#[tokio::test]
async fn dynamic_receive_bare_authority_is_apparatus() {
let (outcome, _recorders) = run_doc_one_partner(BARE_AUTHORITY_DOC).await;
assert_eq!(outcome.verdict, None, "the receive must fail");
let failure = outcome
.per_action
.first()
.and_then(|result| result.as_ref().err())
.expect("the receive must have failed");
let ScenarioFailure::ActionTransport { action, source } = failure else {
panic!("expected an apparatus transport failure, got {failure:?}");
};
assert_eq!(*action, 0, "the receive is the failing action");
let TransportError::Other { message } = source else {
panic!("expected a transport failure, got {source:?}");
};
assert!(
message.contains("empty or absent path"),
"the error must name the empty path: {message}"
);
assert!(
message.contains("http://127.0.0.1:"),
"the error must name the declaration: {message}"
);
}
#[tokio::test]
async fn dynamic_receive_query_matches_wire_path_and_query() {
let routes = single_dial_routes(&[("probe", "MOCK_DIAL_0")]);
let (outcome, _recorders) = run_doc_route_dialed(QUERY_MATCH_DOC, &routes, &["/api?x=1"]).await;
assert_eq!(outcome.verdict, None, "the ?x=2 receive must time out");
assert!(
matches!(&outcome.per_action[2], Ok(ScenarioVerdict::Pass)),
"the exact path-and-query receive must drain its lane: {outcome:?}"
);
let failure = outcome
.per_action
.get(3)
.and_then(|result| result.as_ref().err())
.expect("the ?x=2 receive must have timed out");
let ScenarioFailure::ReceiveTimeout { lanes, .. } = failure else {
panic!("expected a receive-timeout, got {failure:?}");
};
assert!(
lanes.contains("/api?x=1"),
"the timeout must list the arrived wire lane: {lanes}"
);
}
#[tokio::test]
async fn declared_key_receive_unchanged() {
let routes = single_dial_routes(&[("probe", "MOCK_DIAL_0")]);
let (outcome, _recorders) = run_doc_route_dialed(DECLARED_KEY_DOC, &routes, &["/orders"]).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the declared-key receive must drain its lane: {outcome:?}"
);
}
#[tokio::test]
async fn bare_authority_secret_query_redacted() {
let (outcome, _recorders) = run_doc_one_partner(SECRET_QUERY_DOC).await;
assert_eq!(outcome.verdict, None, "the receive must fail");
let failure = outcome
.per_action
.first()
.and_then(|result| result.as_ref().err())
.expect("the receive must have failed");
let ScenarioFailure::ActionTransport { source, .. } = failure else {
panic!("expected an apparatus transport failure, got {failure:?}");
};
let TransportError::Other { message } = source else {
panic!("expected a transport failure, got {source:?}");
};
assert!(
message.contains("empty or absent path"),
"the error must name the empty path: {message}"
);
assert!(
!message.contains("authPassword=x"),
"the raw secret value must never print: {message}"
);
assert!(
message.contains("authPassword=***"),
"the secret value must render masked: {message}"
);
}
#[tokio::test]
async fn roundtrip_receive_first_then_take_still_works() {
let (outcome, recorders) = run_doc_one_partner(ROUNDTRIP_STANDALONE_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"the standalone receive must drain the parked roundtrip: {outcome:?}"
);
let recorded = recorders.recorded_requests();
assert_eq!(recorded.len(), 1, "exactly one request must reach the wire");
assert_eq!(recorded[0].path, "/orders");
}
#[tokio::test]
async fn standalone_roundtrip_receives_drain_their_own_path() {
let (outcome, recorders) = run_doc_one_partner(PATH_AWARE_DOC).await;
assert_eq!(
outcome.verdict,
Some(ScenarioVerdict::Pass),
"each receive must drain its own path's parked roundtrip: {outcome:?}"
);
let recorded = recorders.recorded_requests();
assert_eq!(recorded.len(), 2, "both sends must reach the wire");
assert_eq!(recorded[0].path, "/a");
assert_eq!(recorded[1].path, "/b");
}