spate_test/
lib.rs

1#![cfg_attr(docsrs, feature(doc_cfg))]
2// `loom` is never set in a published build, so its cfg carries no badge.
3#![cfg_attr(docsrs, doc(auto_cfg(hide(loom))))]
4//! Testing utilities for the Spate framework.
5//!
6//! Every mock pairs with a scripting/observation handle: [`MemorySource`] +
7//! [`SourceHandle`] for the source side, [`CaptureWriter`] + [`SinkScript`]
8//! for the sink side, plus [`TestDeserializer`], [`TestEncoder`], and
9//! [`EmitCollector`] to exercise the stages in between. All are deterministic
10//! and need no external infrastructure.
11//!
12//! # Example: source → deserialize → encode → acknowledge → commit
13//!
14//! The full life of a record, as a pipeline thread drives it:
15//!
16//! ```
17//! use spate_core::checkpoint::Checkpointer;
18//! use spate_core::deser::Deserializer;
19//! use spate_core::record::PartitionId;
20//! use spate_core::sink::RowEncoder;
21//! use spate_core::source::{LaneId, PayloadBatch, Source, SourceCtx, SourceEvent, SourceLane};
22//! use spate_test::{EmitCollector, TestDeserializer, TestEncoder, memory_source};
23//! use std::time::Duration;
24//!
25//! // Wire a real checkpointer to the in-memory source.
26//! let mut cp = Checkpointer::new();
27//! let (mut source, handle) = memory_source();
28//! source.open(SourceCtx::new(cp.handle())).unwrap();
29//!
30//! let p = PartitionId(0);
31//! cp.begin_epoch(&[p], 1);
32//! handle.assign_lanes(&[(LaneId(0), p)]);
33//! let SourceEvent::LanesAssigned(mut lanes) =
34//!     source.poll_events(Duration::from_millis(10)).unwrap()
35//! else {
36//!     panic!("expected an assignment");
37//! };
38//!
39//! // Produce, then poll → deserialize → encode.
40//! handle.push(p, Some(b"key"), b"hello");
41//! let mut batch = lanes[0].poll(512, Duration::from_millis(100)).unwrap().unwrap();
42//! let ack = batch.ack().clone();
43//! let mut out = EmitCollector::new();
44//! let mut deser = TestDeserializer::passthrough();
45//! while let Some(raw) = batch.next_payload() {
46//!     deser.deserialize(&raw, &ack, &mut out).unwrap();
47//! }
48//! let mut frame = bytes::BytesMut::new();
49//! for rec in &out.records {
50//!     TestEncoder.encode(rec, &mut frame).unwrap();
51//! }
52//! assert_eq!(spate_test::decode_rows(&frame), vec![b"hello".to_vec()]);
53//!
54//! // Dropping every record (and the batch) resolves the acknowledgement;
55//! // the watermark advances; the source records the commit.
56//! drop(out);
57//! drop(batch);
58//! drop(ack);
59//! cp.drain();
60//! let watermarks = cp.take_watermarks();
61//! assert_eq!(watermarks, vec![(p, 1)]);
62//! source.commit(&watermarks).unwrap();
63//! assert_eq!(handle.last_committed(p), Some(1));
64//! ```
65//!
66//! For failure-path testing, script the sink
67//! ([`SinkScript::enqueue_for`]) and the deserializer
68//! ([`TestDeserializer::fail_on_prefix`]); for property tests, enable the
69//! `proptest` feature and use [`strategies`].
70//!
71//! To test one stage without a source, build its input with [`raw_payload`],
72//! [`record`] and [`test_ack`], and its config with [`component_config`].
73//! To assert on metrics, render them with [`render_metrics`] and read series
74//! back with [`metric_value`], [`metric_sum`] or [`metric_series`].
75
76mod coordination;
77mod deser;
78mod exposition;
79mod fixtures;
80mod logs;
81mod run;
82mod sink;
83mod source;
84mod tls;
85
86#[cfg(feature = "proptest")]
87pub mod strategies;
88
89#[expect(deprecated, reason = "v0.4.0 removes this with spate-coordination")]
90pub use coordination::{CoordinatorScript, ScriptedCoordinator, scripted_coordinator};
91pub use deser::{EmitCollector, TestDeserializer};
92pub use exposition::{MetricSample, metric_series, metric_sum, metric_value, render_metrics};
93pub use fixtures::{component_config, raw_payload, record, test_ack, unique_name};
94pub use logs::{LogCapture, capture_logs, show_logs};
95pub use run::{PipelineRun, wait_until};
96pub use sink::{
97    CaptureSink, CaptureWriter, CapturedWrite, ReplicaTag, ScriptedResult, SinkScript, TestEncoder,
98    WriteOutcome, capture_sink, capture_writer, decode_rows,
99};
100pub use source::{
101    DrainBarrierProbe, MemoryBatch, MemoryLane, MemorySource, SourceHandle, memory_source,
102};
103
104/// Re-export of the framework's byte-passthrough deserializer.
105pub use spate_core::deser::BytesPassthrough;
106pub use tls::tls_alert_server;
spate_test/lib.rs

spate_test/
lib.rs
