Testing

Drop a pile of bespoke test fixtures for a one-line use cano::testing::*;.

Everything on this page is behind the testing feature gate (features = ["testing"]) — typically enabled under [dev-dependencies].

The cano::testing module wraps existing public surface (WorkflowObserver, CheckpointStore, Resources, MemoryStore, Task) into ready-made fixtures. It adds no new traits and changes no existing public items — it just saves you from re-writing the same recording observer and in-memory store in every test crate.

It is deliberately not in the prelude: import it explicitly with use cano::testing::*; so production code never picks it up by accident.

---

Enabling the feature

Add cano with the testing feature to your [dev-dependencies] so the helpers compile only for tests and never ship in your release build:

[dev-dependencies]
cano = { version = "0.15", features = ["testing"] }

The testing feature pulls in no extra dependencies and is zero-cost when off.

RecordingObserver

A WorkflowObserver that records every lifecycle event into an inspectable Vec<RecordedEvent>. Attach it, drive the workflow, then assert against the recorded path — or use the assert_path / assert_completed_with convenience checks.

use cano::prelude::*;
use cano::testing::*;
use std::sync::Arc;

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum S { Start, Done }

struct OkTask;
#[task]
impl Task<S> for OkTask {
    async fn run_bare(&self) -> Result<TaskResult<S>, CanoError> {
        Ok(TaskResult::Single(S::Done))
    }
}

#[tokio::test]
async fn recording_observer_captures_the_path() {
    let observer = Arc::new(RecordingObserver::new());
    let wf = Workflow::bare()
        .register(S::Start, OkTask)
        .add_exit_state(S::Done)
        .with_observer(observer.clone());

    assert_eq!(wf.orchestrate(S::Start, CancellationToken::disabled()).await.unwrap(), S::Done);

    // Assert the whole path, or inspect events directly.
    observer.assert_path(&["Start", "Done"]);
    observer.assert_completed_with("Done");
    assert!(observer.events().iter().any(|e| matches!(e, RecordedEvent::TaskSucceeded { .. })));
}

RecordedEvent mirrors the observer hooks (StateEntered, TaskStarted, TaskSucceeded, TaskFailed, Retry, CircuitOpen, Checkpoint, Resume). It is #[non_exhaustive] — match with a wildcard arm.

State-coverage assertions

Two helpers turn the recorded on_state_enter events into CI guardrails that catch dead states (a state you registered but nothing ever routes to) and routing regressions. Both return Result<(), Vec<String>> — Ok(()) when every expected state was entered, or Err with the missing state labels (in input order, deduplicated). Unlike assert_path, they return rather than panic, so you can inspect the gap (?-propagate, .expect(..), or .unwrap_err()).

• assert_all_states_entered(&[..]) — check an explicit list of states (compared by their Debug rendering).
• assert_registered_states_entered(&workflow) — check every state the workflow registered a handler for (via Workflow::registered_states). Exit-only states added with add_exit_state and no handler are not required.

use cano::prelude::*;
use cano::testing::RecordingObserver;
use std::sync::Arc;

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum S { Start, Worker, Done }

#[derive(Clone)]
struct Go(S);
#[task]
impl Task<S> for Go {
    async fn run_bare(&self) -> Result<TaskResult<S>, CanoError> {
        Ok(TaskResult::Single(self.0.clone()))
    }
}

#[tokio::test]
async fn every_registered_state_is_reached() {
    let observer = Arc::new(RecordingObserver::new());
    let wf = Workflow::bare()
        .register(S::Start, Go(S::Worker))
        .register(S::Worker, Go(S::Done))
        .add_exit_state(S::Done)
        .with_observer(observer.clone());
    wf.orchestrate(S::Start, CancellationToken::disabled()).await.unwrap();

    // Every registered handler was actually reached — no dead states.
    observer.assert_registered_states_entered(&wf).expect("no dead states");

    // Or assert an explicit set; Err lists what was never entered.
    observer.assert_all_states_entered(&[S::Start, S::Worker, S::Done]).unwrap();
}

InMemoryCheckpointStore

A process-local CheckpointStore for resume / recovery tests — no recovery feature and no on-disk file needed. Share one Arc<InMemoryCheckpointStore> across an orchestrate run and a later resume_from to exercise the crash-recovery path entirely in memory. It honors the full trait contract: duplicate (workflow_id, sequence) is rejected, load_run returns rows sorted by sequence, and clear is per-id.

use cano::prelude::*;
use cano::testing::*;
use std::sync::Arc;

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum S { Start, Done }

struct Go;
#[task]
impl Task<S> for Go {
    async fn run_bare(&self) -> Result<TaskResult<S>, CanoError> {
        Ok(TaskResult::Single(S::Done))
    }
}

#[tokio::test]
async fn checkpoints_run_in_memory() {
    let store = Arc::new(InMemoryCheckpointStore::new());
    let wf = Workflow::bare()
        .register(S::Start, Go)
        .add_exit_state(S::Done)
        .with_checkpoint_store(store.clone())
        .with_workflow_id("run-1");
    assert_eq!(wf.orchestrate(S::Start, CancellationToken::disabled()).await.unwrap(), S::Done);
}

TestResources

A small builder for the Resources a workflow needs. with_store drops in a fresh MemoryStore at a key; with_resource takes any Resource.

use cano::prelude::*;
use cano::testing::TestResources;

let resources = TestResources::new()
    .with_store("store")
    .build();
assert!(resources.get::<MemoryStore, _>("store").is_ok());

panic_on_attempt

A Task that panics on its first N attempts, for exercising panic safety: the engine converts a panicking task into a CanoError (message starting "panic: ") instead of unwinding the workflow loop.

Note: the engine wraps its catch-unwind around the whole retry loop, so a panic fails fast — panics are never retried (only a returned Err is). With panics >= 1 the run fails on the first attempt; the success branch is reached only with panics == 0.

use cano::prelude::*;
use cano::testing::panic_on_attempt;

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum S { Start, Done }

#[tokio::test]
async fn panicking_task_fails_fast() {
    let wf = Workflow::bare()
        .register(S::Start, panic_on_attempt(1, S::Done))
        .add_exit_state(S::Done);

    let err = wf.orchestrate(S::Start, CancellationToken::disabled()).await.unwrap_err();
    assert!(err.to_string().contains("panic"));
}

assert_compensation_ran

A saga assertion that compares the recorded compensation order against an expected sequence. Have each compensate push its identifier into a shared Vec (typically held in a resource), then pass that slice as actual. Compensations drain in reverse of completion order, so list expected in the order you expect them to undo.

use cano::testing::assert_compensation_ran;

// Charge ran last, so it compensates first; then Reserve.
let ran = vec!["charge".to_string(), "reserve".to_string()];
assert_compensation_ran(&ran, &["charge", "reserve"]);

---

Runnable end-to-end demo of every helper: cargo run --example testing_helpers --features testing.