Crate spate_test 

Testing utilities for the Spate framework.

Every mock pairs with a scripting/observation handle: MemorySource + SourceHandle for the source side, CaptureWriter + SinkScript for the sink side, plus TestDeserializer, TestEncoder, and EmitCollector to exercise the stages in between. All are deterministic and need no external infrastructure.

Example: source → deserialize → encode → acknowledge → commit

The full life of a record, 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.

To test one stage without a source, build its input with raw_payload, record and test_ack, and its config with component_config. To assert on metrics, render them with render_metrics and read series back with metric_value, metric_sum or metric_series.

Modules

strategiesproptest

Property-testing strategies for framework types (feature proptest).

Structs

BytesPassthrough

Re-export of the framework’s byte-passthrough deserializer. 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.

CapturedWrite

One recorded ShardWriter::write_batch attempt.

CoordinatorScriptDeprecated

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.

LogCapture

Everything the subscriber holding it has formatted.

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.

MetricSample

One sample line of a text exposition.

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.

ReplicaTag

A replica endpoint tag, CaptureWriter’s Endpoint.

ScriptedCoordinatorDeprecated

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_logs

Run f under a subscriber at level and return the lines it formatted.

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.

component_config

A component config tagged type_tag, with yaml as its body.

decode_rows

Decode a frame produced by TestEncoder back into rows.

memory_source

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

metric_series

Every sample named exactly name that carries all of labels, in the order they appear.

metric_sum

The sum of every sample metric_series selects, or None if it selects none.

metric_value

The value of the one sample metric_series selects, or None if it selects none.

raw_payload

A keyless payload over bytes at partition 0, offset 0, timestamp 0.

record

A record carrying payload, with zeroed metadata and a test_ack handle.

render_metrics

Run f under a recorder local to the calling thread, and return its Prometheus text exposition with the production histogram buckets.

scripted_coordinatorDeprecated

A scripted coordinator and its handle.

show_logs

Run f under a subscriber at level writing to the test harness.

test_ack

An acknowledgment handle whose resolution is discarded.

tls_alert_server

A TCP server on 127.0.0.1 that answers every TLS client with the fatal alert whose description is alert (RFC 8446 numbering, e.g. 40 for handshake_failure).

unique_name

prefix with a suffix no other call in this process returns.

wait_until

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