use regex::Regex;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, Instant};
pub const DEFAULT_TIMEOUT_SECS: u64 = 90;
const DIRECTIVE_KEYS: &[&str] = &[
"run",
"expect",
"expect-fail",
"expect_fail",
"requires",
"timeout",
"case",
];
pub struct VerifyCase {
pub name: String,
pub run_args: Vec<String>,
pub again: Vec<Vec<String>>,
pub session_cwd: bool,
pub expects: Vec<Regex>,
pub expect_fails: Vec<Regex>,
pub timeout: u64,
}
impl VerifyCase {
fn new(name: impl Into<String>) -> Self {
VerifyCase {
name: name.into(),
run_args: Vec::new(),
again: Vec::new(),
session_cwd: false,
expects: Vec::new(),
expect_fails: Vec::new(),
timeout: DEFAULT_TIMEOUT_SECS,
}
}
}
pub struct VerifyPlan {
pub requires: Option<String>,
pub cases: Vec<VerifyCase>,
}
impl VerifyPlan {
pub fn parse(src: &str) -> Result<VerifyPlan, String> {
let mut plan = parse_directives(src)?;
merge_verify_block(src, &mut plan)?;
Ok(plan)
}
pub fn is_empty(&self) -> bool {
self.requires.is_none() && self.cases.is_empty()
}
}
fn compile(v: &str) -> Result<Regex, String> {
Regex::new(v).map_err(|e| format!("bad regex {v:?}: {e}"))
}
fn parse_directives(src: &str) -> Result<VerifyPlan, String> {
let mut requires: Option<String> = None;
let mut cases: Vec<VerifyCase> = Vec::new();
let mut cur: Option<VerifyCase> = None;
for raw in src.lines() {
let line = raw.trim_start();
let Some(rest) = line.strip_prefix("#@") else {
continue;
};
let rest = rest.trim();
let (kw, val) = match rest.split_once(char::is_whitespace) {
Some((k, v)) => (k.trim_end_matches(':'), v.trim()),
None => (rest.trim_end_matches(':'), ""),
};
macro_rules! case {
() => {
cur.get_or_insert_with(|| VerifyCase::new("default"))
};
}
match kw {
"requires" => requires = Some(val.to_string()),
"case" => {
if let Some(c) = cur.take() {
cases.push(c);
}
cur = Some(VerifyCase::new(val));
}
"run" => case!().run_args = val.split_whitespace().map(String::from).collect(),
"again" => case!()
.again
.push(val.split_whitespace().map(String::from).collect()),
"session" => match val {
"cwd" => case!().session_cwd = true,
other => return Err(format!("unknown `#@ session {other}` (only `cwd`)")),
},
"expect" => case!().expects.push(compile(val)?),
"expect-fail" | "expect_fail" => case!().expect_fails.push(compile(val)?),
"timeout" => {
case!().timeout = val.parse().map_err(|_| format!("bad timeout {val:?}"))?
}
other => return Err(format!("unknown `#@ {other}` directive")),
}
}
if let Some(c) = cur.take() {
cases.push(c);
}
Ok(VerifyPlan { requires, cases })
}
fn merge_verify_block(src: &str, plan: &mut VerifyPlan) -> Result<(), String> {
let Ok(doc) = serde_yaml::from_str::<serde_yaml::Value>(src) else {
return Ok(());
};
let Some(block) = doc.get("verify") else {
return Ok(());
};
match block {
serde_yaml::Value::Sequence(items) => {
for (i, item) in items.iter().enumerate() {
plan.cases.push(case_from_map(item, None, i)?);
}
}
serde_yaml::Value::Mapping(m) => {
let all_directives = m
.keys()
.all(|k| k.as_str().is_some_and(|s| DIRECTIVE_KEYS.contains(&s)));
if all_directives {
if let Some(req) = m.get("requires").and_then(|v| v.as_str()) {
plan.requires = Some(req.to_string());
} else {
plan.cases.push(case_from_map(block, None, 0)?);
}
} else {
for (k, v) in m {
let name = k.as_str().ok_or("verify: case name must be a string")?;
plan.cases.push(case_from_map(v, Some(name), 0)?);
}
}
}
_ => return Err("verify: must be a map or a list of cases".to_string()),
}
Ok(())
}
fn case_from_map(
v: &serde_yaml::Value,
name_override: Option<&str>,
idx: usize,
) -> Result<VerifyCase, String> {
let m = v.as_mapping().ok_or("verify: each case must be a map")?;
let get = |k: &str| m.get(serde_yaml::Value::from(k));
let name = name_override
.map(String::from)
.or_else(|| get("case").and_then(|x| x.as_str()).map(String::from))
.unwrap_or_else(|| format!("case-{idx}"));
let mut case = VerifyCase::new(name);
if let Some(run) = get("run").and_then(|x| x.as_str()) {
case.run_args = run.split_whitespace().map(String::from).collect();
}
for s in strings_of(get("expect")) {
case.expects.push(compile(&s)?);
}
for s in strings_of(get("expect-fail").or_else(|| get("expect_fail"))) {
case.expect_fails.push(compile(&s)?);
}
if let Some(t) = get("timeout").and_then(|x| x.as_u64()) {
case.timeout = t;
}
Ok(case)
}
fn strings_of(v: Option<&serde_yaml::Value>) -> Vec<String> {
match v {
Some(serde_yaml::Value::String(s)) => vec![s.clone()],
Some(serde_yaml::Value::Sequence(items)) => items
.iter()
.filter_map(|x| x.as_str().map(String::from))
.collect(),
_ => Vec::new(),
}
}
pub enum Outcome {
Pass,
Skip(String),
Fail(String),
}
pub fn run_case(
binary: &Path,
workload_ref: &str,
sandbox: &Path,
label: &str,
case: &VerifyCase,
) -> Result<(), String> {
let safe_label = label.replace(['/', ' ', ':'], "_");
let session = sandbox.join(format!("session-{safe_label}"));
let case_cwd;
let workdir: &Path = if case.session_cwd {
case_cwd = sandbox.join(format!("cwd-{safe_label}"));
let _ = std::fs::remove_dir_all(&case_cwd);
std::fs::create_dir_all(&case_cwd).map_err(|e| format!("create case cwd: {e}"))?;
&case_cwd
} else {
let _ = std::fs::remove_dir_all(&session);
sandbox
};
let invoke = |args: &[String]| -> Result<(String, bool, bool), String> {
use std::io::Read as _;
use std::process::Stdio;
let mut cmd = Command::new(binary);
cmd.arg("run")
.arg(format!("workload={workload_ref}"))
.args(args);
if !case.session_cwd {
cmd.arg("--session-path").arg(&session);
}
let mut child = cmd
.current_dir(workdir)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.map_err(|e| format!("spawn failed: {e}"))?;
let mut out_pipe = child.stdout.take().expect("stdout piped");
let mut err_pipe = child.stderr.take().expect("stderr piped");
let out_h = std::thread::spawn(move || {
let mut buf = Vec::new();
let _ = out_pipe.read_to_end(&mut buf);
buf
});
let err_h = std::thread::spawn(move || {
let mut buf = Vec::new();
let _ = err_pipe.read_to_end(&mut buf);
buf
});
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(case.timeout);
let (status, timed_out) = loop {
if let Some(st) = child.try_wait().map_err(|e| format!("wait failed: {e}"))? {
break (st, false);
}
if std::time::Instant::now() >= deadline {
let _ = child.kill();
let st = child.wait().map_err(|e| format!("wait failed: {e}"))?;
break (st, true);
}
std::thread::sleep(std::time::Duration::from_millis(25));
};
let stdout = out_h.join().unwrap_or_default();
let stderr = err_h.join().unwrap_or_default();
let combined = format!(
"{}{}",
String::from_utf8_lossy(&stdout),
String::from_utf8_lossy(&stderr)
);
Ok((combined, status.success(), timed_out))
};
let (mut combined, mut succeeded, mut timed_out) = invoke(&case.run_args)?;
for (i, extra) in case.again.iter().enumerate() {
if !succeeded || timed_out {
return Err(format!(
"invocation {} of {} failed before the `again` steps completed:
{combined}",
i + 1,
case.again.len() + 1
));
}
let (c, s, to) = invoke(extra)?;
combined.push_str(&c);
succeeded = s;
timed_out = to;
}
check_case_output(case, &combined, succeeded, timed_out)
}
pub fn check_case_output(
case: &VerifyCase,
combined: &str,
succeeded: bool,
timed_out: bool,
) -> Result<(), String> {
if timed_out {
return Err(format!("timed out after {}s", case.timeout));
}
if case.expect_fails.is_empty() {
if !succeeded {
let err = combined
.lines()
.find(|l| l.contains("error:") || l.contains("panic"))
.unwrap_or("(no error line)");
return Err(format!("run failed (expected success): {err}"));
}
} else {
if succeeded {
return Err("expected a failure (`expect-fail`) but the run succeeded".to_string());
}
for re in &case.expect_fails {
if !re.is_match(combined) {
return Err(format!(
"expect-fail /{re}/ did not match the failure output"
));
}
}
}
for re in &case.expects {
if !re.is_match(combined) {
return Err(format!("expect /{re}/ did not match the output"));
}
}
Ok(())
}
pub fn requires_verification_rules(run_ref: &str) -> bool {
std::path::Path::new(run_ref)
.components()
.any(|c| c.as_os_str() == "examples")
}
pub fn verify_source(
binary: &Path,
label_root: &str,
run_ref: &str,
rule_text: &str,
sandbox: &Path,
) -> Vec<(String, Outcome)> {
let plan = match VerifyPlan::parse(rule_text) {
Ok(p) => p,
Err(e) => return vec![(label_root.to_string(), Outcome::Fail(e))],
};
if let Some(reason) = plan.requires {
return vec![(label_root.to_string(), Outcome::Skip(reason))];
}
if plan.cases.is_empty() {
return vec![(
label_root.to_string(),
if requires_verification_rules(run_ref) {
Outcome::Fail(
"no verification rules — every workload under `examples/` must \
declare them (add `#@ expect …` comments or a `verify:` block)"
.into(),
)
} else {
Outcome::Skip("no verification rules — nothing to check".into())
},
)];
}
plan.cases
.iter()
.map(|c| {
let label = format!("{label_root}::{}", c.name);
let outcome = match run_case(binary, run_ref, sandbox, &label, c) {
Ok(()) => Outcome::Pass,
Err(e) => Outcome::Fail(format!("{label}: {e}")),
};
(label, outcome)
})
.collect()
}
pub fn verify_file(
binary: &Path,
label_root: &str,
workload: &Path,
sandbox: &Path,
) -> Vec<(String, Outcome)> {
let src = match std::fs::read_to_string(workload) {
Ok(s) => s,
Err(e) => {
return vec![(
label_root.to_string(),
Outcome::Fail(format!("read error: {e}")),
)];
}
};
let abs = workload
.canonicalize()
.unwrap_or_else(|_| workload.to_path_buf());
verify_source(binary, label_root, &abs.to_string_lossy(), &src, sandbox)
}
pub enum WorkloadSource {
File(PathBuf),
Catalog { name: String, source: String },
}
pub fn resolve_ref(reference: &str) -> Option<WorkloadSource> {
let p = Path::new(reference);
if p.is_file() {
return Some(WorkloadSource::File(
p.canonicalize().unwrap_or_else(|_| p.to_path_buf()),
));
}
crate::catalog::lookup(reference).map(|w| WorkloadSource::Catalog {
name: w.name.to_string(),
source: w.source.to_string(),
})
}
pub fn declared_params(reference: &str) -> Option<std::collections::HashMap<String, String>> {
let merged = match resolve_ref(reference)? {
WorkloadSource::File(path) => crate::extends::load_and_merge(&path).ok()?.0,
WorkloadSource::Catalog { name, .. } => {
crate::extends::load_and_merge_bundled(crate::catalog::lookup(&name)?)
.ok()?
.0
}
};
let doc: serde_yaml::Value = serde_yaml::from_str(&merged).ok()?;
let params = doc.get("params")?.as_mapping()?;
let mut out = std::collections::HashMap::new();
for (k, v) in params {
let Some(key) = k.as_str() else { continue };
let val = match v {
serde_yaml::Value::String(s) => s.clone(),
serde_yaml::Value::Number(n) => n.to_string(),
serde_yaml::Value::Bool(b) => b.to_string(),
_ => continue,
};
out.insert(key.to_string(), val);
}
Some(out)
}
#[derive(Default)]
pub struct VerifySummary {
pub passed: usize,
pub skipped: Vec<String>,
pub failures: Vec<String>,
pub timings: Vec<WorkloadTiming>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CheckStatus {
Pass,
Skip,
Fail,
}
#[derive(Debug, Clone)]
pub struct WorkloadTiming {
pub label: String,
pub elapsed: Duration,
pub status: CheckStatus,
}
#[derive(Debug, Clone)]
pub enum CheckProgress {
Begin { total: usize },
Started { label: String },
Finished {
label: String,
elapsed: Duration,
status: CheckStatus,
},
}
pub type ProgressFn<'a> = dyn Fn(CheckProgress) + Sync + 'a;
pub fn no_progress(_: CheckProgress) {}
fn aggregate_status(outcomes: &[(String, Outcome)]) -> CheckStatus {
let mut saw_pass = false;
let mut saw_skip = false;
for (_, o) in outcomes {
match o {
Outcome::Fail(_) => return CheckStatus::Fail,
Outcome::Pass => saw_pass = true,
Outcome::Skip(_) => saw_skip = true,
}
}
if saw_skip && !saw_pass {
CheckStatus::Skip
} else {
CheckStatus::Pass
}
}
pub fn verify_path(
binary: &Path,
path: &Path,
sandbox: &Path,
progress: &ProgressFn,
) -> VerifySummary {
let _ = std::fs::create_dir_all(sandbox);
let mut files: Vec<PathBuf> = Vec::new();
if path.is_dir() {
collect_yaml(path, &mut files);
files.sort();
} else {
files.push(path.to_path_buf());
}
progress(CheckProgress::Begin { total: files.len() });
let acc: std::sync::Mutex<VerifySummary> = std::sync::Mutex::new(VerifySummary::default());
let next = std::sync::atomic::AtomicUsize::new(0);
let workers = files.len().min(
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4)
.saturating_mul(2)
.clamp(1, 16),
);
std::thread::scope(|s| {
for _ in 0..workers {
s.spawn(|| {
loop {
let i = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(f) = files.get(i) else { break };
let label = f
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("?")
.to_string();
progress(CheckProgress::Started {
label: label.clone(),
});
let start = Instant::now();
let outcomes = verify_file(binary, &label, f, sandbox);
let elapsed = start.elapsed();
let status = aggregate_status(&outcomes);
let mut g = acc.lock().unwrap();
for (lbl, outcome) in outcomes {
match outcome {
Outcome::Pass => g.passed += 1,
Outcome::Skip(r) => g.skipped.push(format!("{lbl}: {r}")),
Outcome::Fail(m) => g.failures.push(m),
}
}
g.timings.push(WorkloadTiming {
label: label.clone(),
elapsed,
status,
});
drop(g);
progress(CheckProgress::Finished {
label,
elapsed,
status,
});
}
});
}
});
let mut sum = acc.into_inner().unwrap();
sum.skipped.sort();
sum.failures.sort();
sum
}
pub fn verify_target(
binary: &Path,
target: &str,
sandbox: &Path,
progress: &ProgressFn,
) -> VerifySummary {
let p = Path::new(target);
if p.is_dir() {
return verify_path(binary, p, sandbox, progress);
}
let _ = std::fs::create_dir_all(sandbox);
progress(CheckProgress::Begin { total: 1 });
let label = match &resolve_ref(target) {
Some(WorkloadSource::File(path)) => path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or(target)
.to_string(),
_ => target.to_string(),
};
progress(CheckProgress::Started {
label: label.clone(),
});
let start = Instant::now();
let cases: Vec<(String, Outcome)> = match resolve_ref(target) {
Some(WorkloadSource::File(path)) => verify_file(binary, &label, &path, sandbox),
Some(WorkloadSource::Catalog { name, source }) => {
verify_source(binary, &name, &name, &source, sandbox)
}
None => vec![(
target.to_string(),
Outcome::Fail(format!(
"no such workload '{target}': not a local file, not a directory, and \
no bundled workload by that name (try `nmbrs describe workloads --all`).{}",
crate::suggest::did_you_mean(&crate::suggest::suggest_workloads(target))
)),
)],
};
let elapsed = start.elapsed();
let status = aggregate_status(&cases);
let mut sum = VerifySummary::default();
for (lbl, outcome) in cases {
match outcome {
Outcome::Pass => sum.passed += 1,
Outcome::Skip(r) => sum.skipped.push(format!("{lbl}: {r}")),
Outcome::Fail(m) => sum.failures.push(m),
}
}
sum.timings.push(WorkloadTiming {
label: label.clone(),
elapsed,
status,
});
progress(CheckProgress::Finished {
label,
elapsed,
status,
});
sum
}
pub fn collect_workload_files(dir: &Path) -> Vec<PathBuf> {
let mut files = Vec::new();
collect_yaml(dir, &mut files);
files.sort();
files
}
fn collect_yaml(dir: &Path, out: &mut Vec<PathBuf>) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for e in entries.flatten() {
let p = e.path();
if p.is_dir() {
collect_yaml(&p, out);
} else if p.extension().is_some_and(|x| x == "yaml" || x == "yml") {
out.push(p);
}
}
}
#[cfg(test)]
mod rules_required_tests {
use super::requires_verification_rules;
#[test]
fn examples_require_rules() {
assert!(requires_verification_rules(
"/repo/examples/optimizer/control.yaml"
));
assert!(requires_verification_rules("examples/w.yaml"));
assert!(requires_verification_rules(
"/repo/nmbrs/examples/modules/module_test.yaml"
));
}
#[test]
fn other_locations_do_not_require_rules() {
assert!(!requires_verification_rules(
"/repo/nmbrs/workloads/cql/incremental_sweep.yaml"
));
assert!(!requires_verification_rules("/tmp/scratch.yaml"));
assert!(!requires_verification_rules("some_catalog_name"));
}
#[test]
fn matches_path_components_not_substrings() {
assert!(!requires_verification_rules(
"/home/me/examples-scratch/w.yaml"
));
assert!(!requires_verification_rules("/home/me/myexamples/w.yaml"));
assert!(requires_verification_rules(
"/home/me/examples/scratch/w.yaml"
));
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn comment_and_yaml_forms_are_equivalent() {
let comment = "ops: { a: { raw: x } }\n#@ run cycles=3\n#@ expect 0 failed\n";
let single = "ops: { a: { raw: x } }\nverify: { run: cycles=3, expect: \"0 failed\" }\n";
let listed =
"ops: { a: { raw: x } }\nverify:\n - { run: cycles=3, expect: \"0 failed\" }\n";
let named =
"ops: { a: { raw: x } }\nverify:\n smoke: { run: cycles=3, expect: \"0 failed\" }\n";
for src in [comment, single, listed, named] {
let p = VerifyPlan::parse(src).expect("parse");
assert_eq!(p.cases.len(), 1, "one case for: {src}");
assert_eq!(p.cases[0].run_args, vec!["cycles=3"], "run for: {src}");
assert_eq!(p.cases[0].expects.len(), 1, "expect for: {src}");
}
}
#[test]
fn name_keyed_map_yields_named_cases() {
let src = "verify:\n alpha: { run: scenario=a, expect: \"x\" }\n beta: { expect: [\"y\", \"z\"] }\n";
let p = VerifyPlan::parse(src).unwrap();
let names: Vec<&str> = p.cases.iter().map(|c| c.name.as_str()).collect();
assert!(
names.contains(&"alpha") && names.contains(&"beta"),
"names: {names:?}"
);
let beta = p.cases.iter().find(|c| c.name == "beta").unwrap();
assert_eq!(beta.expects.len(), 2);
}
#[test]
fn requires_block_skips_the_file() {
let p = VerifyPlan::parse("verify: { requires: needs a backend }\n").unwrap();
assert_eq!(p.requires.as_deref(), Some("needs a backend"));
assert!(p.cases.is_empty());
}
#[test]
fn comment_and_block_cases_combine() {
let src = "#@ case fromcomment\n#@ expect a\nverify:\n fromblock: { expect: b }\n";
let p = VerifyPlan::parse(src).unwrap();
let names: Vec<&str> = p.cases.iter().map(|c| c.name.as_str()).collect();
assert!(
names.contains(&"fromcomment") && names.contains(&"fromblock"),
"{names:?}"
);
}
#[test]
fn resolve_ref_finds_files_and_rejects_unknown_names() {
let dir = std::env::temp_dir().join(format!("nmbrs-verify-resolve-{}", std::process::id()));
let _ = std::fs::create_dir_all(&dir);
let file = dir.join("w.yaml");
std::fs::write(&file, "ops: { a: { raw: x } }\n").unwrap();
match resolve_ref(file.to_str().unwrap()) {
Some(WorkloadSource::File(p)) => assert!(p.is_absolute(), "absolute: {p:?}"),
other => panic!(
"expected File, got {}",
matches!(other, Some(WorkloadSource::Catalog { .. })) as i32
),
}
assert!(resolve_ref("definitely/not/a/workload").is_none());
let _ = std::fs::remove_dir_all(&dir);
}
}