Merge pull request #22 from nassor/testing-module

Add a public `testing` module with batteries-included test fixtures

Author: nassor

cano/Cargo.toml

@@ -31,8 +31,9 @@ metrics = { version = "0.24.5", optional = true }
 scheduler = ["dep:chrono", "dep:cron", "tokio/signal"]
 tracing = ["dep:tracing"]
 recovery = ["dep:redb", "dep:postcard"]
-all = ["scheduler", "tracing", "recovery", "metrics"]
 metrics = ["dep:metrics"]
+testing = []
+all = ["scheduler", "tracing", "recovery", "metrics", "testing"]
 
 [dev-dependencies]
 axum = "0.8.9"
@@ -263,6 +264,11 @@ path = "examples/store_custom_backend.rs"
 name = "resources_advanced"
 path = "examples/resources_advanced.rs"
 
+[[example]]
+name = "testing_helpers"
+path = "examples/testing_helpers.rs"
+required-features = ["testing"]
+
 [[bench]]
 name = "workflow_performance"
 harness = false

cano/examples/testing_helpers.rs

@@ -0,0 +1,161 @@
+//! # `cano::testing` — batteries-included test fixtures
+//!
+//! Run with: `cargo run --example testing_helpers --features testing`
+//!
+//! Everything here lives behind the `testing` feature and is imported wholesale with
+//! `use cano::testing::*;`. This example exercises each helper the way a test would:
+//!
+//! - [`RecordingObserver`] — capture and assert the path a workflow took.
+//! - [`InMemoryCheckpointStore`] — a process-local checkpoint store (no `recovery`
+//!   feature, no on-disk file) whose `Checkpoint` events the observer records.
+//! - [`TestResources`] — build the [`Resources`] a workflow needs in one chain.
+//! - [`panic_on_attempt`] — a task that panics; the engine converts it to an error
+//!   (panic safety) and fails fast.
+//! - [`assert_compensation_ran`] — assert the order a saga rolled back in.
+
+use std::sync::{Arc, Mutex};
+
+use cano::prelude::*;
+use cano::testing::*;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+enum Step {
+    Start,
+    Work,
+    Finish,
+    Done,
+}
+
+struct StartTask;
+#[task(state = Step)]
+impl StartTask {
+    async fn run_bare(&self) -> Result<TaskResult<Step>, CanoError> {
+        Ok(TaskResult::Single(Step::Work))
+    }
+}
+
+struct WorkTask;
+#[task(state = Step)]
+impl WorkTask {
+    async fn run(&self, res: &Resources) -> Result<TaskResult<Step>, CanoError> {
+        let store = res.get::<MemoryStore, _>("store")?;
+        store.put("processed", 42_u32)?;
+        Ok(TaskResult::Single(Step::Done))
+    }
+}
+
+// ---- A tiny saga whose compensations record the order they undo in. -------------------
+
+/// Shared, ordered log of which compensations ran. Registered as a resource so every
+/// step can reach it; `Clone` shares the inner `Arc`, so a handle kept outside the
+/// workflow reads the same `Vec`.
+#[derive(Default, Clone)]
+struct CompensationLog(Arc<Mutex<Vec<String>>>);
+#[resource]
+impl Resource for CompensationLog {}
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+struct Unit;
+
+struct Reserve;
+#[saga::task(state = Step)]
+impl Reserve {
+    type Output = Unit;
+    async fn run(&self, _res: &Resources) -> Result<(TaskResult<Step>, Unit), CanoError> {
+        Ok((TaskResult::Single(Step::Work), Unit))
+    }
+    async fn compensate(&self, res: &Resources, _out: Unit) -> Result<(), CanoError> {
+        res.get::<CompensationLog, _>("log")?
+            .0
+            .lock()
+            .unwrap()
+            .push("reserve".into());
+        Ok(())
+    }
+}
+
+struct Charge;
+#[saga::task(state = Step)]
+impl Charge {
+    type Output = Unit;
+    async fn run(&self, _res: &Resources) -> Result<(TaskResult<Step>, Unit), CanoError> {
+        Ok((TaskResult::Single(Step::Finish), Unit))
+    }
+    async fn compensate(&self, res: &Resources, _out: Unit) -> Result<(), CanoError> {
+        res.get::<CompensationLog, _>("log")?
+            .0
+            .lock()
+            .unwrap()
+            .push("charge".into());
+        Ok(())
+    }
+}
+
+struct Boom;
+#[task(state = Step)]
+impl Boom {
+    fn config(&self) -> TaskConfig {
+        TaskConfig::minimal()
+    }
+    async fn run_bare(&self) -> Result<TaskResult<Step>, CanoError> {
+        Err(CanoError::task_execution("boom"))
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // 1. RecordingObserver + InMemoryCheckpointStore + TestResources --------------------
+    let observer = Arc::new(RecordingObserver::new());
+    let checkpoints = Arc::new(InMemoryCheckpointStore::new());
+    let resources = TestResources::new().with_store("store").build();
+
+    let workflow = Workflow::new(resources)
+        .register(Step::Start, StartTask)
+        .register(Step::Work, WorkTask)
+        .add_exit_state(Step::Done)
+        .with_observer(observer.clone())
+        .with_checkpoint_store(checkpoints.clone())
+        .with_workflow_id("demo-run");
+
+    let final_state = workflow.orchestrate(Step::Start).await?;
+    assert_eq!(final_state, Step::Done);
+
+    // The observer captured the whole path and the checkpoint appends along the way.
+    observer.assert_path(&["Start", "Work", "Done"]);
+    observer.assert_completed_with("Done");
+    let checkpoint_events = observer
+        .events()
+        .into_iter()
+        .filter(|e| matches!(e, RecordedEvent::Checkpoint { .. }))
+        .count();
+    println!("observer recorded path {:?}", observer.states_entered());
+    println!("observer saw {checkpoint_events} checkpoint append(s)");
+
+    // 2. panic_on_attempt — panic safety: the panic becomes an error, fails fast. -------
+    let panicky = Workflow::bare()
+        .register(Step::Start, panic_on_attempt(1, Step::Done))
+        .add_exit_state(Step::Done);
+    match panicky.orchestrate(Step::Start).await {
+        Ok(_) => unreachable!("the task panics on its first attempt"),
+        Err(e) => println!("panic_on_attempt surfaced as error: {e}"),
+    }
+
+    // 3. assert_compensation_ran — a saga that rolls back in reverse. --------------------
+    let log = CompensationLog::default();
+    let handle = log.clone();
+    let saga_resources = Resources::new().insert("log", log);
+    let saga = Workflow::new(saga_resources)
+        .register_with_compensation(Step::Start, Reserve)
+        .register_with_compensation(Step::Work, Charge)
+        .register(Step::Finish, Boom) // fails → drains the compensation stack in reverse
+        .add_exit_state(Step::Done);
+    let _ = saga.orchestrate(Step::Start).await; // expected to fail and roll back
+
+    let ran = handle.0.lock().unwrap().clone();
+    // Charge ran last, so it compensates first; then Reserve.
+    assert_compensation_ran(&ran, &["charge", "reserve"]);
+    println!("compensation ran in order: {ran:?}");
+
+    println!("\nall testing helpers exercised ✔");
+    Ok(())
+}

cano/src/lib.rs

@@ -217,6 +217,11 @@
 //! - [`saga`]: [`CompensatableTask`] — pair a forward step with a compensating action; failures roll back via [`Workflow::register_with_compensation`]
 //! - [`rate_limit`]: [`RateLimiter`] (token bucket) and [`WindowedRateLimiter`] (fixed window) throttles, both implementing [`Meter`] so they compose into a [`MultiRateLimiter`] that enforces several weighted tiers (e.g. a 5h + 7d + per-model limit) at once
 //! - [`store`]: [`MemoryStore`] — a typed in-memory store that implements [`Resource`]
+//! - `testing` (requires `testing` feature): batteries-included test fixtures —
+//!   a `RecordingObserver`, a public `InMemoryCheckpointStore`, a `TestResources`
+//!   builder, a `panic_on_attempt` task factory and an `assert_compensation_ran`
+//!   saga helper. Import explicitly with `use cano::testing::*;` — it is deliberately
+//!   **not** in the [`prelude`] so production code never picks it up by accident.
 //! - [`error`]: [`CanoError`] variants and the [`CanoResult`] alias
 //!
 //! ## Getting Started
@@ -242,6 +247,9 @@ pub mod metrics;
 #[cfg(feature = "scheduler")]
 pub mod scheduler;
 
+#[cfg(feature = "testing")]
+pub mod testing;
+
 // Core public API - simplified imports
 pub use circuit::{CircuitBreaker, CircuitPolicy, CircuitState, Permit as CircuitPermit};
 pub use error::{CanoError, CanoResult};

cano/src/observer.rs

@@ -315,46 +315,11 @@ impl WorkflowObserver for MetricsObserver {
 #[cfg(test)]
 mod tests {
     use crate::prelude::*;
+    use crate::workflow::test_support::EventLog;
     use std::borrow::Cow;
-    use std::sync::{Arc, Mutex};
+    use std::sync::Arc;
     use std::time::Duration;
 
-    /// Test observer that appends a stringified record of every event it receives.
-    #[derive(Default)]
-    struct RecordingObserver {
-        events: Mutex<Vec<String>>,
-    }
-
-    impl RecordingObserver {
-        fn events(&self) -> Vec<String> {
-            self.events.lock().unwrap().clone()
-        }
-        fn record(&self, event: String) {
-            self.events.lock().unwrap().push(event);
-        }
-    }
-
-    impl WorkflowObserver for RecordingObserver {
-        fn on_state_enter(&self, state: &str) {
-            self.record(format!("state_enter:{state}"));
-        }
-        fn on_task_start(&self, task_id: &str) {
-            self.record(format!("task_start:{task_id}"));
-        }
-        fn on_task_success(&self, task_id: &str) {
-            self.record(format!("task_success:{task_id}"));
-        }
-        fn on_task_failure(&self, task_id: &str, _err: &CanoError) {
-            self.record(format!("task_failure:{task_id}"));
-        }
-        fn on_retry(&self, task_id: &str, attempt: u32) {
-            self.record(format!("retry:{task_id}:{attempt}"));
-        }
-        fn on_circuit_open(&self, task_id: &str) {
-            self.record(format!("circuit_open:{task_id}"));
-        }
-    }
-
     #[derive(Clone, Debug, PartialEq, Eq, Hash)]
     enum S {
         Start,
@@ -411,15 +376,15 @@ mod tests {
 
     #[tokio::test]
     async fn observer_fires_on_success_path() {
-        let observer = Arc::new(RecordingObserver::default());
+        let (obs, rec) = EventLog::new();
         let workflow = Workflow::bare()
             .register(S::Start, OkTask)
             .add_exit_state(S::Done)
-            .with_observer(observer.clone());
+            .with_observer(Arc::new(obs));
 
         assert_eq!(workflow.orchestrate(S::Start).await.unwrap(), S::Done);
 
-        let events = observer.events();
+        let events = rec.labels();
         assert!(
             events.contains(&"state_enter:Start".to_string()),
             "{events:?}"
@@ -444,15 +409,15 @@ mod tests {
 
     #[tokio::test]
     async fn observer_fires_on_retry_and_failure() {
-        let observer = Arc::new(RecordingObserver::default());
+        let (obs, rec) = EventLog::new();
         let workflow = Workflow::bare()
             .register(S::Start, FailTask)
             .add_exit_state(S::Done)
-            .with_observer(observer.clone());
+            .with_observer(Arc::new(obs));
 
         assert!(workflow.orchestrate(S::Start).await.is_err());
 
-        let events = observer.events();
+        let events = rec.labels();
         assert!(
             events.contains(&"task_start:FailTask".to_string()),
             "{events:?}"
@@ -487,7 +452,7 @@ mod tests {
         let permit = breaker.try_acquire().expect("closed breaker admits");
         breaker.record_failure(permit);
 
-        let observer = Arc::new(RecordingObserver::default());
+        let (obs, rec) = EventLog::new();
         let workflow = Workflow::bare()
             .register(
                 S::Start,
@@ -496,13 +461,13 @@ mod tests {
                 },
             )
             .add_exit_state(S::Done)
-            .with_observer(observer.clone());
+            .with_observer(Arc::new(obs));
 
         let err = workflow.orchestrate(S::Start).await.unwrap_err();
         // The FSM wraps the failure with state context; the inner is CircuitOpen.
         assert!(matches!(err.inner(), CanoError::CircuitOpen(_)), "{err}");
 
-        let events = observer.events();
+        let events = rec.labels();
         assert!(
             events.contains(&"circuit_open:CircuitTask".to_string()),
             "{events:?}"
@@ -571,21 +536,12 @@ mod tests {
 
     #[test]
     fn on_workflow_timeout_can_be_overridden_to_record_event() {
-        use std::sync::{Arc, Mutex};
-        use std::time::Duration;
-        #[derive(Default)]
-        struct Recorder(Mutex<Vec<(Duration, Duration)>>);
-        impl WorkflowObserver for Recorder {
-            fn on_workflow_timeout(&self, elapsed: Duration, limit: Duration) {
-                self.0.lock().unwrap().push((elapsed, limit));
-            }
-        }
-        let obs = Arc::new(Recorder::default());
-        let dyn_obs: Arc<dyn WorkflowObserver> = obs.clone();
+        // `EventLog` is itself a custom override — proves the hook can be observed.
+        let (obs, rec) = EventLog::new();
+        let dyn_obs: Arc<dyn WorkflowObserver> = Arc::new(obs);
         dyn_obs.on_workflow_timeout(Duration::from_millis(150), Duration::from_millis(100));
-        let events = obs.0.lock().unwrap().clone();
         assert_eq!(
-            events,
+            rec.timeouts(),
             vec![(Duration::from_millis(150), Duration::from_millis(100))]
         );
     }