use std::collections::BTreeMap;
use std::time::Duration;
use camel_api::Value;
#[cfg(feature = "http")]
use crate::adapters::http::HttpWireRequest;
use crate::adapters::{
IncomingMessage, OutgoingMessage, PartnerRouter, ReceiveError, TransportError,
};
use crate::document::{
EndpointRef, Expectation, PartnerExpectation, Provisioning, ScenarioAction, ScenarioDocument,
ScenarioTarget, ValidateExpectation,
};
const SEND_DEADLINE: Duration = Duration::from_secs(30);
#[cfg(feature = "http")]
const PARTNER_POLL_INTERVAL: Duration = Duration::from_millis(100);
#[derive(Debug, Default)]
pub struct ScenarioVars {
variables: BTreeMap<String, Value>,
last_received: BTreeMap<String, IncomingMessage>,
}
impl ScenarioVars {
pub fn new() -> Self {
Self::default()
}
pub fn get(&self, name: &str) -> Option<&Value> {
self.variables.get(name)
}
pub fn set(&mut self, name: impl Into<String>, value: Value) {
self.variables.insert(name.into(), value);
}
pub fn last_received(&self, endpoint: &str) -> Option<&IncomingMessage> {
self.last_received.get(endpoint)
}
fn remember(&mut self, endpoint: String, message: IncomingMessage) {
self.last_received.insert(endpoint, message);
}
}
pub(crate) fn resolve_placeholders(
input: &str,
vars: &ScenarioVars,
) -> Result<String, ScenarioFailure> {
let bytes = input.as_bytes();
let mut out = Vec::with_capacity(bytes.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'$' {
if i + 2 < bytes.len() && bytes[i + 1] == b'$' && bytes[i + 2] == b'{' {
out.extend_from_slice(b"${");
i += 3;
continue;
}
if i + 1 < bytes.len() && bytes[i + 1] == b'{' {
let name_start = i + 2;
let mut j = name_start;
while j < bytes.len() && (bytes[j].is_ascii_alphanumeric() || bytes[j] == b'_') {
j += 1;
}
if j > name_start && j < bytes.len() && bytes[j] == b'}' {
let name = &input[name_start..j];
match vars.get(name) {
Some(value) => {
let replacement = stringify(value);
out.extend_from_slice(replacement.as_bytes());
i = j + 1;
continue;
}
None => {
return Err(ScenarioFailure::VarUnresolved {
name: name.to_string(),
});
}
}
}
}
out.push(b'$');
i += 1;
continue;
}
out.push(bytes[i]);
i += 1;
}
Ok(String::from_utf8(out).expect("placeholder output preserves input UTF-8")) }
pub(crate) fn interpolate_value(
value: &Value,
vars: &ScenarioVars,
) -> Result<Value, ScenarioFailure> {
match value {
Value::String(text) => Ok(Value::String(resolve_placeholders(text, vars)?)),
Value::Array(items) => items
.iter()
.map(|item| interpolate_value(item, vars))
.collect::<Result<Vec<_>, _>>()
.map(Value::Array),
Value::Object(map) => {
let rebuilt = map
.iter()
.map(|(key, item)| Ok((key.clone(), interpolate_value(item, vars)?)))
.collect::<Result<_, _>>()?;
Ok(Value::Object(rebuilt))
}
other => Ok(other.clone()),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ScenarioVerdict {
Pass,
}
#[derive(Debug, Clone, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum ScenarioFailure {
#[error("receive-timeout: {endpoint} delivered nothing within {deadline:?}")]
ReceiveTimeout {
endpoint: String,
deadline: Duration,
},
#[error("validation-mismatch: action {action}: {detail}")]
ValidationMismatch {
action: usize,
detail: String,
},
#[error("scenario-var-unresolved: {name}")]
VarUnresolved {
name: String,
},
#[error("action-transport-failure: action {action}: {source}")]
ActionTransport {
action: usize,
source: TransportError,
},
#[error("partner-startup-failure: {message}")]
PartnerStartup {
message: String,
},
#[error("shutdown-failure: {message}")]
ShutdownFailure {
message: String,
},
}
pub fn fill_bind_vars(wired: &[EndpointRef], router: &PartnerRouter, vars: &mut ScenarioVars) {
for reference in wired {
if reference.provisioning != Some(Provisioning::Harness) {
continue;
}
let Some(bind_var) = reference.bind_var.as_deref() else {
continue;
};
let Some(authority) = router
.adapter(&reference.endpoint)
.and_then(|adapter| adapter.bound_authority())
else {
continue;
};
vars.set(bind_var, Value::String(authority));
}
}
pub async fn run_scenario(
doc: &ScenarioDocument,
router: &PartnerRouter,
vars: &mut ScenarioVars,
) -> Result<ScenarioVerdict, ScenarioFailure> {
for (index, action) in doc.scenario.iter().enumerate() {
run_action(action, index, router, vars).await?;
}
Ok(ScenarioVerdict::Pass)
}
#[derive(Debug, Clone, PartialEq)]
pub struct DocumentOutcome {
pub per_action: Vec<Result<ScenarioVerdict, ScenarioFailure>>,
pub verdict: Option<ScenarioVerdict>,
pub final_failure: Option<ScenarioFailure>,
}
pub async fn run_scenario_document(
doc: &ScenarioDocument,
router: &PartnerRouter,
vars: &mut ScenarioVars,
) -> DocumentOutcome {
let mut per_action = Vec::with_capacity(doc.scenario.len());
let mut failed = false;
for (index, action) in doc.scenario.iter().enumerate() {
if failed {
break;
}
match run_action(action, index, router, vars).await {
Ok(()) => per_action.push(Ok(ScenarioVerdict::Pass)),
Err(failure) => {
per_action.push(Err(failure));
failed = true;
}
}
}
let verdict = if failed {
None
} else {
Some(ScenarioVerdict::Pass)
};
DocumentOutcome {
per_action,
verdict,
final_failure: None,
}
}
async fn run_action(
action: &ScenarioAction,
index: usize,
router: &PartnerRouter,
vars: &mut ScenarioVars,
) -> Result<(), ScenarioFailure> {
match action {
ScenarioAction::Send {
to,
body,
headers,
method,
} => {
send_action(
index,
to,
body.as_ref(),
headers.as_ref(),
method,
router,
vars,
)
.await?;
}
ScenarioAction::Receive {
from,
deadline,
extract,
} => {
receive_action(index, from, *deadline, extract.as_ref(), router, vars).await?;
}
ScenarioAction::Sleep { duration } => {
tokio::time::sleep(*duration).await;
}
ScenarioAction::Validate {
target,
expectation,
deadline,
} => {
validate_action(index, target, expectation, *deadline, router, vars).await?;
}
}
Ok(())
}
async fn send_action(
index: usize,
to: &EndpointRef,
body: Option<&Value>,
headers: Option<&BTreeMap<String, Value>>,
method: &str,
router: &PartnerRouter,
vars: &ScenarioVars,
) -> Result<(), ScenarioFailure> {
let declared = to.endpoint.as_str();
let interpolated = resolve_placeholders(declared, vars)?;
let body = body
.map(|value| interpolate_value(value, vars))
.transpose()?;
let headers = headers
.map(|map| -> Result<BTreeMap<String, Value>, ScenarioFailure> {
map.iter()
.map(|(name, value)| Ok((name.clone(), interpolate_value(value, vars)?)))
.collect()
})
.transpose()?;
let msg = OutgoingMessage {
body: body.unwrap_or(Value::Null),
headers: headers.unwrap_or_default(),
method: method.to_string(),
};
let bounded =
tokio::time::timeout(SEND_DEADLINE, router.send(declared, &interpolated, msg)).await;
let sent = bounded.map_err(|_| ScenarioFailure::ActionTransport {
action: index,
source: TransportError::Deadline {
after: SEND_DEADLINE,
},
})?;
sent.map_err(|source| ScenarioFailure::ActionTransport {
action: index,
source,
})
}
async fn receive_action(
index: usize,
from: &EndpointRef,
deadline: Duration,
extract: Option<&BTreeMap<String, String>>,
router: &PartnerRouter,
vars: &mut ScenarioVars,
) -> Result<(), ScenarioFailure> {
let declared = from.endpoint.as_str();
let interpolated = resolve_placeholders(declared, vars)?;
let message = router
.receive(declared, &interpolated, deadline)
.await
.map_err(|source| match source {
ReceiveError::Timeout(_) => ScenarioFailure::ReceiveTimeout {
endpoint: from.endpoint.clone(),
deadline,
},
ReceiveError::Transport(source) => ScenarioFailure::ActionTransport {
action: index,
source,
},
})?;
if let Some(extract) = extract {
for (name, selector) in extract {
let value = select_from(&message, selector).ok_or_else(|| {
ScenarioFailure::ValidationMismatch {
action: index,
detail: format!(
"extract of `{selector}` into variable `{name}` resolved to nothing"
),
}
})?;
vars.set(name.clone(), value);
}
}
vars.remember(from.endpoint.clone(), message);
Ok(())
}
async fn validate_action(
index: usize,
target: &ScenarioTarget,
expectation: &ValidateExpectation,
deadline: Option<Duration>,
router: &PartnerRouter,
vars: &ScenarioVars,
) -> Result<(), ScenarioFailure> {
match (target, expectation) {
(ScenarioTarget::Partner(endpoint), ValidateExpectation::Partner(expected)) => {
partner_validate_action(index, &endpoint.endpoint, expected, deadline, router).await
}
(_, ValidateExpectation::Message(expectation)) => {
let (value, subject) = match target {
ScenarioTarget::LastReceived(endpoint) => (
vars.last_received(&endpoint.endpoint)
.map(|message| message.body.clone())
.ok_or_else(|| ScenarioFailure::ValidationMismatch {
action: index,
detail: format!(
"no message has been received on {} to validate",
endpoint.endpoint
),
})?,
format!("body last received on {}", endpoint.endpoint),
),
ScenarioTarget::Variable(name) => (
vars.get(name)
.cloned()
.ok_or_else(|| ScenarioFailure::VarUnresolved { name: name.clone() })?,
format!("variable `{name}`"),
),
ScenarioTarget::Partner(_) => return Err(unpaired_validate(index)),
};
match expectation {
Expectation::Equals(expected) => {
if &value == expected {
Ok(())
} else {
Err(ScenarioFailure::ValidationMismatch {
action: index,
detail: format!("{subject}: expected {expected}, got {value}"),
})
}
}
Expectation::Regex(pattern) => {
let regex = regex::Regex::new(pattern).map_err(|error| {
ScenarioFailure::ValidationMismatch {
action: index,
detail: format!("invalid regex `{pattern}`: {error}"),
}
})?;
if regex.is_match(&stringify(&value)) {
Ok(())
} else {
Err(ScenarioFailure::ValidationMismatch {
action: index,
detail: format!("{subject}: `{pattern}` did not match {value}"),
})
}
}
Expectation::Contains(needle) => check(
index,
stringify(&value).contains(needle),
format!("{subject}: did not contain `{needle}`: {value}"),
),
Expectation::StartsWith(prefix) => check(
index,
stringify(&value).starts_with(prefix),
format!("{subject}: did not start with `{prefix}`: {value}"),
),
Expectation::EndsWith(suffix) => check(
index,
stringify(&value).ends_with(suffix),
format!("{subject}: did not end with `{suffix}`: {value}"),
),
Expectation::Exists => {
if value == Value::Null {
Err(ScenarioFailure::ValidationMismatch {
action: index,
detail: format!("{subject}: expected a value, got null"),
})
} else {
Ok(())
}
}
Expectation::JsonSubset(pattern) => check(
index,
json_subset(pattern, &value),
format!("{subject}: not a superset of {pattern}: {value}"),
),
}
}
_ => Err(unpaired_validate(index)),
}
}
fn unpaired_validate(index: usize) -> ScenarioFailure {
ScenarioFailure::ValidationMismatch {
action: index,
detail: "validate target kind does not pair with the expectation kind".to_string(),
}
}
#[cfg(feature = "http")]
pub(crate) fn matching_requests(
requests: &[HttpWireRequest],
method: Option<&str>,
path: Option<&str>,
) -> usize {
requests
.iter()
.filter(|request| {
method.is_none_or(|m| m.eq_ignore_ascii_case(&request.method))
&& path.is_none_or(|p| p == request.path)
})
.count()
}
#[cfg(feature = "http")]
async fn partner_validate_action(
index: usize,
uri: &str,
expected: &PartnerExpectation,
deadline: Option<Duration>,
router: &PartnerRouter,
) -> Result<(), ScenarioFailure> {
let expected_count = usize::try_from(expected.count).unwrap_or(usize::MAX);
let count = || {
matching_requests(
&router.recorded_requests(uri),
expected.method.as_deref(),
expected.path.as_deref(),
)
};
let mismatch = |actual: usize| ScenarioFailure::ValidationMismatch {
action: index,
detail: partner_mismatch_detail(uri, expected, actual),
};
match deadline {
None => {
let actual = count();
if actual == expected_count {
Ok(())
} else {
Err(mismatch(actual))
}
}
Some(deadline) => {
let until = tokio::time::Instant::now() + deadline;
loop {
if count() == expected_count {
return Ok(());
}
let now = tokio::time::Instant::now();
if now >= until {
let actual = count();
return if actual == expected_count {
Ok(())
} else {
Err(mismatch(actual))
};
}
tokio::time::sleep((until - now).min(PARTNER_POLL_INTERVAL)).await;
}
}
}
}
#[cfg(not(feature = "http"))]
async fn partner_validate_action(
index: usize,
uri: &str,
expected: &PartnerExpectation,
deadline: Option<Duration>,
router: &PartnerRouter,
) -> Result<(), ScenarioFailure> {
let _ = (uri, expected, deadline, router);
Err(ScenarioFailure::ValidationMismatch {
action: index,
detail: "partner validation requires the `http` feature".to_string(),
})
}
#[cfg(feature = "http")]
fn partner_mismatch_detail(uri: &str, expected: &PartnerExpectation, actual: usize) -> String {
let mut detail = format!("partner {uri}");
if expected.method.is_some() || expected.path.is_some() {
let filters = [
expected
.method
.as_deref()
.map(|method| format!("method {method}")),
expected.path.as_deref().map(|path| format!("path {path}")),
];
let joined = filters.into_iter().flatten().collect::<Vec<_>>().join(", ");
detail.push_str(&format!(" ({joined})"));
}
detail.push_str(&format!(", expected {}, actual {actual}", expected.count));
detail
}
fn check(index: usize, passed: bool, detail: String) -> Result<(), ScenarioFailure> {
if passed {
Ok(())
} else {
Err(ScenarioFailure::ValidationMismatch {
action: index,
detail,
})
}
}
fn select_from(message: &IncomingMessage, selector: &str) -> Option<Value> {
let (head, rest) = match selector.split_once('.') {
Some((head, rest)) => (head, Some(rest)),
None => (selector, None),
};
match head {
"body" => match rest {
None => Some(message.body.clone()),
Some(path) => walk_path(&message.body, path),
},
"headers" => match rest {
None => Some(Value::Object(
message
.headers
.iter()
.map(|(name, value)| (name.clone(), value.clone()))
.collect(),
)),
Some(name) => lookup_header(&message.headers, name),
},
"status" if rest.is_none() => Some(
message
.status
.map_or(Value::Null, |code| Value::Number(code.into())),
),
"method" if rest.is_none() => {
Some(message.method.clone().map_or(Value::Null, Value::String))
}
"path" if rest.is_none() => Some(message.path.clone().map_or(Value::Null, Value::String)),
_ => None,
}
}
fn lookup_header(headers: &BTreeMap<String, Value>, name: &str) -> Option<Value> {
headers
.iter()
.find(|(key, _)| key.eq_ignore_ascii_case(name))
.map(|(_, value)| value.clone())
}
fn walk_path(value: &Value, path: &str) -> Option<Value> {
let mut current = value;
for key in path.split('.') {
current = current.as_object()?.get(key)?;
}
Some(current.clone())
}
fn stringify(value: &Value) -> String {
match value {
Value::String(text) => text.clone(),
other => other.to_string(),
}
}
fn json_subset(pattern: &Value, actual: &Value) -> bool {
match (pattern, actual) {
(Value::Object(pattern_object), Value::Object(actual_object)) => {
pattern_object.iter().all(|(key, pattern_value)| {
actual_object
.get(key)
.is_some_and(|actual_value| json_subset(pattern_value, actual_value))
})
}
_ => pattern == actual,
}
}