ironflow_engine/testing/engine.rs
1//! [`TestEngine`] -- the builder that runs a handler against mocked steps.
2
3use std::fmt;
4use std::sync::Arc;
5
6use chrono::Utc;
7#[cfg(feature = "secret-store")]
8use rand::random;
9use serde_json::Value;
10use uuid::Uuid;
11
12use ironflow_core::decision::DecisionProvider;
13use ironflow_core::error::{AgentError, OperationError};
14use ironflow_core::provider::{AgentConfig, AgentOutput, AgentProvider};
15use ironflow_core::providers::record_replay::RecordReplayProvider;
16#[cfg(feature = "secret-store")]
17use ironflow_store::crypto::MasterKey;
18use ironflow_store::error::StoreError;
19use ironflow_store::memory::InMemoryStore;
20use ironflow_store::models::{RunStatus, StepApproval, StepKind, StepStatus, TriggerKind};
21use ironflow_store::store::{RunStore, Store};
22#[cfg(feature = "secret-store")]
23use ironflow_store::workflow_secrets::ScopedSecretStore;
24
25use crate::config::{HttpConfig, HumanInputConfig, ShellConfig};
26use crate::context::required_votes;
27use crate::engine::{Engine, WorkflowResult, chain_root};
28use crate::error::EngineError;
29use crate::executor::{ApprovalOutcome, HumanInputOutcome, SignalOutcome, StepInterceptor};
30use crate::handler::WorkflowHandler;
31use crate::testing::mocks::{
32 MissingAgentProvider, MockAgentProvider, MockHttpResponse, MockInterceptor, MockShellOutput,
33};
34use crate::testing::result::TestResult;
35
36/// Message of the assert guarding every builder method.
37const CONFIGURE_BEFORE_RUN: &str = "configure the TestEngine before its first run";
38
39/// Runs a [`WorkflowHandler`] against an in-memory store with mocked steps.
40///
41/// See the [module documentation](crate::testing) for what the harness replaces
42/// and what it does not.
43///
44/// # Examples
45///
46/// ```no_run
47/// use ironflow_engine::prelude::*;
48/// use ironflow_engine::testing::{MockShellOutput, TestEngine};
49/// use ironflow_store::models::RunStatus;
50/// use serde_json::json;
51///
52/// # struct Deploy;
53/// # impl WorkflowHandler for Deploy {
54/// # fn name(&self) -> &str { "deploy" }
55/// # fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
56/// # Box::pin(async move { ctx.shell("deploy", ShellConfig::new("./deploy.sh")).await?; Ok(()) })
57/// # }
58/// # }
59/// # async fn example() -> Result<(), EngineError> {
60/// let result = TestEngine::new()
61/// .with_handler(Deploy)
62/// .with_mock_shell(|_cfg| Ok(MockShellOutput::ok("deployed")))
63/// .run(json!({"env": "prod"}))
64/// .await?;
65///
66/// assert_eq!(result.status(), RunStatus::Completed);
67/// # Ok(())
68/// # }
69/// ```
70pub struct TestEngine {
71 store: Arc<InMemoryStore>,
72 handlers: Vec<Box<dyn WorkflowHandler>>,
73 primary: Option<String>,
74 provider: Option<Arc<dyn AgentProvider>>,
75 decision_provider: Option<Arc<dyn DecisionProvider>>,
76 mocks: MockInterceptor,
77 engine: Option<Engine>,
78 #[cfg(feature = "secret-store")]
79 secrets: Vec<(String, String)>,
80}
81
82impl Default for TestEngine {
83 fn default() -> Self {
84 Self::new()
85 }
86}
87
88impl fmt::Debug for TestEngine {
89 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
90 f.debug_struct("TestEngine")
91 .field("primary", &self.primary)
92 .field("mocks", &self.mocks)
93 .field("built", &self.engine.is_some())
94 .finish_non_exhaustive()
95 }
96}
97
98impl TestEngine {
99 /// A harness with no handler and no mock.
100 ///
101 /// With the `secret-store` feature the store gets a freshly generated
102 /// master key, so reading a secret that was never set resolves to `None`
103 /// instead of failing with "no master key configured". Seed values with
104 /// `with_secret`.
105 ///
106 /// # Examples
107 ///
108 /// ```
109 /// use ironflow_engine::testing::TestEngine;
110 ///
111 /// let harness = TestEngine::new();
112 /// assert!(format!("{harness:?}").contains("TestEngine"));
113 /// ```
114 pub fn new() -> Self {
115 #[cfg_attr(not(feature = "secret-store"), allow(unused_mut))]
116 let mut store = InMemoryStore::new();
117 #[cfg(feature = "secret-store")]
118 store.set_master_key(
119 MasterKey::from_bytes(&random::<[u8; 32]>())
120 .expect("32 bytes is a valid master key length"),
121 );
122 Self {
123 store: Arc::new(store),
124 handlers: Vec::new(),
125 primary: None,
126 provider: None,
127 decision_provider: None,
128 mocks: MockInterceptor::new(),
129 engine: None,
130 #[cfg(feature = "secret-store")]
131 secrets: Vec::new(),
132 }
133 }
134
135 /// Register a handler. The first one registered is what
136 /// [`run`](Self::run) executes.
137 ///
138 /// # Panics
139 ///
140 /// Panics when called after the first run: the engine is built once, so a
141 /// later registration would be silently ignored.
142 ///
143 /// # Examples
144 ///
145 /// ```no_run
146 /// use ironflow_engine::prelude::*;
147 /// use ironflow_engine::testing::TestEngine;
148 ///
149 /// # struct Deploy;
150 /// # impl WorkflowHandler for Deploy {
151 /// # fn name(&self) -> &str { "deploy" }
152 /// # fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
153 /// # Box::pin(async move { Ok(()) })
154 /// # }
155 /// # }
156 /// let harness = TestEngine::new().with_handler(Deploy);
157 /// # let _ = harness;
158 /// ```
159 pub fn with_handler(mut self, handler: impl WorkflowHandler + 'static) -> Self {
160 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
161 if self.primary.is_none() {
162 self.primary = Some(handler.name().to_string());
163 }
164 self.handlers.push(Box::new(handler));
165 self
166 }
167
168 /// Answer every shell step with `f` instead of spawning a process.
169 ///
170 /// Returning `Err` reproduces a shell failure the same way a non-zero
171 /// [`MockShellOutput::exit_code`] does: a non-zero exit code is an error
172 /// unless the step set `exit_code_as_output`.
173 ///
174 /// # Panics
175 ///
176 /// Panics when called after the first run.
177 ///
178 /// # Examples
179 ///
180 /// ```no_run
181 /// use ironflow_engine::testing::{MockShellOutput, TestEngine};
182 ///
183 /// let harness = TestEngine::new().with_mock_shell(|cfg| {
184 /// if cfg.command.starts_with("git ") {
185 /// Ok(MockShellOutput::ok("abc1234"))
186 /// } else {
187 /// Ok(MockShellOutput::failed(127, "command not found"))
188 /// }
189 /// });
190 /// # let _ = harness;
191 /// ```
192 pub fn with_mock_shell(
193 mut self,
194 f: impl Fn(&ShellConfig) -> Result<MockShellOutput, OperationError> + Send + Sync + 'static,
195 ) -> Self {
196 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
197 self.mocks = self.mocks.shell(f);
198 self
199 }
200
201 /// Answer every HTTP step with `f` instead of sending a request.
202 ///
203 /// A non-2xx [`MockHttpResponse`] is a normal output, like in production.
204 /// Return `Err(OperationError::Http { status: None, .. })` to simulate a
205 /// transport failure.
206 ///
207 /// # Panics
208 ///
209 /// Panics when called after the first run.
210 ///
211 /// # Examples
212 ///
213 /// ```no_run
214 /// use ironflow_engine::testing::{MockHttpResponse, TestEngine};
215 /// use serde_json::json;
216 ///
217 /// let harness = TestEngine::new()
218 /// .with_mock_http(|_cfg| Ok(MockHttpResponse::json(201, &json!({"id": 7}))));
219 /// # let _ = harness;
220 /// ```
221 pub fn with_mock_http(
222 mut self,
223 f: impl Fn(&HttpConfig) -> Result<MockHttpResponse, OperationError> + Send + Sync + 'static,
224 ) -> Self {
225 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
226 self.mocks = self.mocks.http(f);
227 self
228 }
229
230 /// Resolve every approval gate with `outcome` instead of suspending.
231 ///
232 /// Without this, a gated handler ends the run in
233 /// [`RunStatus::AwaitingApproval`] and [`resume`](Self::resume) continues it.
234 ///
235 /// # Panics
236 ///
237 /// Panics when called after the first run.
238 ///
239 /// # Examples
240 ///
241 /// ```no_run
242 /// use ironflow_engine::testing::{ApprovalOutcome, TestEngine};
243 ///
244 /// let harness = TestEngine::new().with_mock_approval(ApprovalOutcome::Approved);
245 /// # let _ = harness;
246 /// ```
247 pub fn with_mock_approval(mut self, outcome: ApprovalOutcome) -> Self {
248 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
249 self.mocks = self.mocks.approval(outcome);
250 self
251 }
252
253 /// Answer every human input step with `f` instead of suspending.
254 ///
255 /// `f` receives the step name and its config. Without this, a handler that
256 /// asks for a human input ends the run in [`RunStatus::AwaitingApproval`];
257 /// write the answer on the step through the store, then call
258 /// [`resume`](Self::resume).
259 ///
260 /// # Panics
261 ///
262 /// Panics when called after the first run.
263 ///
264 /// # Examples
265 ///
266 /// ```no_run
267 /// use ironflow_engine::testing::{HumanInputOutcome, TestEngine};
268 /// use serde_json::json;
269 ///
270 /// let harness = TestEngine::new().with_mock_human_input(|_name, _cfg| {
271 /// HumanInputOutcome::Provided(json!({"answers": ["staging"]}))
272 /// });
273 /// # let _ = harness;
274 /// ```
275 pub fn with_mock_human_input(
276 mut self,
277 f: impl Fn(&str, &HumanInputConfig) -> HumanInputOutcome + Send + Sync + 'static,
278 ) -> Self {
279 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
280 self.mocks = self.mocks.human_input(f);
281 self
282 }
283
284 /// Resolve every signal step with `f` instead of waiting for a signal.
285 ///
286 /// `f` receives the step name, the signal name and the key, and returns
287 /// [`SignalOutcome::Received`] with the payload, or
288 /// [`SignalOutcome::TimedOut`] to make `ctx.wait_for_signal` return `None`.
289 /// Without this, a handler that waits for a signal ends the run in
290 /// [`RunStatus::Sleeping`].
291 ///
292 /// # Panics
293 ///
294 /// Panics when called after the first run.
295 ///
296 /// # Examples
297 ///
298 /// ```no_run
299 /// use ironflow_engine::testing::{SignalOutcome, TestEngine};
300 /// use serde_json::json;
301 ///
302 /// let harness = TestEngine::new().with_mock_signal(|_step, _name, _key| {
303 /// SignalOutcome::Received(json!({"status": "success"}))
304 /// });
305 /// # let _ = harness;
306 /// ```
307 pub fn with_mock_signal(
308 mut self,
309 f: impl Fn(&str, &str, &str) -> SignalOutcome + Send + Sync + 'static,
310 ) -> Self {
311 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
312 self.mocks = self.mocks.signal(f);
313 self
314 }
315
316 /// Answer every agent step with `f` instead of invoking a backend.
317 ///
318 /// # Panics
319 ///
320 /// Panics when called after the first run.
321 ///
322 /// # Examples
323 ///
324 /// ```no_run
325 /// use ironflow_core::provider::AgentOutput;
326 /// use ironflow_engine::testing::TestEngine;
327 /// use serde_json::json;
328 ///
329 /// let harness = TestEngine::new()
330 /// .with_mock_agent(|_cfg| Ok(AgentOutput::new(json!({"score": 9}))));
331 /// # let _ = harness;
332 /// ```
333 pub fn with_mock_agent(
334 mut self,
335 f: impl Fn(&AgentConfig) -> Result<AgentOutput, AgentError> + Send + Sync + 'static,
336 ) -> Self {
337 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
338 self.provider = Some(Arc::new(MockAgentProvider::new(f)));
339 self
340 }
341
342 /// Replay agent steps from fixtures recorded in `fixtures_dir`.
343 ///
344 /// `fixtures_dir` is the **directory**, not a file:
345 /// [`RecordReplayProvider`] keys each fixture by a hash of the
346 /// [`AgentConfig`] and stores it as `<hash>.json` inside it. A missing
347 /// fixture falls back to [`MissingAgentProvider`], so the step fails loudly
348 /// instead of reaching the real Claude CLI.
349 ///
350 /// # Panics
351 ///
352 /// Panics when called after the first run.
353 ///
354 /// # Examples
355 ///
356 /// ```no_run
357 /// use ironflow_engine::testing::TestEngine;
358 ///
359 /// let harness = TestEngine::new().with_recorded_agent("tests/fixtures");
360 /// # let _ = harness;
361 /// ```
362 pub fn with_recorded_agent(mut self, fixtures_dir: &str) -> Self {
363 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
364 self.provider = Some(Arc::new(RecordReplayProvider::replay(
365 MissingAgentProvider,
366 fixtures_dir,
367 )));
368 self
369 }
370
371 /// Use an arbitrary [`AgentProvider`] for agent steps.
372 ///
373 /// The escape hatch for anything the three `with_mock_*` methods do not
374 /// cover, such as recording new fixtures.
375 ///
376 /// # Panics
377 ///
378 /// Panics when called after the first run.
379 ///
380 /// # Examples
381 ///
382 /// ```no_run
383 /// use std::sync::Arc;
384 ///
385 /// use ironflow_core::provider::AgentProvider;
386 /// use ironflow_engine::testing::TestEngine;
387 ///
388 /// # fn example(provider: Arc<dyn AgentProvider>) {
389 /// let harness = TestEngine::new().with_agent_provider(provider);
390 /// # let _ = harness;
391 /// # }
392 /// ```
393 pub fn with_agent_provider(mut self, provider: Arc<dyn AgentProvider>) -> Self {
394 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
395 self.provider = Some(provider);
396 self
397 }
398
399 /// Use a [`DecisionProvider`] for `ctx.decision(...)` steps.
400 ///
401 /// Decision steps are not intercepted: without a provider they fail with
402 /// [`EngineError::NoDecisionProvider`].
403 ///
404 /// # Panics
405 ///
406 /// Panics when called after the first run.
407 ///
408 /// # Examples
409 ///
410 /// ```no_run
411 /// use std::sync::Arc;
412 ///
413 /// use ironflow_core::decision::DecisionProvider;
414 /// use ironflow_engine::testing::TestEngine;
415 ///
416 /// # fn example(provider: Arc<dyn DecisionProvider>) {
417 /// let harness = TestEngine::new().with_decision_provider(provider);
418 /// # let _ = harness;
419 /// # }
420 /// ```
421 pub fn with_decision_provider(mut self, provider: Arc<dyn DecisionProvider>) -> Self {
422 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
423 self.decision_provider = Some(provider);
424 self
425 }
426
427 /// Seed a secret readable by the workflow under test.
428 ///
429 /// The secret is written in the scope of the workflow being run, the same
430 /// scope as `ctx.secrets()` and as the operations run through
431 /// `ctx.operation`. Setting the same key twice keeps the last value.
432 /// Requires the `secret-store` feature.
433 ///
434 /// # Panics
435 ///
436 /// Panics when called after the first run.
437 ///
438 /// # Examples
439 ///
440 /// ```no_run
441 /// use ironflow_engine::testing::TestEngine;
442 ///
443 /// let harness = TestEngine::new().with_secret("git_token", "ghp_test");
444 /// # let _ = harness;
445 /// ```
446 #[cfg(feature = "secret-store")]
447 pub fn with_secret(mut self, key: &str, value: &str) -> Self {
448 assert!(self.engine.is_none(), "{CONFIGURE_BEFORE_RUN}");
449 self.secrets.push((key.to_string(), value.to_string()));
450 self
451 }
452
453 /// The store backing this harness, for assertions the accessors do not
454 /// cover (child runs, step dependencies, logs).
455 ///
456 /// # Examples
457 ///
458 /// ```no_run
459 /// use ironflow_engine::error::EngineError;
460 /// use ironflow_engine::testing::TestEngine;
461 /// use ironflow_store::store::RunStore;
462 /// use uuid::Uuid;
463 ///
464 /// # async fn example(harness: &TestEngine, run_id: Uuid) -> Result<(), EngineError> {
465 /// let steps = harness.store().list_steps(run_id).await?;
466 /// assert!(!steps.is_empty());
467 /// # Ok(())
468 /// # }
469 /// ```
470 pub fn store(&self) -> Arc<InMemoryStore> {
471 self.store.clone()
472 }
473
474 /// Build the underlying [`Engine`] on first use.
475 ///
476 /// Handlers are drained into it, which is why every builder method asserts
477 /// that no run has happened yet.
478 fn ensure_engine(&mut self) -> Result<(), EngineError> {
479 if self.engine.is_some() {
480 return Ok(());
481 }
482
483 let store: Arc<dyn Store> = self.store.clone();
484 let provider = self
485 .provider
486 .clone()
487 .unwrap_or_else(|| Arc::new(MissingAgentProvider));
488 let mocks: Arc<dyn StepInterceptor> = Arc::new(self.mocks.clone());
489 let mut engine = Engine::new(store, provider).with_step_interceptor(mocks);
490 if let Some(decision_provider) = self.decision_provider.clone() {
491 engine = engine.with_decision_provider(decision_provider);
492 }
493 for handler in self.handlers.drain(..) {
494 engine.register_boxed(handler)?;
495 }
496
497 self.engine = Some(engine);
498 Ok(())
499 }
500
501 /// Run the first handler registered with [`with_handler`](Self::with_handler).
502 ///
503 /// A handler that fails is not an error: the returned [`TestResult`] then
504 /// carries [`RunStatus::Failed`] and the message in
505 /// [`TestResult::error`].
506 ///
507 /// # Errors
508 ///
509 /// Returns [`EngineError::InvalidWorkflow`] when no handler was registered
510 /// or two handlers share a name, and [`EngineError::Store`] when the
511 /// in-memory store rejects a write.
512 ///
513 /// # Examples
514 ///
515 /// ```no_run
516 /// use ironflow_engine::prelude::*;
517 /// use ironflow_engine::testing::{MockShellOutput, TestEngine};
518 /// use serde_json::json;
519 ///
520 /// # struct Deploy;
521 /// # impl WorkflowHandler for Deploy {
522 /// # fn name(&self) -> &str { "deploy" }
523 /// # fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
524 /// # Box::pin(async move { Ok(()) })
525 /// # }
526 /// # }
527 /// # async fn example() -> Result<(), EngineError> {
528 /// let result = TestEngine::new()
529 /// .with_handler(Deploy)
530 /// .with_mock_shell(|_cfg| Ok(MockShellOutput::ok("done")))
531 /// .run(json!({}))
532 /// .await?;
533 /// assert!(result.is_completed());
534 /// # Ok(())
535 /// # }
536 /// ```
537 pub async fn run(&mut self, payload: Value) -> Result<TestResult, EngineError> {
538 let name = self.primary.clone().ok_or_else(|| {
539 EngineError::InvalidWorkflow(
540 "TestEngine has no handler: call with_handler(...) first".to_string(),
541 )
542 })?;
543 self.run_workflow(&name, payload).await
544 }
545
546 /// Run a specific registered handler by name.
547 ///
548 /// # Errors
549 ///
550 /// Same as [`run`](Self::run), plus [`EngineError::InvalidWorkflow`] when
551 /// `name` matches no registered handler.
552 ///
553 /// # Examples
554 ///
555 /// ```no_run
556 /// use ironflow_engine::error::EngineError;
557 /// use ironflow_engine::testing::TestEngine;
558 /// use serde_json::json;
559 ///
560 /// # async fn example(harness: &mut TestEngine) -> Result<(), EngineError> {
561 /// let result = harness.run_workflow("child", json!({"id": 1})).await?;
562 /// assert!(result.is_completed());
563 /// # Ok(())
564 /// # }
565 /// ```
566 pub async fn run_workflow(
567 &mut self,
568 name: &str,
569 payload: Value,
570 ) -> Result<TestResult, EngineError> {
571 self.ensure_engine()?;
572
573 #[cfg(feature = "secret-store")]
574 {
575 let scoped = ScopedSecretStore::for_workflow(
576 Uuid::new_v5(&Uuid::NAMESPACE_OID, name.as_bytes()),
577 self.store.clone(),
578 );
579 for (key, value) in &self.secrets {
580 scoped.set(key, value).await?;
581 }
582 }
583
584 let engine = self.engine.as_ref().expect("ensure_engine built it");
585
586 // Enqueue then execute, the way the worker does, so the run id is known
587 // before execution and a failed run can still be read back.
588 let trigger = TriggerKind::Manual;
589 let run = engine.enqueue_handler(name, trigger, payload, 0).await?;
590 self.store
591 .update_run_status(run.id, RunStatus::Running)
592 .await?;
593 let execution = engine.execute_handler_run(run.id).await;
594 self.collect(run.id, execution).await
595 }
596
597 /// Resume a run suspended on an approval gate, a human input, a delay or a
598 /// signal wait, the way the API server and the waker do.
599 ///
600 /// An open approval gate of `run_id` is approved on behalf of as many
601 /// approvers as it requires, like the approve route: one vote, or the
602 /// `required_approvers` of its requirement. For a gated sub-workflow
603 /// child, pass the child run id: the gate lives there.
604 ///
605 /// A suspended sub-workflow child run is resumed through its root run,
606 /// like in production: the root replays and re-enters the child. The
607 /// returned [`TestResult`] then describes the root run.
608 ///
609 /// # Errors
610 ///
611 /// Returns [`EngineError::Store`] when the run does not exist or is not
612 /// resumable, and [`EngineError::InvalidWorkflow`] when its handler is no
613 /// longer registered.
614 ///
615 /// # Panics
616 ///
617 /// Never in practice: the engine is built by `ensure_engine` just before.
618 ///
619 /// # Examples
620 ///
621 /// ```no_run
622 /// use ironflow_engine::error::EngineError;
623 /// use ironflow_engine::testing::TestEngine;
624 /// use ironflow_store::models::RunStatus;
625 /// use serde_json::json;
626 ///
627 /// # async fn example(harness: &mut TestEngine) -> Result<(), EngineError> {
628 /// let suspended = harness.run(json!({})).await?;
629 /// assert_eq!(suspended.status(), RunStatus::AwaitingApproval);
630 ///
631 /// let resumed = harness.resume(suspended.run_id()).await?;
632 /// assert_eq!(resumed.status(), RunStatus::Completed);
633 /// # Ok(())
634 /// # }
635 /// ```
636 pub async fn resume(&mut self, run_id: Uuid) -> Result<TestResult, EngineError> {
637 self.ensure_engine()?;
638 let engine = self.engine.as_ref().expect("ensure_engine built it");
639
640 let run = self
641 .store
642 .get_run(run_id)
643 .await?
644 .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;
645 // A sleeping run is woken through `Pending`, like the waker does.
646 if run.status.state == RunStatus::Sleeping {
647 self.store
648 .update_run_status(run_id, RunStatus::Pending)
649 .await?;
650 }
651 for gate in self.store.list_steps(run_id).await? {
652 if gate.kind != StepKind::Approval || gate.status.state != StepStatus::AwaitingApproval
653 {
654 continue;
655 }
656 for _ in gate.approvals.len() as u32..required_votes(&gate) {
657 self.store
658 .record_step_approval(
659 gate.id,
660 StepApproval {
661 user_id: Uuid::now_v7(),
662 approved_by: "test-engine".to_string(),
663 at: Utc::now(),
664 },
665 )
666 .await?;
667 }
668 }
669 self.store
670 .update_run_status(run_id, RunStatus::Running)
671 .await?;
672 let execution = engine.resume_run(run_id).await;
673 let reported = chain_root(&run).unwrap_or(run_id);
674 self.collect(reported, execution).await
675 }
676
677 /// Read the run and its steps back from the store.
678 ///
679 /// The engine returns `Err` for a failed run, so the steps are always read
680 /// from the store rather than from the execution result.
681 async fn collect(
682 &self,
683 run_id: Uuid,
684 execution: Result<WorkflowResult, EngineError>,
685 ) -> Result<TestResult, EngineError> {
686 let run = self
687 .store
688 .get_run(run_id)
689 .await?
690 .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;
691 let steps = self.store.list_steps(run_id).await?;
692
693 let (step_results, error) = match execution {
694 Ok(result) => (result.steps, run.error.clone()),
695 Err(err) => (Vec::new(), Some(err.to_string())),
696 };
697
698 Ok(TestResult::new(run, steps, step_results, error))
699 }
700}
701
702#[cfg(test)]
703mod tests {
704 use std::time::Duration;
705
706 use async_trait::async_trait;
707 use serde_json::json;
708 use tokio::time::timeout;
709
710 use ironflow_core::operation::{Operation, OperationContext};
711
712 use crate::context::WorkflowContext;
713 use crate::handler::HandlerFuture;
714
715 use super::*;
716
717 /// Reads `git_token` through the operation context.
718 struct ReadsToken;
719
720 #[async_trait]
721 impl Operation for ReadsToken {
722 fn kind(&self) -> &str {
723 "reads-token"
724 }
725
726 async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
727 let token = ctx.secrets().get("git_token").await?;
728 Ok(json!({"token": token.map(|secret| secret.value)}))
729 }
730 }
731
732 /// Runs [`ReadsToken`] through `ctx.operation`.
733 struct UsesSecretOp;
734
735 impl WorkflowHandler for UsesSecretOp {
736 fn name(&self) -> &str {
737 "uses-secret-op"
738 }
739
740 fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
741 Box::pin(async move {
742 ctx.operation("read-token", &ReadsToken).await?;
743 Ok(())
744 })
745 }
746 }
747
748 #[tokio::test]
749 async fn operation_reading_absent_secret_completes_under_test_engine() {
750 timeout(Duration::from_secs(10), async {
751 let result = TestEngine::new()
752 .with_handler(UsesSecretOp)
753 .run(json!({}))
754 .await
755 .unwrap();
756
757 assert_eq!(
758 result.status(),
759 RunStatus::Completed,
760 "{:?}",
761 result.error()
762 );
763 assert!(result.step("read-token").output()["token"].is_null());
764 })
765 .await
766 .expect("test timed out");
767 }
768
769 #[cfg(feature = "secret-store")]
770 #[tokio::test]
771 async fn with_secret_value_reaches_operation_via_ctx_operation() {
772 timeout(Duration::from_secs(10), async {
773 let result = TestEngine::new()
774 .with_handler(UsesSecretOp)
775 .with_secret("git_token", "ghp_test")
776 .run(json!({}))
777 .await
778 .unwrap();
779
780 assert_eq!(
781 result.status(),
782 RunStatus::Completed,
783 "{:?}",
784 result.error()
785 );
786 assert_eq!(result.step("read-token").output()["token"], "ghp_test");
787 })
788 .await
789 .expect("test timed out");
790 }
791
792 #[cfg(feature = "secret-store")]
793 #[tokio::test]
794 async fn with_secret_last_value_wins() {
795 timeout(Duration::from_secs(10), async {
796 let result = TestEngine::new()
797 .with_handler(UsesSecretOp)
798 .with_secret("git_token", "first")
799 .with_secret("git_token", "second")
800 .run(json!({}))
801 .await
802 .unwrap();
803
804 assert_eq!(result.step("read-token").output()["token"], "second");
805 })
806 .await
807 .expect("test timed out");
808 }
809
810 #[cfg(feature = "secret-store")]
811 #[tokio::test]
812 async fn with_secret_is_readable_through_ctx_secrets() {
813 timeout(Duration::from_secs(10), async {
814 let mut harness = TestEngine::new()
815 .with_handler(UsesSecretOp)
816 .with_secret("k", "v");
817 harness.run(json!({})).await.unwrap();
818
819 let store = harness.store();
820 let scope = |name: &str| {
821 ScopedSecretStore::for_workflow(
822 Uuid::new_v5(&Uuid::NAMESPACE_OID, name.as_bytes()),
823 store.clone(),
824 )
825 };
826 let own = scope("uses-secret-op").get("k").await.unwrap();
827 assert_eq!(own.unwrap().value, "v");
828 assert!(scope("other-workflow").get("k").await.unwrap().is_none());
829 })
830 .await
831 .expect("test timed out");
832 }
833
834 #[cfg(feature = "secret-store")]
835 #[tokio::test]
836 #[should_panic(expected = "configure the TestEngine before its first run")]
837 async fn with_secret_after_first_run_panics() {
838 let mut harness = TestEngine::new().with_handler(UsesSecretOp);
839 harness.run(json!({})).await.unwrap();
840 let _ = harness.with_secret("git_token", "late");
841 }
842}