Testing

A node is rarely testable on its own: it consumes topics, polls services, and fires actions against dependency nodes that a unit test has no business standing up. peppy node sync therefore generates a test surface alongside the production bindings: typed mocks that play each dependency slot over the real wire, and a harness that boots your node in-process against them.

The tests this enables are not simulations of the messaging layer. The mocks publish real Cap’n Proto payloads through a real (ephemeral, per-test) Zenoh router, the node runs its real setup against its real generated bindings, and services and actions run on the same engines a production producer uses. What you save is everything else: no daemon, no stack, no dependency binaries, and no sleeps, because every step of the boot is gated on a deterministic readiness barrier rather than a timer.

What sync generates

After peppy node sync, two module trees sit next to the production bindings, regenerated on every sync:

• peppygen.mock (Python) / peppygen::mock (Rust) provides one module per dependency, pairing, and observer slot, named by link_id and grouped by namespace: mock.deps.<link_id>, mock.pairings.<link_id>, mock.observed.<link_id>. Each holds a Mock with one typed sub-surface per interface the slot carries.
• peppygen.fixtures (Python) / peppygen::fixtures (Rust) provides the harness (fixtures.harness), plus typed observation of the node’s own surface: fixtures.emitted_topics.<topic>, fixtures.exposed_services.<service>, fixtures.exposed_actions.<action>.

Neither tree can leak into a production build:

• In Rust, the modules compile only under peppygen’s testing feature, and sync enables that feature from the node’s dev-dependencies:

  [dev-dependencies]
  peppygen = { path = ".peppy/libs/peppygen", features = ["testing"] }

  cargo build never resolves dev-dependencies, so cargo test is the only place the test surfaces exist; production binaries cannot reach them even by accident. Sync writes and maintains this entry for you.

• In Python, the mock and fixtures packages ship inert: peppygen/__init__.py never imports them, so nothing loads unless a test file imports them explicitly.

The lib/main split

The harness boots your node by calling its setup function directly, so setup must be importable from test code. peppy node init scaffolds this from the start:

• Python already has it: setup lives in src/<node>/__main__.py, importable as from <node>.__main__ import setup.

• Rust gets a lib/main split: src/lib.rs holds pub async fn setup(...), and src/main.rs only delegates:

  fn main() -> peppygen::Result<()> {
      peppygen::NodeBuilder::new().run(my_node::setup)
  }

init also scaffolds a smoke test (tests/smoke.rs / tests/test_smoke.py) that boots the node through the harness and shuts it down cleanly, plus, for Python, a dev dependency group carrying pytest and pytest-asyncio with asyncio_mode = "auto".

A first test

The hello_receiver node consumes one topic, message_stream, from its hello_world_param dependency slot. Its node code:

Python

src/hello_receiver/__main__.py

import asyncio

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics.hello_world_param import message_stream


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(receive_messages(node_runner))]


async def receive_messages(node_runner: NodeRunner):
    # Subscribe once; the held subscription buffers every message in order, so
    # iterating never drops a message published between iterations.
    subscription = await message_stream.subscribe(node_runner)
    async for producer, message in subscription:
        print(f"Received from {producer.instance_id}: {message.message}")


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

Rust

src/lib.rs

use std::sync::Arc;

use peppygen::consumed_topics::hello_world_param::message_stream;
use peppygen::{NodeRunner, Parameters, Result};

/// The node's entry point. It lives in the library crate so tests can import
/// it: the generated test harness (`peppygen::fixtures::harness::Harness`)
/// boots it in-process, and `main.rs` delegates here for production runs.
pub async fn setup(_params: Parameters, node_runner: Arc<NodeRunner>) -> Result<()> {
    tokio::spawn(receive_messages(node_runner));
    Ok(())
}

