use std::collections::BTreeMap;
use std::time::{Duration, Instant};
use camel_api::Value;
use crate::adapters::redact_wire_path;
use crate::adapters::{
IncomingMessage, OutgoingMessage, PartnerRouter, ReceiveError, TransportError, lanes_suffix,
};
use crate::document::{
EndpointRef, Expectation, Provisioning, ScenarioAction, ScenarioDocument, ScenarioTarget,
ValidateExpectation,
};
mod partner_validate;
use partner_validate::partner_validate_action;
#[cfg(all(test, feature = "http"))]
pub(crate) use partner_validate::{
matching_requests, partner_mismatch_detail, render_bound, render_filters,
};
const SEND_DEADLINE: Duration = Duration::from_secs(30);
#[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:?}{lanes}")]
ReceiveTimeout {
endpoint: String,
deadline: Duration,
lanes: String,
},
#[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> {
let started_at = Instant::now();
for (index, action) in doc.scenario.iter().enumerate() {
run_action(action, index, router, vars, started_at).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 started_at = Instant::now();
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, started_at).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,
started_at: Instant,
) -> 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 { .. } => {
validate_action(action, index, started_at, 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(timeout) => {
let keys = router.secret_query_keys();
ScenarioFailure::ReceiveTimeout {
endpoint: redact_wire_path(&timeout.endpoint, &keys),
deadline,
lanes: lanes_suffix(
&timeout
.lanes_recorded
.iter()
.map(|lane| redact_wire_path(lane, &keys))
.collect::<Vec<_>>(),
),
}
}
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(
action: &ScenarioAction,
index: usize,
started_at: Instant,
router: &PartnerRouter,
vars: &ScenarioVars,
) -> Result<(), ScenarioFailure> {
let ScenarioAction::Validate {
target,
expectation,
deadline,
elapsed_at_least,
} = action
else {
return Err(unpaired_validate(index));
};
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) => {
let redacted =
redact_wire_path(&endpoint.endpoint, &router.secret_query_keys());
let message = vars.last_received(&endpoint.endpoint).ok_or_else(|| {
ScenarioFailure::ValidationMismatch {
action: index,
detail: format!(
"no message has been received on {redacted} to validate"
),
}
})?;
if let Some(bound) = elapsed_at_least {
let actual = message
.arrival
.checked_duration_since(started_at)
.unwrap_or_default();
if actual < *bound {
return Err(ScenarioFailure::ValidationMismatch {
action: index,
detail: format!(
"{redacted}: arrived {} after the scenario started; `elapsedAtLeast` requires {}",
humantime::format_duration(actual),
humantime::format_duration(*bound)
),
});
}
}
(
message.body.clone(),
format!("body last received on {redacted}"),
)
}
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(),
}
}
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,
}
}