Crate spate_test 

Testing utilities for the Spate framework.

Every mock pairs with a scripting/observation handle (the tower-test philosophy): MemorySource + SourceHandle for the source side, CaptureWriter + SinkScript for the sink side, plus TestDeserializer, TestEncoder, and EmitCollector to exercise the stages in between — all deterministic, no external infrastructure.

Example: source → deserialize → encode → acknowledge → commit

The full life of a record, exactly as a pipeline thread drives it:

use spate_core::checkpoint::Checkpointer;
use spate_core::deser::Deserializer;
use spate_core::record::PartitionId;
use spate_core::sink::RowEncoder;
use spate_core::source::{LaneId, PayloadBatch, Source, SourceCtx, SourceEvent, SourceLane};
use spate_test::{EmitCollector, TestDeserializer, TestEncoder, memory_source};
use std::time::Duration;

// Wire a real checkpointer to the in-memory source.
let mut cp = Checkpointer::new();
let (mut source, handle) = memory_source();
source.open(SourceCtx::new(cp.handle())).unwrap();

let p = PartitionId(0);
cp.begin_epoch(&[p], 1);
handle.assign_lanes(&[(LaneId(0), p)]);
let SourceEvent::LanesAssigned(mut lanes) =
    source.poll_events(Duration::from_millis(10)).unwrap()
else {
    panic!("expected an assignment");
};

// Produce, then poll → deserialize → encode.
handle.push(p, Some(b"key"), b"hello");
let mut batch = lanes[0].poll(512, Duration::from_millis(100)).unwrap().unwrap();
let ack = batch.ack().clone();
let mut out = EmitCollector::new();
let mut deser = TestDeserializer::passthrough();
while let Some(raw) = batch.next_payload() {
    deser.deserialize(&raw, &ack, &mut out).unwrap();
}
let mut frame = bytes::BytesMut::new();
for rec in &out.records {
    TestEncoder.encode(rec, &mut frame).unwrap();
}
assert_eq!(spate_test::decode_rows(&frame), vec![b"hello".to_vec()]);

// Dropping every record (and the batch) resolves the acknowledgement;
// the watermark advances; the source records the commit.
drop(out);
drop(batch);
drop(ack);
cp.drain();
let watermarks = cp.take_watermarks();
assert_eq!(watermarks, vec![(p, 1)]);
source.commit(&watermarks).unwrap();
assert_eq!(handle.last_committed(p), Some(1));

For failure-path testing, script the sink (SinkScript::enqueue_for) and the deserializer (TestDeserializer::fail_on_prefix); for property tests, enable the proptest feature and use [strategies].

Structs

BytesPassthrough

Re-export of the framework’s byte-passthrough deserializer, handy next to the mocks here. Passthrough deserializer: yields the raw bytes as owned Vec<u8> records. Useful for byte-oriented pipelines and tests.

CaptureSink

A capturing sink bundle for whole-pipeline tests — see capture_sink.

CaptureWriter

A ShardWriter that records every attempt and resolves it according to the paired SinkScript (default: success). Unscripted writes succeed, so happy-path tests need no scripting at all.

CapturedWrite

One recorded ShardWriter::write_batch attempt.

CoordinatorScript

Scripting and observation handle for a ScriptedCoordinator.

DrainBarrierProbe

Observation side of a revocation’s DrainBarrier.

EmitCollector

An EmitRecord implementation that collects records into a Vec, optionally reporting Flow::Blocked once a capacity is reached (rejected records are counted, not stored) — for testing blocked-path handling.

MemoryBatch

One poll’s payloads, borrowing the lane’s buffer.

MemoryLane

Data-plane lane of a MemorySource: drains pushed payloads for one partition. Polls block (up to the timeout) while the lane is paused or its queue is empty, waking on push and resume.

MemorySource

In-memory Source driven entirely by its SourceHandle.

PipelineRun

A pipeline running on its own OS thread, with its exit result delivered over a channel so tests can wait with a bounded recv_timeout instead of polling JoinHandle::is_finished.

ReplicaTag

A replica endpoint tag — CaptureWriter’s Endpoint.

ScriptedCoordinator

A SplitCoordinator whose events and outcomes are scripted by the paired CoordinatorScript. Build both with scripted_coordinator.

SinkScript

Scripts outcomes for a CaptureWriter and reads back its captures. Cloneable; all clones address the same writer.

SourceHandle

Scripts and observes a MemorySource. Cloneable; all clones address the same source.

TestDeserializer

A scriptable Deserializer over owned byte-vector records.

TestEncoder

RowEncoder writing each row as u32 little-endian length + bytes. Decode captured frames with decode_rows.

WriteOutcome

One scripted write outcome: an optional delay, then a result.

Enums

ScriptedResult

How a scripted write resolves.

Functions

capture_sink

Build a ready-to-run capturing sink: a shards × replicas in-memory topology over a CaptureWriter, with its scripting handle.

capture_writer

Build a paired scriptable writer and its handle.

decode_rows

Decode a frame produced by TestEncoder back into rows.

memory_source

Build a paired in-memory source and its scripting handle.

scripted_coordinator

A scripted coordinator and its handle.

wait_until

Poll check until it returns true or timeout elapses; panics with what on timeout.