async fn receive_messages(node_runner: Arc<NodeRunner>) {
    // Subscribe once; the held subscription buffers every message in order, so
    // looping on `next` never drops a message published between iterations.
    let mut subscription = match message_stream::subscribe(&node_runner).await {
        Ok(subscription) => subscription,
        Err(e) => {
            eprintln!("Failed to subscribe: {e}");
            return;
        }
    };

    loop {
        match subscription.next().await {
            Ok(Some((producer, message))) => {
                println!("Received from {}: {}", producer.instance_id, message.message)
            }
            Ok(None) => break,
            Err(e) => {
                eprintln!("Error receiving message: {e}");
                break;
            }
        }
    }
}

And its complete test, runnable with cargo test / uv run --group dev pytest with nothing else running on the machine:

Python

tests/test_hello.py

"""Boots `hello_receiver` in-process under the generated test harness and
drives one message through its consumed topic from the mocked producer: no
daemon, no real `hello_world_param` node, and no sleeps."""

from peppygen.consumed_topics.hello_world_param import message_stream
from peppygen.fixtures import harness

from hello_receiver.__main__ import setup


async def test_receives_a_message_from_the_mocked_producer():
    async with harness.start(setup) as h:
        # The first publish waits for the node's subscription to match before
        # delivering, so this is deterministic: a return means the node
        # received the message, and no subscriber within the readiness
        # timeout is a loud error instead of a silent drop.
        await h.mocks.deps.hello_world_param.message_stream.publish(
            message_stream.Message(message="hello from the mock")
        )
        # The matched subscription is opened by the task `setup` spawned,
        # and that task cannot run before `setup` returned: a returned
        # publish proves the node's setup finished.
        assert h.setup_finished()

Rust

tests/hello.rs

//! Boots `hello_receiver` in-process under the generated test harness and
//! drives one message through its consumed topic from the mocked producer:
//! no daemon, no real `hello_world_param` node, and no sleeps.

use peppygen::fixtures::harness::Harness;
use peppygen::mock::deps::hello_world_param::message_stream;

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn receives_a_message_from_the_mocked_producer() {
    let (harness, mocks) = Harness::start(hello_receiver::setup)
        .await
        .expect("the harness should boot the node");

    // The first publish waits for the node's subscription to match before
    // delivering, so this is deterministic: an `Ok` means the node received
    // the message, and no subscriber within the readiness timeout is a loud
    // error instead of a silent drop.
    mocks
        .deps
        .hello_world_param
        .message_stream
        .publish(&message_stream::Message {
            message: "hello from the mock".to_string(),
        })
        .await
        .expect("the node's subscription should receive the message");

    // The matched subscription is opened by the task `setup` spawned, and
    // that task cannot run before `setup` returned: a returned publish
    // proves the node's setup finished.
    assert!(
        harness.setup_finished(),
        "the matched publish proves setup returned"
    );

    harness
        .shutdown()
        .await
        .expect("the node should shut down cleanly");
}

Note what is absent: no hello_world_param process, no daemon, no launcher, and no sleep before the publish. The mock’s first publish waits for the node’s subscription to become visible on the router, so “the node is ready” is observed, never assumed.

Even the assertion is deterministic, not a hope: the publish can only return once the node’s subscription matched, that subscription is opened by the task setup spawns, and that task cannot run before setup returned. A returned publish therefore proves setup finished, which is exactly what setup_finished() asserts. (For a node whose setup itself loops forever, setup_finished() stays false by design; assert on the surfaces instead.)

The harness lifecycle

start takes your setup function, the exact shape NodeBuilder().run takes, and returns with the node running:

Python

from peppygen.fixtures import harness

from my_node.__main__ import setup

async def test_something():
    # As a context manager, shutdown runs even when the test body fails:
    async with harness.start(setup) as h:
        assert h.instance_id  # the generated unique id
        ...  # drive the node; assert on the mocks and h.emitted
    # Or awaited directly, with an explicit shutdown:
    h = await harness.start(setup)
    ...
    await h.shutdown()

Rust

