use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::Duration;
use camel_api::Value;
use camel_core::RouteDefinition;
use noyalib::compat::serde_yaml;
use serde::Deserialize;
use error::{RawEndpointRef, classify_yaml_error, endpoint_from_raw, parse_duration};
pub use crate::partner_script::{PartnerFault, PartnerScript, PartnerScriptResponse};
pub use camel_matchers::RequestExpectation as PartnerExpectation;
pub use camel_matchers::{CountBound, Expectation, PathFilter, RowsExpectation};
pub mod error;
pub use error::DocError;
pub mod logs;
pub use logs::{LogLevel, LogsAssertion};
pub mod validate;
pub use validate::{ScenarioTarget, SqlTarget, ValidateExpectation};
use validate::{backticked, partner_expectation_from_value};
pub(crate) use validate::{
expectation_from_value, sql_expectation_from_value, sql_query_lacks_order_by,
};
#[derive(Debug)]
pub struct ScenarioDocument {
pub source_path: std::path::PathBuf,
pub route_source: RouteSource,
pub scenario: Vec<ScenarioAction>,
pub partners: Option<BTreeMap<String, Vec<PartnerScript>>>,
pub env: Option<BTreeMap<String, String>>,
pub env_passthrough: Option<Vec<String>>,
pub profile: Option<String>,
pub send_deadline: Option<Duration>,
pub inbound: Option<InboundListener>,
pub logs: Option<LogsAssertion>,
}
#[non_exhaustive]
pub enum RouteSource {
RouteFiles(Vec<PathBuf>),
RouteFilesFromRoot(Vec<PathBuf>),
Inline(Vec<RouteDefinition>),
}
impl std::fmt::Debug for RouteSource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::RouteFiles(files) => f.debug_tuple("RouteFiles").field(files).finish(),
Self::RouteFilesFromRoot(files) => {
f.debug_tuple("RouteFilesFromRoot").field(files).finish()
}
Self::Inline(routes) => f
.debug_tuple("Inline")
.field(&format_args!("{} route definitions", routes.len()))
.finish(),
}
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum ScenarioAction {
Send {
to: EndpointRef,
body: Option<Value>,
headers: Option<BTreeMap<String, Value>>,
method: String,
expect_reply: Option<Expectation>,
},
Receive {
from: EndpointRef,
deadline: Duration,
extract: Option<BTreeMap<String, String>>,
},
Sleep {
duration: Duration,
},
Validate {
target: ScenarioTarget,
expectation: ValidateExpectation,
deadline: Option<Duration>,
elapsed_at_least: Option<Duration>,
},
Sql {
datasource: String,
prepare: Vec<String>,
},
}
impl ScenarioAction {
fn bindings(&self) -> Vec<(&str, &str)> {
fn endpoint_bindings(endpoint: &EndpointRef) -> Vec<(&str, &str)> {
endpoint.binding().into_iter().collect()
}
match self {
Self::Send { to, .. } => endpoint_bindings(to),
Self::Receive { from, .. } => endpoint_bindings(from),
Self::Validate { target, .. } => match target {
ScenarioTarget::LastReceived(endpoint) => endpoint_bindings(endpoint),
ScenarioTarget::Partner(endpoint)
if endpoint.provisioning == Some(Provisioning::Harness) =>
{
endpoint_bindings(endpoint)
}
ScenarioTarget::Partner(_) => Vec::new(),
ScenarioTarget::Variable(_) => Vec::new(),
ScenarioTarget::Sql(_) => Vec::new(),
},
Self::Sleep { .. } => Vec::new(),
Self::Sql { .. } => Vec::new(),
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct EndpointRef {
pub endpoint: String,
pub provisioning: Option<Provisioning>,
pub bind_var: Option<String>,
}
impl EndpointRef {
fn binding(&self) -> Option<(&str, &str)> {
self.bind_var
.as_deref()
.map(|bind_var| (bind_var, self.endpoint.as_str()))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Provisioning {
Harness,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InboundListener {
pub bind_var: String,
}
#[cfg(feature = "http")]
pub fn partner_scripts_for(
doc: &ScenarioDocument,
endpoint: &str,
) -> Option<Vec<crate::adapters::http::ScriptedResponse>> {
use crate::adapters::http::ScriptedResponse;
let scripts = doc.partners.as_ref()?.get(endpoint)?;
Some(
scripts
.iter()
.map(|script| {
let (status, headers, body) = match script.response.as_ref() {
Some(response) => (
response.status.unwrap_or(200),
response.headers.clone().unwrap_or_default(),
response.body.as_ref().map_or_else(Vec::new, |value| {
crate::adapters::http::value_to_wire(value)
}),
),
None => (200, BTreeMap::new(), Vec::new()),
};
ScriptedResponse {
method: script.method.clone(),
path: script.path.clone(),
times: script.times.unwrap_or(1),
delay: script.delay,
fault: script.fault.clone(),
status,
headers,
body,
}
})
.collect(),
)
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct RawDocument {
route_files: Option<Vec<String>>,
route_files_from_root: Option<Vec<String>>,
routes: Option<serde_yaml::Value>,
scenario: Option<Vec<serde_yaml::Value>>,
env: Option<BTreeMap<String, String>>,
env_passthrough: Option<Vec<String>>,
profile: Option<String>,
partners: Option<BTreeMap<String, serde_yaml::Value>>,
send_deadline: Option<String>,
inbound: Option<serde_yaml::Value>,
logs: Option<serde_yaml::Value>,
inputs: Option<serde_yaml::Value>,
expects: Option<serde_yaml::Value>,
intercepts: Option<serde_yaml::Value>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct RawSend {
to: RawEndpointRef,
body: Option<Value>,
headers: Option<BTreeMap<String, Value>>,
method: Option<String>,
expect_reply: Option<Value>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct RawReceive {
from: RawEndpointRef,
deadline: Option<String>,
extract: Option<BTreeMap<String, String>>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct RawSleep {
duration: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct RawSqlTarget {
datasource: String,
query: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct RawValidate {
target: serde_yaml::Value,
expectation: Value,
deadline: Option<String>,
elapsed_at_least: Option<String>,
}
pub fn parse_scenario_document(path: &Path) -> Result<ScenarioDocument, DocError> {
if !camel_dsl::discovery::is_test_document(path) {
return Err(DocError::NotTestDocument {
path: path.to_path_buf(),
});
}
let text = std::fs::read_to_string(path).map_err(|source| DocError::Io {
path: path.to_path_buf(),
source,
})?;
let raw = serde_yaml::from_str::<RawDocument>(&text)
.map_err(|e| classify_yaml_error(&e.to_string()))?;
let Some(raw_scenario) = raw.scenario else {
return Err(DocError::MissingScenario);
};
if raw_scenario.is_empty() {
return Err(DocError::Validation {
index: 0,
message: "`scenario` must declare at least one action".to_string(),
});
}
let mut unit_tier: Vec<&str> = Vec::new();
if raw.inputs.is_some() {
unit_tier.push("inputs");
}
if raw.expects.is_some() {
unit_tier.push("expects");
}
if raw.intercepts.is_some() {
unit_tier.push("intercepts");
}
if !unit_tier.is_empty() {
return Err(DocError::MixedVocabulary {
found: backticked(&unit_tier),
});
}
let mut present: Vec<&'static str> = Vec::new();
if raw.route_files.is_some() {
present.push("routeFiles");
}
if raw.route_files_from_root.is_some() {
present.push("routeFilesFromRoot");
}
if raw.routes.is_some() {
present.push("routes");
}
let route_source = match present.as_slice() {
["routeFiles"] => RouteSource::RouteFiles(
raw.route_files
.unwrap_or_default()
.into_iter()
.map(PathBuf::from)
.collect(),
),
["routeFilesFromRoot"] => RouteSource::RouteFilesFromRoot(
raw.route_files_from_root
.unwrap_or_default()
.into_iter()
.map(PathBuf::from)
.collect(),
),
["routes"] => {
let value = raw.routes.unwrap_or(serde_yaml::Value::Null);
RouteSource::Inline(parse_inline_routes(&value)?)
}
[] => return Err(DocError::RouteSourceMissing),
_ => {
return Err(DocError::RouteSourceConflict {
present: backticked(&present),
});
}
};
if matches!(route_source, RouteSource::Inline(_)) {
return Err(DocError::InlineRoutesRejected);
}
let mut scenario = Vec::with_capacity(raw_scenario.len());
for (index, item) in raw_scenario.into_iter().enumerate() {
scenario.push(build_action(item, index)?);
}
let partners = crate::partner_script::partners_from_raw(raw.partners)?;
let inbound = raw.inbound.map(inbound_from_raw).transpose()?;
if let Some(env) = raw.env.as_ref() {
if let Some(inbound) = inbound.as_ref()
&& env.contains_key(&inbound.bind_var)
{
return Err(DocError::ReservedEnvKey {
key: inbound.bind_var.clone(),
endpoint: "inbound".to_string(),
});
}
for action in &scenario {
for (bind_var, endpoint) in action.bindings() {
if env.contains_key(bind_var) {
return Err(DocError::ReservedEnvKey {
key: bind_var.to_string(),
endpoint: endpoint.to_string(),
});
}
}
}
}
let mut harness_uris: Vec<&str> = Vec::new();
let mut self_declared: Vec<&str> = Vec::new();
let mut partner_targets: Vec<(usize, &EndpointRef)> = Vec::new();
for (index, action) in scenario.iter().enumerate() {
match action {
ScenarioAction::Send { to, .. } => {
if to.provisioning == Some(Provisioning::Harness) {
harness_uris.push(to.endpoint.as_str());
}
}
ScenarioAction::Receive { from, .. } => {
if from.provisioning == Some(Provisioning::Harness) {
harness_uris.push(from.endpoint.as_str());
}
}
ScenarioAction::Validate {
target: ScenarioTarget::Partner(endpoint),
..
} => {
partner_targets.push((index, endpoint));
if endpoint.provisioning == Some(Provisioning::Harness)
&& ref_scheme(&endpoint.endpoint) == Some("http")
&& partners
.as_ref()
.is_some_and(|map| map.contains_key(&endpoint.endpoint))
{
self_declared.push(endpoint.endpoint.as_str());
}
}
_ => {}
}
}
for (index, endpoint) in partner_targets {
let declared = harness_uris.contains(&endpoint.endpoint.as_str())
|| (endpoint.provisioning == Some(Provisioning::Harness)
&& self_declared.contains(&endpoint.endpoint.as_str()));
if !declared {
return Err(DocError::Validation {
index,
message: format!(
"validate `partner` target `{}` matches no harness partner: declare the URI through a `send`/`receive` reference with `provisioning: harness`, or self-declare it with an object-form target carrying `provisioning: harness` and a `partners:` entry naming the URI",
endpoint.endpoint
),
});
}
}
if let Some(partners) = &partners {
let declared = |key: &str| {
(ref_scheme(key) == Some("http") && harness_uris.contains(&key))
|| self_declared.contains(&key)
};
if let Some(key) = partners.keys().find(|key| !declared(key)) {
return Err(DocError::Validation {
index: 0,
message: format!(
"partners[{key}]: no wired harness `http` endpoint reference declares this key"
),
});
}
}
let send_deadline = raw
.send_deadline
.as_deref()
.map(|raw_deadline| parse_duration(raw_deadline, 0, "sendDeadline"))
.transpose()?;
let logs = raw.logs.map(logs::logs_from_raw).transpose()?;
Ok(ScenarioDocument {
source_path: path.to_path_buf(),
route_source,
scenario,
partners,
env: raw.env,
env_passthrough: raw.env_passthrough,
profile: raw.profile,
send_deadline,
inbound,
logs,
})
}
fn inbound_from_raw(value: serde_yaml::Value) -> Result<InboundListener, DocError> {
let section_error = |message: String| DocError::Validation { index: 0, message };
let serde_yaml::Value::Mapping(ref map) = value else {
return Err(section_error(format!(
"`inbound` must be a map with a `bindVar` key, got {value:?}"
)));
};
let mut bind_var: Option<String> = None;
for (key, value) in map {
match key.as_str() {
"bindVar" => {
let text = value.as_str().ok_or_else(|| {
section_error(format!(
"`inbound`: `bindVar` must be a string, got {value:?}"
))
})?;
bind_var = Some(text.to_string());
}
other => {
return Err(section_error(format!(
"`inbound`: unknown field `{other}`; expected `bindVar`"
)));
}
}
}
let bind_var =
bind_var.ok_or_else(|| section_error("`inbound` requires a `bindVar` key".to_string()))?;
#[cfg(not(feature = "http"))]
{
let _ = bind_var;
Err(section_error(
"`inbound` requires the `http` feature, which this harness build does \
not enable: rebuild with `--features http` (demand-gated activation)"
.to_string(),
))
}
#[cfg(feature = "http")]
Ok(InboundListener { bind_var })
}
#[cfg(feature = "sql")]
fn sql_action_from_raw(
raw: crate::sql_action::RawSqlAction,
index: usize,
) -> Result<ScenarioAction, DocError> {
let validated = crate::sql_action::validate_sql_action(&raw, index)
.map_err(|message| DocError::Validation { index, message })?;
Ok(ScenarioAction::Sql {
datasource: validated.datasource,
prepare: validated.prepare,
})
}
#[cfg(not(feature = "sql"))]
fn sql_action_from_raw(
raw: crate::sql_action::RawSqlAction,
index: usize,
) -> Result<ScenarioAction, DocError> {
if let Err(message) = crate::sql_action::validate_sql_action(&raw, index) {
return Err(DocError::Validation { index, message });
}
Err(DocError::Validation {
index,
message: "`sql` requires the `sql` feature, which this harness build does \
not enable: rebuild with `--features sql` (demand-gated activation)"
.to_string(),
})
}
fn parse_inline_routes(value: &serde_yaml::Value) -> Result<Vec<RouteDefinition>, DocError> {
let mut mapping = serde_yaml::Mapping::new();
mapping.insert("routes", value.clone());
let text = serde_yaml::to_string(&serde_yaml::Value::Mapping(mapping))
.map_err(|e| DocError::InlineRoutes(format!("failed to serialize inline routes: {e}")))?;
camel_dsl::parse_yaml(&text).map_err(|e| DocError::InlineRoutes(e.to_string()))
}
fn build_action(item: serde_yaml::Value, index: usize) -> Result<ScenarioAction, DocError> {
let action_error = |message: String| DocError::Validation { index, message };
let serde_yaml::Value::Mapping(ref map) = item else {
return Err(action_error(format!(
"action must be a single-key map (`send`, `receive`, `sleep`, `validate`, `sql`), got {item:?}"
)));
};
let Some((key, content)) = map.iter().next() else {
return Err(action_error(
"action must be a single-key map (`send`, `receive`, `sleep`, `validate`, `sql`), got an empty map"
.to_string(),
));
};
if map.len() != 1 {
return Err(action_error(format!(
"action must declare exactly one key, got {}",
backticked(&map.keys().map(String::as_str).collect::<Vec<_>>())
)));
}
let action_error_from_serde = |e: serde_yaml::Error| action_error(e.to_string());
match key.as_str() {
"send" => {
let raw: RawSend =
serde_yaml::from_value(content.clone()).map_err(action_error_from_serde)?;
let method = match raw.method {
Some(method) => {
let upper = method.trim().to_ascii_uppercase();
if !is_http_token(&upper) {
return Err(action_error(format!(
"send action `method` must be a valid HTTP method name, got `{method}`"
)));
}
upper
}
None => {
if raw.body.is_some() {
"POST".to_string()
} else {
"GET".to_string()
}
}
};
let scheme = ref_scheme(&raw.to.endpoint);
if raw.expect_reply.is_some() && scheme != Some("direct") {
return Err(DocError::ExpectReplyOnUnsupportedSend {
index,
scheme: scheme.unwrap_or("no scheme").to_string(),
});
}
let expect_reply = raw
.expect_reply
.map(|value| expectation_from_value(&value, index, "expectReply"))
.transpose()?;
Ok(ScenarioAction::Send {
to: endpoint_from_raw(raw.to)?,
body: raw.body,
headers: raw.headers,
method,
expect_reply,
})
}
"receive" => {
let raw: RawReceive =
serde_yaml::from_value(content.clone()).map_err(action_error_from_serde)?;
let deadline = raw.deadline.ok_or_else(|| {
action_error(
"receive action requires a `deadline` (humantime string, e.g. `5s`)"
.to_string(),
)
})?;
Ok(ScenarioAction::Receive {
from: endpoint_from_raw(raw.from)?,
deadline: parse_duration(&deadline, index, "deadline")?,
extract: raw.extract,
})
}
"sleep" => {
let raw: RawSleep =
serde_yaml::from_value(content.clone()).map_err(action_error_from_serde)?;
Ok(ScenarioAction::Sleep {
duration: parse_duration(&raw.duration, index, "sleep duration")?,
})
}
"validate" => {
let raw: RawValidate =
serde_yaml::from_value(content.clone()).map_err(action_error_from_serde)?;
let target = build_target(&raw.target, index)?;
let deadline = match raw.deadline.as_deref() {
None => None,
Some(raw_deadline)
if matches!(target, ScenarioTarget::Partner(_) | ScenarioTarget::Sql(_)) =>
{
Some(parse_duration(raw_deadline, index, "deadline")?)
}
Some(raw_deadline) => {
return Err(action_error(format!(
"`deadline` is only valid on a `partner` or `sql` validate target, got `{raw_deadline}`"
)));
}
};
let elapsed_at_least = match raw.elapsed_at_least.as_deref() {
None => None,
Some(raw_bound) if matches!(target, ScenarioTarget::LastReceived(_)) => {
Some(parse_duration(raw_bound, index, "elapsedAtLeast")?)
}
Some(raw_bound) => {
return Err(action_error(format!(
"`elapsedAtLeast` is only valid on a `lastReceived` validate target, got `{raw_bound}`"
)));
}
};
let expectation = match &target {
ScenarioTarget::Partner(_) => ValidateExpectation::Partner(
partner_expectation_from_value(&raw.expectation, index)?,
),
ScenarioTarget::Sql(_) => {
ValidateExpectation::Rows(sql_expectation_from_value(&raw.expectation, index)?)
}
_ => ValidateExpectation::Message(expectation_from_value(
&raw.expectation,
index,
"expectation",
)?),
};
if let (ScenarioTarget::Sql(target), ValidateExpectation::Rows(rows)) =
(&target, &expectation)
&& !rows.unordered
&& rows.rows.is_some()
&& sql_query_lacks_order_by(&target.query)
{
tracing::warn!(
"validate action {index}: sql query has no `ORDER BY`; the ordered `rows` \
assertion is nondeterministic without it — declare `unordered: true` or \
add `ORDER BY`"
);
}
Ok(ScenarioAction::Validate {
target,
expectation,
deadline,
elapsed_at_least,
})
}
crate::sql_action::SQL_ACTION_KEY => {
let raw: crate::sql_action::RawSqlAction =
serde_yaml::from_value(content.clone()).map_err(action_error_from_serde)?;
sql_action_from_raw(raw, index)
}
other => Err(action_error(format!(
"unknown action `{other}`; expected `send`, `receive`, `sleep`, `validate`, or `sql`"
))),
}
}
fn build_target(value: &serde_yaml::Value, index: usize) -> Result<ScenarioTarget, DocError> {
let action_error = |message: String| DocError::Validation { index, message };
let serde_yaml::Value::Mapping(map) = value else {
return Err(action_error(format!(
"validate `target` must be a single-key map (`lastReceived`, `variable`, `partner`, or `sql`), got {value:?}"
)));
};
let Some((key, content)) = map.iter().next() else {
return Err(action_error(
"validate `target` must be a single-key map (`lastReceived`, `variable`, `partner`, or `sql`), got an empty map"
.to_string(),
));
};
match key.as_str() {
"lastReceived" => {
let raw: RawEndpointRef =
serde_yaml::from_value(content.clone()).map_err(|e| action_error(e.to_string()))?;
Ok(ScenarioTarget::LastReceived(endpoint_from_raw(raw)?))
}
"variable" => match content.as_str() {
Some(name) => Ok(ScenarioTarget::Variable(name.to_string())),
None => Err(action_error(format!(
"validate `variable` target must be a string, got {content:?}"
))),
},
"partner" => {
let raw: RawEndpointRef =
serde_yaml::from_value(content.clone()).map_err(|e| action_error(e.to_string()))?;
Ok(ScenarioTarget::Partner(endpoint_from_raw(raw)?))
}
"sql" => {
let raw: RawSqlTarget =
serde_yaml::from_value(content.clone()).map_err(|e| action_error(e.to_string()))?;
if raw.datasource.is_empty() {
return Err(action_error(
"validate `sql` target requires a non-empty `datasource`".to_string(),
));
}
if raw.query.is_empty() {
return Err(action_error(
"validate `sql` target requires a non-empty `query`".to_string(),
));
}
if !crate::sql_action::is_read_statement(&raw.query) {
return Err(action_error(
"validate `sql` target: query is not a read (select/with prefix); reads \
belong to the validate sql target, the `sql:` prepare action owns mutations"
.to_string(),
));
}
Ok(ScenarioTarget::Sql(SqlTarget {
datasource: raw.datasource,
query: raw.query,
}))
}
other => Err(action_error(format!(
"unknown validate target `{other}`; expected `lastReceived`, `variable`, `partner`, or `sql`"
))),
}
}
fn ref_scheme(endpoint: &str) -> Option<&str> {
let (scheme, _) = endpoint.split_once(':')?;
(!scheme.is_empty()).then_some(scheme)
}
pub(crate) fn is_http_token(s: &str) -> bool {
!s.is_empty()
&& s.chars().all(|c| {
c.is_ascii_alphanumeric()
|| matches!(
c,
'!' | '#'
| '$'
| '%'
| '&'
| '\''
| '*'
| '+'
| '-'
| '.'
| '^'
| '_'
| '`'
| '|'
| '~'
)
})
}