#![cfg(feature = "http")]
use std::collections::BTreeMap;
use std::sync::Arc;
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 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 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).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");
}
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).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");
}