1use std::time::Duration;
8
9use rust_decimal::Decimal;
10use serde_json::Value;
11use uuid::Uuid;
12
13use ironflow_store::models::{Run, RunStatus, Step, StepKind, StepStatus};
14
15use crate::executor::{StepOutput, StepResult};
16
17static NULL: Value = Value::Null;
19
20#[derive(Debug, Clone)]
35pub struct TestStep {
36 step: Step,
37 output: Value,
38}
39
40impl TestStep {
41 pub(crate) fn new(step: Step) -> Self {
43 let output = step.output.clone().unwrap_or(Value::Null);
44 Self { step, output }
45 }
46
47 pub fn name(&self) -> &str {
49 &self.step.name
50 }
51
52 pub fn kind(&self) -> &StepKind {
54 &self.step.kind
55 }
56
57 pub fn status(&self) -> StepStatus {
59 self.step.status.state
60 }
61
62 pub fn output(&self) -> &Value {
64 &self.output
65 }
66
67 pub fn input(&self) -> &Value {
69 self.step.input.as_ref().unwrap_or(&NULL)
70 }
71
72 pub fn error(&self) -> Option<&str> {
74 self.step.error.as_deref()
75 }
76
77 pub fn duration(&self) -> Duration {
79 Duration::from_millis(self.step.duration_ms)
80 }
81
82 pub fn cost_usd(&self) -> Decimal {
84 self.step.cost_usd
85 }
86
87 pub fn is_completed(&self) -> bool {
89 self.status() == StepStatus::Completed
90 }
91
92 pub fn is_error_handler(&self) -> bool {
95 self.step.is_error_handler
96 }
97
98 pub fn raw(&self) -> &Step {
100 &self.step
101 }
102
103 pub fn step_output(&self) -> StepOutput {
120 StepOutput::from(&self.step)
121 }
122}
123
124#[derive(Debug, Clone)]
143pub struct TestResult {
144 run: Run,
145 steps: Vec<TestStep>,
146 step_results: Vec<StepResult>,
147 error: Option<String>,
148}
149
150impl TestResult {
151 pub(crate) fn new(
153 run: Run,
154 steps: Vec<Step>,
155 step_results: Vec<StepResult>,
156 error: Option<String>,
157 ) -> Self {
158 Self {
159 run,
160 steps: steps.into_iter().map(TestStep::new).collect(),
161 step_results,
162 error,
163 }
164 }
165
166 pub fn status(&self) -> RunStatus {
168 self.run.status.state
169 }
170
171 pub fn run(&self) -> &Run {
173 &self.run
174 }
175
176 pub fn run_id(&self) -> Uuid {
178 self.run.id
179 }
180
181 pub fn steps(&self) -> &[TestStep] {
183 &self.steps
184 }
185
186 pub fn step_names(&self) -> Vec<&str> {
188 self.steps.iter().map(TestStep::name).collect()
189 }
190
191 pub fn step(&self, name: &str) -> &TestStep {
211 self.try_step(name).unwrap_or_else(|| {
212 panic!(
213 "no step named {name:?} in this run; steps are {:?}",
214 self.step_names()
215 )
216 })
217 }
218
219 pub fn try_step(&self, name: &str) -> Option<&TestStep> {
221 self.steps.iter().find(|step| step.name() == name)
222 }
223
224 pub fn output(&self) -> &Value {
226 self.steps.last().map_or(&NULL, TestStep::output)
227 }
228
229 pub fn duration(&self) -> Duration {
231 Duration::from_millis(self.run.duration_ms)
232 }
233
234 pub fn cost_usd(&self) -> Decimal {
236 self.run.cost_usd
237 }
238
239 pub fn error(&self) -> Option<&str> {
241 self.error.as_deref()
242 }
243
244 pub fn step_results(&self) -> &[StepResult] {
250 &self.step_results
251 }
252
253 pub fn is_completed(&self) -> bool {
255 self.status() == RunStatus::Completed
256 }
257}
258
259#[cfg(test)]
260mod tests {
261 use std::collections::HashMap;
262 use std::sync::Arc;
263
264 use chrono::Utc;
265 use serde_json::json;
266
267 use ironflow_store::memory::InMemoryStore;
268 use ironflow_store::models::{NewRun, NewStep, StepUpdate, TriggerKind, step_trace_id};
269 use ironflow_store::store::RunStore;
270
271 use super::*;
272
273 async fn persisted_run() -> (Run, Vec<Step>) {
278 let store = Arc::new(InMemoryStore::new());
279 let run = store
280 .create_run(NewRun {
281 created_by: None,
282 workflow_name: "deploy".to_string(),
283 trigger: TriggerKind::Manual,
284 payload: json!({}),
285 max_retries: 0,
286 handler_version: None,
287 labels: HashMap::new(),
288 scheduled_at: None,
289 idempotency_key: None,
290 max_cost_usd: None,
291 })
292 .await
293 .expect("create run")
294 .into_run();
295
296 for (position, name) in [(0u32, "build"), (1, "deploy")] {
297 let step = store
298 .create_step(NewStep {
299 run_id: run.id,
300 trace_id: step_trace_id(run.id, name, position),
301 name: name.to_string(),
302 kind: StepKind::Shell,
303 position,
304 input: Some(json!({"type": "shell", "command": name})),
305 is_error_handler: false,
306 })
307 .await
308 .expect("create step");
309
310 store
311 .update_step(
312 step.id,
313 StepUpdate {
314 status: Some(StepStatus::Running),
315 started_at: Some(Utc::now()),
316 ..StepUpdate::default()
317 },
318 )
319 .await
320 .expect("start step");
321
322 let update = if name == "build" {
323 StepUpdate {
324 status: Some(StepStatus::Completed),
325 output: Some(json!({"stdout": "built", "stderr": "", "exit_code": 0})),
326 duration_ms: Some(12),
327 completed_at: Some(Utc::now()),
328 ..StepUpdate::default()
329 }
330 } else {
331 StepUpdate {
332 status: Some(StepStatus::Failed),
333 error: Some("boom".to_string()),
334 completed_at: Some(Utc::now()),
335 ..StepUpdate::default()
336 }
337 };
338 store
339 .update_step(step.id, update)
340 .await
341 .expect("finish the step");
342 }
343
344 let steps = store.list_steps(run.id).await.expect("list steps");
345 let run = store
346 .get_run(run.id)
347 .await
348 .expect("get run")
349 .expect("the run exists");
350 (run, steps)
351 }
352
353 #[tokio::test]
354 async fn accessors_read_back_what_was_persisted() {
355 let (run, steps) = persisted_run().await;
356 let result = TestResult::new(run, steps, Vec::new(), Some("boom".to_string()));
357
358 assert_eq!(result.step_names(), vec!["build", "deploy"]);
359 assert_eq!(result.error(), Some("boom"));
360 assert!(result.step_results().is_empty());
361 assert!(!result.is_completed());
362
363 let build = result.step("build");
364 assert!(build.is_completed());
365 assert_eq!(build.output()["stdout"], "built");
366 assert_eq!(build.input()["command"], "build");
367 assert_eq!(build.duration(), Duration::from_millis(12));
368 assert_eq!(build.cost_usd(), Decimal::ZERO);
369 assert_eq!(build.kind(), &StepKind::Shell);
370 assert!(!build.is_error_handler());
371 assert_eq!(build.raw().name, "build");
372 assert!(build.error().is_none());
373 }
374
375 #[tokio::test]
376 async fn output_is_the_last_step_output() {
377 let (run, steps) = persisted_run().await;
378 let result = TestResult::new(run, steps, Vec::new(), None);
379
380 assert_eq!(result.output(), &Value::Null);
382 assert_eq!(result.step("deploy").status(), StepStatus::Failed);
383 assert_eq!(result.step("deploy").error(), Some("boom"));
384 }
385
386 #[tokio::test]
387 async fn run_level_accessors_reflect_the_store() {
388 let (run, steps) = persisted_run().await;
389 let run_id = run.id;
390 let result = TestResult::new(run, steps, Vec::new(), None);
391
392 assert_eq!(result.run_id(), run_id);
393 assert_eq!(result.run().workflow_name, "deploy");
394 assert_eq!(result.status(), RunStatus::Pending);
395 assert_eq!(result.duration(), Duration::ZERO);
396 assert_eq!(result.cost_usd(), Decimal::ZERO);
397 assert_eq!(result.steps().len(), 2);
398 }
399
400 #[tokio::test]
401 async fn try_step_returns_none_for_an_unknown_name() {
402 let (run, steps) = persisted_run().await;
403 let result = TestResult::new(run, steps, Vec::new(), None);
404
405 assert!(result.try_step("nope").is_none());
406 }
407
408 #[tokio::test]
409 #[should_panic(expected = "no step named \"nope\"")]
410 async fn step_panics_for_an_unknown_name() {
411 let (run, steps) = persisted_run().await;
412 let result = TestResult::new(run, steps, Vec::new(), None);
413
414 result.step("nope");
415 }
416
417 #[tokio::test]
418 async fn a_run_without_steps_has_a_null_output() {
419 let (run, _) = persisted_run().await;
420 let result = TestResult::new(run, Vec::new(), Vec::new(), None);
421
422 assert_eq!(result.output(), &Value::Null);
423 assert!(result.step_names().is_empty());
424 }
425}