spate_test/
lib.rs

1//! Testing utilities for the Spate framework.
2//!
3//! Every mock pairs with a scripting/observation handle (the
4//! `tower-test` philosophy): [`MemorySource`] + [`SourceHandle`] for the
5//! source side, [`CaptureWriter`] + [`SinkScript`] for the sink side, plus
6//! [`TestDeserializer`], [`TestEncoder`], and [`EmitCollector`] to exercise
7//! the stages in between — all deterministic, no external infrastructure.
8//!
9//! # Example: source → deserialize → encode → acknowledge → commit
10//!
11//! The full life of a record, exactly as a pipeline thread drives it:
12//!
13//! ```
14//! use spate_core::checkpoint::Checkpointer;
15//! use spate_core::deser::Deserializer;
16//! use spate_core::record::PartitionId;
17//! use spate_core::sink::RowEncoder;
18//! use spate_core::source::{LaneId, PayloadBatch, Source, SourceCtx, SourceEvent, SourceLane};
19//! use spate_test::{EmitCollector, TestDeserializer, TestEncoder, memory_source};
20//! use std::time::Duration;
21//!
22//! // Wire a real checkpointer to the in-memory source.
23//! let mut cp = Checkpointer::new();
24//! let (mut source, handle) = memory_source();
25//! source.open(SourceCtx::new(cp.handle())).unwrap();
26//!
27//! let p = PartitionId(0);
28//! cp.begin_epoch(&[p], 1);
29//! handle.assign_lanes(&[(LaneId(0), p)]);
30//! let SourceEvent::LanesAssigned(mut lanes) =
31//!     source.poll_events(Duration::from_millis(10)).unwrap()
32//! else {
33//!     panic!("expected an assignment");
34//! };
35//!
36//! // Produce, then poll → deserialize → encode.
37//! handle.push(p, Some(b"key"), b"hello");
38//! let mut batch = lanes[0].poll(512, Duration::from_millis(100)).unwrap().unwrap();
39//! let ack = batch.ack().clone();
40//! let mut out = EmitCollector::new();
41//! let mut deser = TestDeserializer::passthrough();
42//! while let Some(raw) = batch.next_payload() {
43//!     deser.deserialize(&raw, &ack, &mut out).unwrap();
44//! }
45//! let mut frame = bytes::BytesMut::new();
46//! for rec in &out.records {
47//!     TestEncoder.encode(rec, &mut frame).unwrap();
48//! }
49//! assert_eq!(spate_test::decode_rows(&frame), vec![b"hello".to_vec()]);
50//!
51//! // Dropping every record (and the batch) resolves the acknowledgement;
52//! // the watermark advances; the source records the commit.
53//! drop(out);
54//! drop(batch);
55//! drop(ack);
56//! cp.drain();
57//! let watermarks = cp.take_watermarks();
58//! assert_eq!(watermarks, vec![(p, 1)]);
59//! source.commit(&watermarks).unwrap();
60//! assert_eq!(handle.last_committed(p), Some(1));
61//! ```
62//!
63//! For failure-path testing, script the sink
64//! ([`SinkScript::enqueue_for`]) and the deserializer
65//! ([`TestDeserializer::fail_on_prefix`]); for property tests, enable the
66//! `proptest` feature and use [`strategies`].
67
68mod coordination;
69mod deser;
70mod run;
71mod sink;
72mod source;
73
74#[cfg(feature = "proptest")]
75pub mod strategies;
76
77pub use coordination::{CoordinatorScript, ScriptedCoordinator, scripted_coordinator};
78pub use deser::{EmitCollector, TestDeserializer};
79pub use run::{PipelineRun, wait_until};
80pub use sink::{
81    CaptureSink, CaptureWriter, CapturedWrite, ReplicaTag, ScriptedResult, SinkScript, TestEncoder,
82    WriteOutcome, capture_sink, capture_writer, decode_rows,
83};
84pub use source::{
85    DrainBarrierProbe, MemoryBatch, MemoryLane, MemorySource, SourceHandle, memory_source,
86};
87
88/// Re-export of the framework's byte-passthrough deserializer, handy next
89/// to the mocks here.
90pub use spate_core::deser::BytesPassthrough;
spate_test/lib.rs

spate_test/
lib.rs