use peppygen::fixtures::harness::Harness;

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_something() -> peppygen::Result<()> {
    let (mut harness, mut mocks) = Harness::start(my_node::setup).await?;
    assert!(harness.instance_id().starts_with("test-"));
    // ... drive the node through `mocks` / `harness.emitted`, asserting as you go ...
    harness.shutdown().await?;
    Ok(())
}

By the time start returns, all of the following happened, in order:

1. An ephemeral Zenoh router started on a free port, private to this harness. Every test gets its own router and shares no messaging state with any other test. Routers within one test binary are serialized by a process-wide guard (running several peer meshes at once makes gossip discovery flaky), so parallel tests queue on start instead of contending.
2. One mock started per link (dependency, pairing, and observer slots alike; a zero_or_more slot starts none by default, see Configuration), each on its own messaging session, with every publisher, service queryable, and action engine declared up front. In Rust the started mocks come back as the second element of the start tuple (mocks.deps.<link_id>, mocks.pairings.<link_id>, mocks.observed.<link_id>); in Python they hang off the harness as h.mocks.deps.<link_id> and so on.
3. The harness subscribed to every topic the node emits, before the node booted, so even the very first message your setup publishes is captured.
4. A standalone configuration was seeded so the node boots exactly as if a launcher had wired it: the router’s address, a unique instance id, your parameters, and, per slot, a bound producer, peer pin, or observed source pointing at the matching mock’s wire identity.
5. Readiness barriers passed: the harness’s subscriptions to the node’s emitted topics are visible from the node’s session, and every mock service and action answers a reachability probe. Only then does the node’s setup run, so a setup that immediately polls a dependency service cannot race the mock’s declaration.

shutdown tears down in lifecycle order: it cancels the node and awaits its convergence (a setup that returned an error propagates that error out of shutdown, and shutdown hooks run, bounded exactly as in production), then closes the observation session and the mocks, then stops the router. In Python the async with form calls it for you; in Rust, dropping the harness without calling shutdown still cancels and aborts the node, but skips the error propagation and the shutdown hooks, so tests should end with harness.shutdown().await.

That propagation is itself a plain assertion: start returns normally even when setup is doomed, because setup has only been spawned; the error surfaces out of shutdown instead.

Python

import pytest

from peppygen.fixtures import harness

async def broken_setup(_params, _node_runner):
    raise RuntimeError("hardware missing")

async def test_a_failing_setup_fails_through_shutdown():
    h = await harness.start(broken_setup)  # returns normally
    with pytest.raises(RuntimeError, match="hardware missing"):
        await h.shutdown()

Rust

use peppygen::fixtures::harness::Harness;

