use crate::dag::{DagError, NodeId, WorkflowDag};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::time::Duration;
#[derive(Debug, Clone)]
pub struct SimulationConfig {
pub simulated_outputs: HashMap<String, HashMap<String, serde_json::Value>>,
pub failing_task_types: HashSet<String>,
pub estimated_durations: HashMap<String, Duration>,
pub continue_on_failure: bool,
pub max_nodes: usize,
pub condition_overrides: HashMap<String, bool>,
}
impl Default for SimulationConfig {
fn default() -> Self {
Self {
simulated_outputs: HashMap::new(),
failing_task_types: HashSet::new(),
estimated_durations: HashMap::new(),
continue_on_failure: true,
max_nodes: 10_000,
condition_overrides: HashMap::new(),
}
}
}
impl SimulationConfig {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn with_outputs(
mut self,
task_type: impl Into<String>,
outputs: HashMap<String, serde_json::Value>,
) -> Self {
self.simulated_outputs.insert(task_type.into(), outputs);
self
}
#[must_use]
pub fn with_failing_type(mut self, task_type: impl Into<String>) -> Self {
self.failing_task_types.insert(task_type.into());
self
}
#[must_use]
pub fn with_duration(mut self, task_type: impl Into<String>, duration: Duration) -> Self {
self.estimated_durations.insert(task_type.into(), duration);
self
}
#[must_use]
pub fn with_continue_on_failure(mut self, cont: bool) -> Self {
self.continue_on_failure = cont;
self
}
#[must_use]
pub fn with_condition_override(mut self, condition: impl Into<String>, value: bool) -> Self {
self.condition_overrides.insert(condition.into(), value);
self
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SimulationTraceEntry {
pub node_id: NodeId,
pub task_type: String,
pub outcome: SimulatedOutcome,
pub estimated_duration: Option<Duration>,
pub dependencies: Vec<NodeId>,
pub evaluated_conditions: Vec<EvaluatedCondition>,
pub visit_order: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum SimulatedOutcome {
WouldSucceed,
WouldFail {
reason: String,
},
WouldSkip {
reason: String,
},
Unreachable,
Blocked {
failed_dependency: NodeId,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EvaluatedCondition {
pub condition: String,
pub result: bool,
pub source: ConditionSource,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ConditionSource {
Override,
DefaultTrue,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SimulationResult {
pub trace: Vec<SimulationTraceEntry>,
pub summary: SimulationSummary,
pub node_statuses: HashMap<NodeId, SimulatedOutcome>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SimulationSummary {
pub total_nodes: usize,
pub would_succeed: usize,
pub would_fail: usize,
pub would_skip: usize,
pub unreachable: usize,
pub blocked: usize,
pub estimated_total_duration: Duration,
pub estimated_critical_path_duration: Duration,
pub would_complete: bool,
}
#[derive(Debug)]
pub struct WorkflowSimulator {
config: SimulationConfig,
}
impl WorkflowSimulator {
#[must_use]
pub fn new(config: SimulationConfig) -> Self {
Self { config }
}
#[must_use]
pub fn default_simulator() -> Self {
Self::new(SimulationConfig::default())
}
pub fn simulate(&self, dag: &WorkflowDag) -> Result<SimulationResult, DagError> {
let order = dag.topological_sort()?;
let mut trace = Vec::new();
let mut node_statuses: HashMap<NodeId, SimulatedOutcome> = HashMap::new();
let mut failed_nodes: HashSet<NodeId> = HashSet::new();
let mut node_outputs: HashMap<NodeId, HashMap<String, serde_json::Value>> = HashMap::new();
let mut visit_order = 0usize;
for &node_id in &order {
let node = match dag.nodes.get(&node_id) {
Some(n) => n,
None => continue,
};
let deps = dag.predecessors(node_id);
let blocked_by = if !self.config.continue_on_failure {
deps.iter().find(|d| failed_nodes.contains(d)).copied()
} else {
None
};
let evaluated_conditions = self.evaluate_incoming_conditions(dag, node_id);
let conditions_met = evaluated_conditions.iter().all(|ec| ec.result);
let outcome = if let Some(failed_dep) = blocked_by {
SimulatedOutcome::Blocked {
failed_dependency: failed_dep,
}
} else if !conditions_met {
let failed_conds: Vec<&str> = evaluated_conditions
.iter()
.filter(|ec| !ec.result)
.map(|ec| ec.condition.as_str())
.collect();
SimulatedOutcome::WouldSkip {
reason: format!("conditions not met: {}", failed_conds.join(", ")),
}
} else if self.config.failing_task_types.contains(&node.task_type) {
failed_nodes.insert(node_id);
SimulatedOutcome::WouldFail {
reason: format!("task type '{}' configured to fail", node.task_type),
}
} else {
if let Some(outputs) = self.config.simulated_outputs.get(&node.task_type) {
node_outputs.insert(node_id, outputs.clone());
}
SimulatedOutcome::WouldSucceed
};
let estimated_duration = self
.config
.estimated_durations
.get(&node.task_type)
.copied();
let entry = SimulationTraceEntry {
node_id,
task_type: node.task_type.clone(),
outcome: outcome.clone(),
estimated_duration,
dependencies: deps,
evaluated_conditions,
visit_order,
};
node_statuses.insert(node_id, outcome);
trace.push(entry);
visit_order += 1;
}
let visited: HashSet<NodeId> = node_statuses.keys().copied().collect();
for &node_id in dag.nodes.keys() {
if !visited.contains(&node_id) {
node_statuses.insert(node_id, SimulatedOutcome::Unreachable);
}
}
let summary = self.compute_summary(dag, &node_statuses, &trace);
Ok(SimulationResult {
trace,
summary,
node_statuses,
})
}
fn evaluate_incoming_conditions(
&self,
dag: &WorkflowDag,
node_id: NodeId,
) -> Vec<EvaluatedCondition> {
let mut results = Vec::new();
for edge in &dag.edges {
if edge.to_node != node_id {
continue;
}
if let Some(ref condition) = edge.condition {
let (result, source) =
if let Some(&override_val) = self.config.condition_overrides.get(condition) {
(override_val, ConditionSource::Override)
} else {
(true, ConditionSource::DefaultTrue)
};
results.push(EvaluatedCondition {
condition: condition.clone(),
result,
source,
});
}
}
results
}
fn compute_summary(
&self,
dag: &WorkflowDag,
statuses: &HashMap<NodeId, SimulatedOutcome>,
trace: &[SimulationTraceEntry],
) -> SimulationSummary {
let total_nodes = dag.nodes.len();
let mut would_succeed = 0usize;
let mut would_fail = 0usize;
let mut would_skip = 0usize;
let mut unreachable = 0usize;
let mut blocked = 0usize;
let mut total_duration = Duration::ZERO;
for outcome in statuses.values() {
match outcome {
SimulatedOutcome::WouldSucceed => would_succeed += 1,
SimulatedOutcome::WouldFail { .. } => would_fail += 1,
SimulatedOutcome::WouldSkip { .. } => would_skip += 1,
SimulatedOutcome::Unreachable => unreachable += 1,
SimulatedOutcome::Blocked { .. } => blocked += 1,
}
}
for entry in trace {
if entry.outcome == SimulatedOutcome::WouldSucceed {
if let Some(d) = entry.estimated_duration {
total_duration += d;
}
}
}
let critical_path = self.compute_critical_path_duration(dag, trace);
SimulationSummary {
total_nodes,
would_succeed,
would_fail,
would_skip,
unreachable,
blocked,
estimated_total_duration: total_duration,
estimated_critical_path_duration: critical_path,
would_complete: would_fail == 0 && blocked == 0,
}
}
fn compute_critical_path_duration(
&self,
dag: &WorkflowDag,
trace: &[SimulationTraceEntry],
) -> Duration {
let duration_map: HashMap<NodeId, Duration> = trace
.iter()
.filter(|e| e.outcome == SimulatedOutcome::WouldSucceed)
.filter_map(|e| e.estimated_duration.map(|d| (e.node_id, d)))
.collect();
if duration_map.is_empty() {
return Duration::ZERO;
}
let order = match dag.topological_sort() {
Ok(o) => o,
Err(_) => return Duration::ZERO,
};
let mut dist: HashMap<NodeId, Duration> = HashMap::new();
for &node_id in &order {
let self_dur = duration_map
.get(&node_id)
.copied()
.unwrap_or(Duration::ZERO);
let max_pred = dag
.predecessors(node_id)
.iter()
.filter_map(|p| dist.get(p))
.max()
.copied()
.unwrap_or(Duration::ZERO);
dist.insert(node_id, max_pred + self_dur);
}
dist.values().max().copied().unwrap_or(Duration::ZERO)
}
}
pub fn quick_simulate(dag: &WorkflowDag) -> Result<bool, DagError> {
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(dag)?;
Ok(result.summary.would_complete)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dag::{WorkflowEdge, WorkflowNode};
fn make_node(task_type: &str) -> WorkflowNode {
WorkflowNode::new(task_type)
}
fn make_linear_dag() -> (WorkflowDag, NodeId, NodeId, NodeId) {
let mut dag = WorkflowDag::new();
let a = dag.add_node(make_node("ingest")).expect("add node");
let b = dag.add_node(make_node("transcode")).expect("add node");
let c = dag.add_node(make_node("deliver")).expect("add node");
dag.add_edge(WorkflowEdge::new(a, b, "raw"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(b, c, "encoded"))
.expect("add edge");
(dag, a, b, c)
}
#[test]
fn test_simulate_all_succeed() {
let (dag, _, _, _) = make_linear_dag();
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
assert!(result.summary.would_complete);
assert_eq!(result.summary.would_succeed, 3);
assert_eq!(result.summary.would_fail, 0);
assert_eq!(result.trace.len(), 3);
}
#[test]
fn test_simulate_with_failing_task() {
let (dag, _, _, _) = make_linear_dag();
let config = SimulationConfig::new().with_failing_type("transcode");
let sim = WorkflowSimulator::new(config);
let result = sim.simulate(&dag).expect("simulate");
assert!(!result.summary.would_complete);
assert_eq!(result.summary.would_fail, 1);
assert_eq!(result.summary.would_succeed, 2); }
#[test]
fn test_simulate_blocked_on_failure() {
let (dag, _, _, _) = make_linear_dag();
let config = SimulationConfig::new()
.with_failing_type("transcode")
.with_continue_on_failure(false);
let sim = WorkflowSimulator::new(config);
let result = sim.simulate(&dag).expect("simulate");
assert!(!result.summary.would_complete);
assert_eq!(result.summary.blocked, 1); assert_eq!(result.summary.would_fail, 1);
assert_eq!(result.summary.would_succeed, 1); }
#[test]
fn test_simulate_with_conditions() {
let mut dag = WorkflowDag::new();
let a = dag.add_node(make_node("check")).expect("add node");
let b = dag.add_node(make_node("high_res")).expect("add node");
let c = dag.add_node(make_node("low_res")).expect("add node");
dag.add_edge(WorkflowEdge::with_condition(
a,
b,
"video",
"resolution == 4k",
))
.expect("add edge");
dag.add_edge(WorkflowEdge::with_condition(
a,
c,
"video",
"resolution != 4k",
))
.expect("add edge");
let config = SimulationConfig::new()
.with_condition_override("resolution == 4k", true)
.with_condition_override("resolution != 4k", false);
let sim = WorkflowSimulator::new(config);
let result = sim.simulate(&dag).expect("simulate");
assert_eq!(result.summary.would_succeed, 2); assert_eq!(result.summary.would_skip, 1); }
#[test]
fn test_conditions_default_true() {
let mut dag = WorkflowDag::new();
let a = dag.add_node(make_node("source")).expect("add node");
let b = dag.add_node(make_node("sink")).expect("add node");
dag.add_edge(WorkflowEdge::with_condition(a, b, "data", "some_condition"))
.expect("add edge");
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
assert_eq!(result.summary.would_succeed, 2);
assert!(result.trace[1]
.evaluated_conditions
.iter()
.all(|ec| ec.source == ConditionSource::DefaultTrue));
}
#[test]
fn test_estimated_durations() {
let (dag, _, _, _) = make_linear_dag();
let config = SimulationConfig::new()
.with_duration("ingest", Duration::from_secs(10))
.with_duration("transcode", Duration::from_secs(60))
.with_duration("deliver", Duration::from_secs(5));
let sim = WorkflowSimulator::new(config);
let result = sim.simulate(&dag).expect("simulate");
assert_eq!(
result.summary.estimated_total_duration,
Duration::from_secs(75)
);
assert_eq!(
result.summary.estimated_critical_path_duration,
Duration::from_secs(75)
);
}
#[test]
fn test_critical_path_parallel() {
let mut dag = WorkflowDag::new();
let root = dag.add_node(make_node("start")).expect("add node");
let branch_a = dag.add_node(make_node("fast")).expect("add node");
let branch_b = dag.add_node(make_node("slow")).expect("add node");
let join = dag.add_node(make_node("finish")).expect("add node");
dag.add_edge(WorkflowEdge::new(root, branch_a, "x"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(root, branch_b, "x"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(branch_a, join, "x"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(branch_b, join, "x"))
.expect("add edge");
let config = SimulationConfig::new()
.with_duration("start", Duration::from_secs(5))
.with_duration("fast", Duration::from_secs(10))
.with_duration("slow", Duration::from_secs(30))
.with_duration("finish", Duration::from_secs(5));
let sim = WorkflowSimulator::new(config);
let result = sim.simulate(&dag).expect("simulate");
assert_eq!(
result.summary.estimated_total_duration,
Duration::from_secs(50)
);
assert_eq!(
result.summary.estimated_critical_path_duration,
Duration::from_secs(40)
);
}
#[test]
fn test_trace_visit_order() {
let (dag, _, _, _) = make_linear_dag();
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
for (i, entry) in result.trace.iter().enumerate() {
assert_eq!(entry.visit_order, i);
}
}
#[test]
fn test_trace_dependencies_recorded() {
let (dag, a, _, _) = make_linear_dag();
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
let transcode_entry = result
.trace
.iter()
.find(|e| e.task_type == "transcode")
.expect("find transcode");
assert_eq!(transcode_entry.dependencies.len(), 1);
assert_eq!(transcode_entry.dependencies[0], a);
}
#[test]
fn test_simulate_empty_dag() {
let dag = WorkflowDag::new();
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
assert!(result.summary.would_complete);
assert_eq!(result.summary.total_nodes, 0);
assert!(result.trace.is_empty());
}
#[test]
fn test_simulate_single_node() {
let mut dag = WorkflowDag::new();
dag.add_node(make_node("solo")).expect("add node");
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
assert!(result.summary.would_complete);
assert_eq!(result.summary.would_succeed, 1);
}
#[test]
fn test_quick_simulate_success() {
let (dag, _, _, _) = make_linear_dag();
assert!(quick_simulate(&dag).expect("simulate"));
}
#[test]
fn test_simulate_with_outputs() {
let (dag, _, _, _) = make_linear_dag();
let mut outputs = HashMap::new();
outputs.insert(
"path".to_string(),
serde_json::json!(std::env::temp_dir()
.join("oximedia-workflow-sim-out.mp4")
.to_string_lossy()),
);
let config = SimulationConfig::new().with_outputs("ingest", outputs);
let sim = WorkflowSimulator::new(config);
let result = sim.simulate(&dag).expect("simulate");
assert!(result.summary.would_complete);
}
#[test]
fn test_simulate_diamond_dag() {
let mut dag = WorkflowDag::new();
let root = dag.add_node(make_node("root")).expect("add node");
let node_a = dag.add_node(make_node("branch_a")).expect("add node");
let node_b = dag.add_node(make_node("branch_b")).expect("add node");
let join = dag.add_node(make_node("join")).expect("add node");
dag.add_edge(WorkflowEdge::new(root, node_a, "x"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(root, node_b, "x"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(node_a, join, "x"))
.expect("add edge");
dag.add_edge(WorkflowEdge::new(node_b, join, "x"))
.expect("add edge");
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
assert!(result.summary.would_complete);
assert_eq!(result.summary.would_succeed, 4);
}
#[test]
fn test_node_statuses_map() {
let (dag, a, b, c) = make_linear_dag();
let sim = WorkflowSimulator::default_simulator();
let result = sim.simulate(&dag).expect("simulate");
assert_eq!(result.node_statuses.len(), 3);
assert_eq!(
result.node_statuses.get(&a),
Some(&SimulatedOutcome::WouldSucceed)
);
assert_eq!(
result.node_statuses.get(&b),
Some(&SimulatedOutcome::WouldSucceed)
);
assert_eq!(
result.node_statuses.get(&c),
Some(&SimulatedOutcome::WouldSucceed)
);
}
}
#[derive(Debug, Clone)]
pub struct SimulationReport {
pub step_count: usize,
pub estimated_duration_ms: u64,
pub dependencies: Vec<String>,
pub warnings: Vec<String>,
}
impl SimulationReport {
#[must_use]
pub fn is_clean(&self) -> bool {
self.warnings.is_empty()
}
}
pub struct WorkflowDryRun;
impl WorkflowDryRun {
#[must_use]
pub fn simulate(workflow: &crate::workflow::Workflow) -> SimulationReport {
let mut warnings = Vec::new();
let mut dependencies = Vec::new();
let tasks: Vec<&crate::task::Task> = workflow.tasks().collect();
let step_count = tasks.len();
if step_count == 0 {
warnings.push("Workflow has no tasks".to_string());
return SimulationReport {
step_count: 0,
estimated_duration_ms: 0,
dependencies,
warnings,
};
}
let mut estimated_duration_ms: u64 = 0;
for task in &tasks {
let dur = Self::estimate_task_duration_ms(task);
estimated_duration_ms = estimated_duration_ms.saturating_add(dur);
for dep_id in &task.dependencies {
let dep_name = workflow
.get_task(dep_id)
.map(|t| t.name.as_str())
.unwrap_or("<unknown>");
dependencies.push(format!(
"'{}' [{}] depends on '{}' [{}]",
task.name, task.id, dep_name, dep_id
));
}
}
for edge in &workflow.edges {
let from_name = workflow
.get_task(&edge.from)
.map(|t| t.name.as_str())
.unwrap_or("<unknown>");
let to_name = workflow
.get_task(&edge.to)
.map(|t| t.name.as_str())
.unwrap_or("<unknown>");
let dep_str = if let Some(ref cond) = edge.condition {
format!(
"edge: '{}' [{}] -> '{}' [{}] (condition: {cond})",
from_name, edge.from, to_name, edge.to
)
} else {
format!(
"edge: '{}' [{}] -> '{}' [{}]",
from_name, edge.from, to_name, edge.to
)
};
if !dependencies.contains(&dep_str) {
dependencies.push(dep_str);
}
}
let cycle_warnings = Self::check_cycles(workflow);
warnings.extend(cycle_warnings);
for task in &tasks {
if task.timeout == std::time::Duration::from_secs(3600) {
}
for cond in &task.conditions {
if cond.trim().is_empty() {
warnings.push(format!(
"Task '{}' has an empty condition string",
task.name
));
}
}
}
SimulationReport {
step_count,
estimated_duration_ms,
dependencies,
warnings,
}
}
fn check_cycles(workflow: &crate::workflow::Workflow) -> Vec<String> {
let mut adj: std::collections::HashMap<crate::task::TaskId, Vec<crate::task::TaskId>> =
std::collections::HashMap::new();
for task in workflow.tasks() {
adj.insert(task.id, task.dependencies.clone());
}
for edge in &workflow.edges {
adj.entry(edge.to).or_default().push(edge.from);
}
let mut warnings = Vec::new();
let mut color: std::collections::HashMap<crate::task::TaskId, u8> =
std::collections::HashMap::new();
for &start in adj.keys() {
if color.get(&start).copied().unwrap_or(0) == 0 {
let mut path = Vec::new();
if Self::dfs_cycle(start, &adj, &mut color, &mut path) {
warnings.push(format!("Potential cycle detected near task [{}]", start));
}
}
}
warnings
}
fn dfs_cycle(
node: crate::task::TaskId,
adj: &std::collections::HashMap<crate::task::TaskId, Vec<crate::task::TaskId>>,
color: &mut std::collections::HashMap<crate::task::TaskId, u8>,
path: &mut Vec<crate::task::TaskId>,
) -> bool {
color.insert(node, 1); path.push(node);
if let Some(neighbors) = adj.get(&node) {
for &next in neighbors {
let c = color.get(&next).copied().unwrap_or(0);
if c == 1 {
return true; }
if c == 0 && Self::dfs_cycle(next, adj, color, path) {
return true;
}
}
}
path.pop();
color.insert(node, 2); false
}
#[must_use]
fn estimate_task_duration_ms(task: &crate::task::Task) -> u64 {
match &task.task_type {
crate::task::TaskType::Wait { duration } => {
duration.as_millis().try_into().unwrap_or(u64::MAX)
}
crate::task::TaskType::Transcode { .. } => 60_000,
crate::task::TaskType::QualityControl { .. } => 30_000,
crate::task::TaskType::Transfer { .. } => 20_000,
crate::task::TaskType::Analysis { .. } => 15_000,
crate::task::TaskType::Notification { .. } => 1_000,
crate::task::TaskType::HttpRequest { .. } => 1_000,
crate::task::TaskType::CustomScript { .. } => 5_000,
crate::task::TaskType::Conditional { .. } => 5_000,
}
}
}
#[cfg(test)]
mod dry_run_tests {
use super::*;
use crate::task::{Task, TaskType};
use crate::workflow::Workflow;
use std::path::PathBuf;
use std::time::Duration;
fn make_transcode_task(name: &str) -> Task {
Task::new(
name,
TaskType::Transcode {
input: PathBuf::from("/in/video.mp4"),
output: PathBuf::from("/out/video.mp4"),
preset: "h264".to_string(),
params: std::collections::HashMap::new(),
},
)
}
fn make_wait_task(name: &str, secs: u64) -> Task {
Task::new(
name,
TaskType::Wait {
duration: Duration::from_secs(secs),
},
)
}
fn make_notify_task(name: &str) -> Task {
Task::new(
name,
TaskType::Notification {
channel: crate::task::NotificationChannel::Webhook {
url: "https://example.com/hook".to_string(),
},
message: "done".to_string(),
metadata: std::collections::HashMap::new(),
},
)
}
#[test]
fn test_dry_run_empty_workflow() {
let wf = Workflow::new("empty");
let report = WorkflowDryRun::simulate(&wf);
assert_eq!(report.step_count, 0);
assert_eq!(report.estimated_duration_ms, 0);
assert!(!report.warnings.is_empty(), "should warn about no tasks");
assert!(
report.warnings.iter().any(|w| w.contains("no tasks")),
"warning: {:?}",
report.warnings
);
}
#[test]
fn test_dry_run_single_task_step_count() {
let mut wf = Workflow::new("single");
wf.add_task(make_transcode_task("encode"));
let report = WorkflowDryRun::simulate(&wf);
assert_eq!(report.step_count, 1);
}
#[test]
fn test_dry_run_single_task_no_deps() {
let mut wf = Workflow::new("single");
wf.add_task(make_transcode_task("encode"));
let report = WorkflowDryRun::simulate(&wf);
assert!(report.dependencies.is_empty());
}
#[test]
fn test_dry_run_multiple_tasks_step_count() {
let mut wf = Workflow::new("multi");
wf.add_task(make_transcode_task("encode"));
wf.add_task(make_notify_task("notify"));
wf.add_task(make_wait_task("pause", 5));
let report = WorkflowDryRun::simulate(&wf);
assert_eq!(report.step_count, 3);
}
#[test]
fn test_dry_run_wait_task_uses_actual_duration() {
let mut wf = Workflow::new("wait-wf");
wf.add_task(make_wait_task("sleep", 10)); let report = WorkflowDryRun::simulate(&wf);
assert_eq!(report.estimated_duration_ms, 10_000);
}
#[test]
fn test_dry_run_estimated_duration_sum() {
let mut wf = Workflow::new("sum-wf");
wf.add_task(make_transcode_task("encode")); wf.add_task(make_notify_task("notify")); let report = WorkflowDryRun::simulate(&wf);
assert_eq!(report.estimated_duration_ms, 61_000);
}
#[test]
fn test_dry_run_dependencies_listed_via_task_deps() {
let mut wf = Workflow::new("dep-wf");
let encode_id = wf.add_task(make_transcode_task("encode"));
let mut notify = make_notify_task("notify");
notify.add_dependency(encode_id);
wf.add_task(notify);
let report = WorkflowDryRun::simulate(&wf);
assert!(!report.dependencies.is_empty(), "should record dependency");
assert!(
report
.dependencies
.iter()
.any(|d| d.contains("notify") && d.contains("encode")),
"deps: {:?}",
report.dependencies
);
}
#[test]
fn test_dry_run_dependencies_listed_via_edges() {
let mut wf = Workflow::new("edge-wf");
let encode_id = wf.add_task(make_transcode_task("encode"));
let notify_id = wf.add_task(make_notify_task("notify"));
wf.add_edge(encode_id, notify_id).expect("add edge");
let report = WorkflowDryRun::simulate(&wf);
assert!(
report
.dependencies
.iter()
.any(|d| d.contains("encode") && d.contains("notify")),
"deps: {:?}",
report.dependencies
);
}
#[test]
fn test_dry_run_warnings_for_empty_workflow() {
let wf = Workflow::new("empty");
let report = WorkflowDryRun::simulate(&wf);
assert!(!report.warnings.is_empty());
}
#[test]
fn test_dry_run_no_warnings_for_simple_valid_workflow() {
let mut wf = Workflow::new("clean");
wf.add_task(make_transcode_task("encode"));
let report = WorkflowDryRun::simulate(&wf);
assert!(
report.warnings.is_empty(),
"unexpected warnings: {:?}",
report.warnings
);
}
#[test]
fn test_dry_run_simulation_report_all_fields_accessible() {
let mut wf = Workflow::new("fields-test");
wf.add_task(make_transcode_task("encode"));
let report = WorkflowDryRun::simulate(&wf);
let _ = report.step_count;
let _ = report.estimated_duration_ms;
let _ = report.dependencies;
let _ = report.warnings;
assert!(report.is_clean());
}
#[test]
fn test_dry_run_multiple_dependencies() {
let mut wf = Workflow::new("multi-dep");
let a = wf.add_task(make_transcode_task("encode-a"));
let b = wf.add_task(make_transcode_task("encode-b"));
let notify_id = wf.add_task(make_notify_task("notify"));
wf.add_edge(a, notify_id).expect("edge a->notify");
wf.add_edge(b, notify_id).expect("edge b->notify");
let report = WorkflowDryRun::simulate(&wf);
let edge_deps: Vec<_> = report
.dependencies
.iter()
.filter(|d| d.contains("notify"))
.collect();
assert!(
edge_deps.len() >= 2,
"expected 2 deps to notify, got: {:?}",
report.dependencies
);
}
}