async fn broken_setup(
    _parameters: peppygen::Parameters,
    _node_runner: std::sync::Arc<peppygen::NodeRunner>,
) -> peppygen::Result<()> {
    Err(peppygen::Error::Io(std::io::Error::other("hardware missing")))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failing_setup_fails_through_shutdown() {
    let (harness, _mocks) = Harness::start(broken_setup).await.unwrap(); // normal
    let error = harness.shutdown().await.expect_err("the setup error surfaces here");
    assert!(error.to_string().contains("hardware missing"));
}

The harness also exposes the running node directly: node_runner() (h.node_runner) for calling its runtime surface in-process, instance_id() (h.instance_id) for the generated instance id, setup_finished() to ask whether setup has returned, and session() (h.session), the fixture caller/observer session the fixtures callers drive the node through.

Driving each surface

The sections below use one example node: it consumes a video_stream topic and an enable_camera service from a camera slot, a plan_motion action from a brain slot, participates in an arm pairing (emits joint_commands, consumes joint_states), observes the same pairing through an observed_arm observer slot, and exposes a status emitted topic, a ping service, and a move_arm action of its own.

Every fragment in the rest of this page is the body of a test with the harness running as the lifecycle starts it: h in Python (async with harness.start(setup) as h:), harness and mocks in Rust (let (mut harness, mut mocks) = Harness::start(setup).await?;). The asserts and ?s land where they would in that test.

Consumed topics

Each consumed topic gets a typed publisher on the slot’s mock. Messages are the same generated Message type the node’s subscription yields:

Python

from peppygen.consumed_topics.camera import video_stream

# Wait for the node's subscription alone; the publish below then skips
# its own lazy wait:
assert await h.mocks.deps.camera.video_stream.wait_for_subscriber(10.0)

await h.mocks.deps.camera.video_stream.publish(
    video_stream.Message(width=640, frame=b"\x01\x02")
)

Rust

use peppygen::mock::deps::camera::video_stream;
use std::time::Duration;

// Wait for the node's subscription alone; the publish below then skips
// its own lazy wait:
assert!(
    mocks
        .deps
        .camera
        .video_stream
        .wait_for_subscriber(Duration::from_secs(10))
        .await?
);

mocks
    .deps
    .camera
    .video_stream
    .publish(&video_stream::Message { width: 640, frame: vec![1, 2] })
    .await?;

The first publish lazily waits for the node’s subscription to match before delivering, so the first message is never lost to a subscribe race, and a node that never subscribes turns the publish into a loud error after the readiness timeout (10 seconds) rather than a silent drop. To separate “the node is subscribed” from “deliver now”, wait_for_subscriber(timeout) waits for the match alone and returns whether it happened.

Consumed services

Each consumed service gets a typed mock server. A background pump captures every inbound request, and you answer them in one of two modes:

Manual: next_request parks until the node polls, then hands you the decoded request and a one-shot responder:

Python

from peppygen.consumed_services.camera import enable_camera

request, responder = await h.mocks.deps.camera.enable_camera.next_request(10.0)
assert request.enable
await responder.respond(enable_camera.ResponseData(enabled=True))
# or: await responder.respond_error("hardware fault")

Rust

use peppygen::mock::deps::camera::enable_camera;
use std::time::Duration;

let (request, responder) = mocks
    .deps
    .camera
    .enable_camera
    .next_request(Duration::from_secs(10))
    .await?;
assert!(request.enable);
responder.respond(enable_camera::ResponseData::new(true)).await?;
// or: responder.respond_error("hardware fault").await?;

Scripted: enqueue_response queues typed responses served automatically, FIFO, to the next inbound requests. Only unscripted requests park for next_request, so a test that does not care about per-request choreography scripts the answers up front and asserts afterwards:

Python

h.mocks.deps.camera.enable_camera.enqueue_response(
    enable_camera.ResponseData(enabled=True)
)
# ... let the node run ...
requests = h.mocks.deps.camera.enable_camera.captured()
assert [r.enable for r in requests] == [True]

Rust

mocks
    .deps
    .camera
    .enable_camera
    .enqueue_response(enable_camera::ResponseData::new(true))?;
// ... let the node run ...
let requests = mocks.deps.camera.enable_camera.captured()?;
assert_eq!(requests.len(), 1);

captured() decodes every request received so far, scripted and manual alike, in arrival order, so both modes leave a complete audit trail. A mock torn down with parked requests, or with scripted responses the node never consumed, reports them loudly at teardown, so a call the test forgot to answer cannot pass silently.

Consumed actions

Each consumed action gets a mock server on the real action engine, so the goal lifecycle is exactly the production one. The test plays the producer’s side of it: receive the goal, admit it, feed it, finish it.

Python

from peppygen.consumed_actions.brain import plan_motion

pending = await h.mocks.deps.brain.plan_motion.next_goal(10.0)
assert pending.request.arm_id == 7

active = await pending.accept(plan_motion.GoalResponseData(accepted=True))
# or: await pending.reject("busy", plan_motion.GoalResponseData(accepted=False))

await active.publish_feedback(plan_motion.FeedbackMessage(progress=0.5))
await active.complete(plan_motion.ResultResponseData(success=True))

Rust

use peppygen::mock::deps::brain::plan_motion;
use std::time::Duration;

let pending = mocks
    .deps
    .brain
    .plan_motion
    .next_goal(Duration::from_secs(10))
    .await?;
assert_eq!(pending.request.arm_id, 7);

let active = pending.accept(plan_motion::GoalResponseData::new(true)).await?;
// or: pending.reject(Some("busy"), None).await?;

active.publish_feedback(&plan_motion::FeedbackMessage { progress: 0.5 }).await?;
active.complete(&plan_motion::ResultResponseData::new(true)).await?;

next_goal parks until the node fires a goal (bounded by the timeout) and returns it with the decoded request. accept returns the active goal handle; from there publish_feedback streams typed feedback, and complete / complete_cancelled terminate it. To test your node’s cancellation path, await cancel_signal() on the active goal (it resolves when the node requests cancellation), then answer with complete_cancelled. is_cancelled() polls the same state non-blocking.

Pairing peers

A pairing slot’s mock is a peer: it publishes the topics your node consumes on the slot and subscribes to the topics your node emits on it, pinned to your node’s identity exactly as a real paired peer would be. The harness seeds the peer pin, so from the node’s point of view the pairing is established at boot: paired() / wait_paired() resolve immediately to the mock’s identity, exposed as peer_info() on the mock module.

Python

from peppygen.mock.pairings import arm as mock_arm
from peppygen.paired_topics.arm import joint_states

# What the node's paired() resolves to under the harness.
assert mock_arm.peer_info().peer_link_id == mock_arm.PEER_LINK_ID

# The direction the node consumes: publish as the peer.
await h.mocks.pairings.arm.joint_states.publish(
    joint_states.Message(positions=[0.0, 0.5, 1.0])
)

# The direction the node emits: receive as the peer.
command = await h.mocks.pairings.arm.joint_commands.next()
assert command.max_velocity == 0.5

Rust

use peppygen::mock::pairings::arm as mock_arm;
use peppygen::paired_topics::arm::joint_states;

// What the node's paired() resolves to under the harness.
assert_eq!(mock_arm::peer_info().peer_link_id, mock_arm::PEER_LINK_ID);

// The direction the node consumes: publish as the peer.
mocks
    .pairings
    .arm
    .joint_states
    .publish(&joint_states::Message { positions: [0.0, 0.5, 1.0] })
    .await?;

// The direction the node emits: receive as the peer.
let command = mocks
    .pairings
    .arm
    .joint_commands
    .next()
    .await?
    .expect("pairing subscription should be open");
assert!((command.max_velocity - 0.5).abs() < f64::EPSILON);

Both directions carry the pairing’s deterministic delivery guarantees: the consuming direction’s first publish waits for the node’s pinned subscription, and the emitting direction’s subscription was opened before the node booted.

Observed sources

An observer slot’s mock is a source: it publishes the observed topics under the identity the harness seeded into the slot, so the node’s source() / sources() accessor and observation subscription see it as a live member.

Python

from peppygen.mock.observed import observed_arm as mock_arm
from peppygen.paired_topics.observed_arm import joint_states

# The node's source() resolves to the mock's seeded identity:
assert joint_states.source(h.node_runner) == mock_arm.source()

await h.mocks.observed.observed_arm.joint_states.publish(
    joint_states.Message(positions=[0.0, 0.5, 1.0])
)

Rust

use peppygen::mock::observed::observed_arm as mock_arm;
use peppygen::paired_topics::observed_arm::joint_states;

// The node's source() resolves to the mock's seeded identity:
assert_eq!(
    joint_states::source(harness.node_runner())?,
    mock_arm::source()
);

mocks
    .observed
    .observed_arm
    .joint_states
    .publish(&joint_states::Message { positions: [0.0, 0.5, 1.0] })
    .await?;

A multi-cardinality observer slot starts as many mock sources as the configuration asks for; they arrive as a list (h.mocks.observed.fleet_arms[0] / mocks.observed.fleet_arms[0]), each publishing under its own instance identity, in the same order the node’s sources() reports them.

Emitted topics

The harness subscribes to every topic the node emits before the node boots, and barriers on those subscriptions being visible, so the very first message your setup publishes is captured, not raced. Each emitted topic is a typed subscription on harness.emitted:

Python

status = await h.emitted.status.next()
assert status.outcome == "done"

Rust

let status = harness
    .emitted
    .status
    .next()
    .await?
    .expect("status subscription should be open");
assert_eq!(status.outcome, "done");

next awaits the node’s next message on the topic, in order, and returns None once the harness’s session closes.

Exposed services

The fixtures callers drive the node’s exposed surface from the outside, as a fresh consumer with its own wire identity, the same path a real consumer node would take. Every call gates on a reachability probe first, so a test cannot race the node’s own declaration of the service:

Python

from peppygen.fixtures.exposed_services import ping

# Poll the node's service once.
response = await ping.poll(h, ping.RequestData(value=21), 10.0)
assert response.doubled == 42

Rust

use peppygen::fixtures::exposed_services::ping;
use std::time::Duration;

// Poll the node's service once.
let response = ping::poll(
    &harness,
    &ping::RequestData { value: 21 },
    Duration::from_secs(10),
)
.await?;
assert_eq!(response.doubled, 42);

poll sends one request and awaits the node’s typed response; timeout bounds the reachability gate and the poll individually.

Exposed actions

The node’s exposed actions get the same treatment from the fixture session: send_goal gates on the reachability probe, then drives the goal through its full lifecycle:

Python

from peppygen.fixtures.exposed_actions import move_arm
import peppylib

goal = await move_arm.send_goal(
    h, move_arm.GoalRequestData(arm_id=7), peppylib.QoSProfile.Reliable, 10.0
)
assert goal.accepted
feedback = await goal.on_next_feedback()
assert feedback.progress > 0.0
result = await goal.get_result(10.0)
assert result.status == move_arm.ResultStatus.COMPLETED
assert result.data.success

Rust

use peppygen::fixtures::exposed_actions::move_arm;
use std::time::Duration;

let mut goal = move_arm::send_goal(
    &harness,
    &move_arm::GoalRequestData { arm_id: 7 },
    peppygen::QoSProfile::Reliable,
    Duration::from_secs(10),
)
.await?;
assert!(goal.accepted);
let feedback = goal.on_next_feedback().await?;
assert!(feedback.progress > 0.0);
let result = goal.get_result(Duration::from_secs(10)).await?;
assert!(matches!(
    result.outcome,
    move_arm::ResultOutcome::Completed(move_arm::ResultData { success: true })
));

The goal handle carries the decoded admission reply (accepted, the optional rejection reason, and the typed goal-response data), streams typed feedback with on_next_feedback, and finishes with get_result, whose outcome is one of completed, cancelled, abandoned, or expired, with the typed result payload on the first two. cancel_goal(timeout) requests cancellation, exercising the cancel_signal path in your node’s goal handler.

Producer loss, deterministically

Every mock owns one messaging session, so stop() is a whole-producer loss: every declaration and the session drop together, releasing the liveliness the node’s subscriptions and calls latch on. Live action goals are disarmed first so their teardown emits no clean close: the node observes exactly what it would observe if the producer process died mid-goal, and observes it deterministically rather than after an arbitrary timeout.

Python

pending = await h.mocks.deps.brain.plan_motion.next_goal(10.0)
active = await pending.accept(plan_motion.GoalResponseData(accepted=True))

# The producer dies with the goal still active. The node's
# on_next_feedback_message raises ConnectionError: not a clean
# end-of-stream, and not a hang.
await h.mocks.deps.brain.stop()

# The node's recovery branch ran and reported the loss on its status
# topic, so the branch is an ordinary assertion:
status = await h.emitted.status.next()
assert status.outcome == "producer-gone"

Rust

let pending = mocks
    .deps
    .brain
    .plan_motion
    .next_goal(Duration::from_secs(10))
    .await?;
let _active = pending.accept(plan_motion::GoalResponseData::new(true)).await?;

// The producer dies with the goal still active. The node's
// on_next_feedback_message returns
// Err(Error::ActionFeedbackProducerGone { .. }): not a clean close,
// and not a hang.
mocks.deps.brain.stop();

// The node's recovery branch ran and reported the loss on its status
// topic, so the branch is an ordinary assertion:
let status = harness
    .emitted
    .status
    .next()
    .await?
    .expect("status subscription should be open");
assert_eq!(status.outcome, "producer-gone");

On the node side the loss is typed: a feedback drain loop in Rust distinguishes Error::ActionFeedbackChannelClosed (the producer closed the stream cleanly) from Error::ActionFeedbackProducerGone (the producer instance disappeared without closing it); Python raises RuntimeError for the clean close and ConnectionError for the loss. Your node’s recovery branch for a dead dependency, usually untestable without killing processes, becomes an ordinary assertion.

A stopped mock stays stopped for the rest of the test; the bound set never rebinds, exactly as in production.

Configuration

start with no configuration uses the schema-default parameters, a unique generated instance id, one mock per one, zero_or_one, and one_or_more slot, and no mocks for zero_or_more slots. Every knob is an override:

Python

from peppygen.parameters import Parameters

async with harness.start(
    setup,
    parameters=Parameters(gain=2.5),  # typed override; None uses schema defaults
    instance_id="my_test_node",       # explicit instead of generated
    use_sim_time=True,                # boot in sim time; see "The clock"
    fleet_arms_instances=2,           # per multi-cardinality slot: mock count
    # per zero_or_one dependency or observer slot: boot the slot
    # empty, no mock started
    # <link_id>_vacant=True,
    node_dir="/path/to/node",         # see below
) as h:
    ...

Rust

use peppygen::fixtures::harness::{Config, Harness};

let (mut harness, mut mocks) = Harness::start_with(
    Config {
        parameters: Some(peppygen::Parameters { gain: 2.5 }),
        instance_id: Some("my_test_node".to_string()),
        use_sim_time: true, // boot in sim time; see "The clock"
        fleet_arms_instances: 2,
        // per zero_or_one dependency or observer slot: boot the
        // slot empty, no mock started
        // <link_id>_vacant: true,
        ..Config::default()
    },
    my_node::setup,
)
.await?;

• parameters is the node’s typed Parameters; leaving it unset hydrates the schema defaults exactly as a launch would (a required parameter without a default is then an error at boot, as in production).
• instance_id replaces the generated unique id (test-<pid>-<counter>). The node’s wire identity under the harness is ("standalone-core", <instance_id>), returned by harness.node_producer_ref().
• use_sim_time boots the node in sim time, exactly as a launcher’s framework: { use_sim_time: true } would, with the harness clock in sim mode: the test is the simulator, and no time exists until it calls harness.clock.tick(...). See The clock.
• <link_id>_vacant (per zero_or_one dependency or observer slot) boots the slot written vacant: no mock is started (the mock field is None), and the node’s bound_producer() / source() accessor answers None, the branch this cardinality exists for.
• <link_id>_instances (per multi-cardinality dependency or observer slot) sets how many mock instances to start, each under its own instance identity, delivered as a list. The default is 1 for one_or_more and 0 for zero_or_more.
• <link_id>_instance_ids (same slots) names those instances explicitly, overriding the count when set. Reach for it when the node classifies producers by instance name (a health monitor keying reports on left_arm/right_gripper identities, a recorder naming limbs after its sources), so the mocks wear identities the node’s classifier accepts.
• <link_id>_vacant (per optional: true pairing slot) boots the slot unpaired: the peer pin is never seeded, so the node’s paired() stays false, wait_paired() pends, and its publishes on the slot are no-ops. The peer mock still starts; its subscriptions simply stay silent (and publishing from it reaches nothing), so a test can assert the slot’s silence. Required pairing slots offer no such knob.
• node_dir (Python only) points at the directory holding the node’s peppy.json5 when neither of the automatic resolutions finds it: the harness first walks up from the current working directory to the nearest peppy.json5, then falls back to the absolute node path baked in at sync time. Running pytest from anywhere inside the node directory needs nothing; a runner with a foreign working directory on a moved checkout passes node_dir explicitly. The error on failure names all three sources. (Rust needs no equivalent: the config path is resolved from the generated crate’s own manifest directory at compile time.)

The clock

The harness serves the daemon’s clock surface on the standalone-core identity, through the same request handler the daemon runs, so everything in System clock works under it: peppylib::clock::synchronize / peppylib.clock.synchronize completes its NTP-style exchange (gated by the pre-setup readiness barrier, so a synchronize inside setup cannot race discovery), subscribe receives ticks, and peppygen::clock::init + now_ns read whichever source the boot resolved. The serving side is harness.clock, in one of two modes:

Wall mode (the default) mirrors a wall-mode daemon: service stamps and the 10 Hz tick stream come from the OS clock. set_offset_ns skews everything it serves by a signed offset, which is how a test scripts “the daemon’s clock disagrees with mine” and asserts the node’s offset handling without touching a host clock:

Python

harness.clock.set_offset_ns(3_600_000_000_000)  # daemon runs 1h ahead
sync = await synchronize(h.node_runner)
assert sync.offset_ns > 1_800_000_000_000

Rust

harness.clock.set_offset_ns(3_600_000_000_000)?; // daemon runs 1h ahead
let sync = peppylib::clock::synchronize(harness.node_runner(), None).await?;
assert!(sync.offset_ns > 1_800_000_000_000);

Sim mode (use_sim_time in the harness config) mirrors a sim-mode stack with the test playing the external simulator: the node boots with framework.use_sim_time resolved to true, and no time exists until the test drives it. Each tick lands atomically in the service’s answers (a synchronize issued right after tick returns can never observe the older instant) and is published on the clock topic for the node’s peppygen::clock subscription. Before the first tick, synchronize fails with “clock not ready” and now_ns reports not-ready, exactly as against a real sim-mode daemon whose simulator has not started publishing:

Python

async with harness.start(setup, use_sim_time=True) as h:
    await h.clock.tick(1_000_000_000)  # the node's now_ns() reads 1s
    ...
    await h.clock.tick(2_500_000_000)  # time advances only when the test says so

Rust

let (harness, mocks) = Harness::start_with(
    Config { use_sim_time: true, ..Config::default() },
    my_node::setup,
)
.await?;
harness.clock.tick(1_000_000_000).await?; // the node's now_ns() reads 1s
// ...
harness.clock.tick(2_500_000_000).await?; // time advances only when the test says so

The first tick waits until the node’s clock subscription is visible (the subscription peppygen::clock::init opens in sim mode), so it cannot be dropped by discovery; ticking a node that never reads sim time is surfaced as a loud error, not a silent drop. Mode and knob agree or fail fast: tick on a wall-mode clock and set_offset_ns on a sim-mode clock are errors.

Caveats

Caution

• cargo test and cargo build compile peppygen differently. The testing feature is on for test builds (via the dev-dependency) and off for production builds, so alternating between cargo build and cargo test recompiles peppygen each time. This is the cost of the guarantee that test surfaces cannot reach production binaries; it does not affect correctness.

Everything under peppygen::mock and peppygen::fixtures is regenerated by every peppy node sync, like the rest of peppygen: add an interface to your peppy.json5, re-sync, and its mock and fixtures appear with it. Never edit the generated files; your tests, like your node, live outside .peppy/.