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//! To test a source that reads a work feed, script its deliveries,
67//! redeliveries, publish outcomes, refused feed calls and depth reads with
68//! [`scripted_feed`].
69//!
70//! For failure-path testing, script the sink
71//! ([`SinkScript::enqueue_for`]) and the deserializer
72//! ([`TestDeserializer::fail_on_prefix`]); for property tests, enable the
73//! `proptest` feature and use [`strategies`].
74//!
75//! To test one stage without a source, build its input with [`raw_payload`],
76//! [`record`] and [`test_ack`], and its config with [`component_config`].
77//! To assert on metrics, render them with [`render_metrics`] and read series
78//! back with [`metric_value`], [`metric_sum`] or [`metric_series`].
79
80mod deser;
81mod exposition;
82mod feed;
83mod fixtures;
84mod logs;
85mod run;
86mod sink;
87mod source;
88mod tls;
89
90#[cfg(feature = "proptest")]
91pub mod strategies;
92
93pub use deser::{EmitCollector, TestDeserializer};
94pub use exposition::{MetricSample, metric_series, metric_sum, metric_value, render_metrics};
95pub use feed::{FeedScript, ScriptedFeed, scripted_feed};
96pub use fixtures::{component_config, raw_payload, record, test_ack, unique_name};
97pub use logs::{LogCapture, capture_logs, show_logs};
98pub use run::{PipelineRun, wait_until};
99pub use sink::{
100    CaptureSink, CaptureWriter, CapturedWrite, ReplicaTag, ScriptedResult, SinkScript, TestEncoder,
101    WriteOutcome, capture_sink, capture_writer, decode_rows,
102};
103pub use source::{
104    DrainBarrierProbe, MemoryBatch, MemoryLane, MemorySource, SourceHandle, memory_source,
105};
106
107/// Re-export of the framework's byte-passthrough deserializer.
108pub use spate_core::deser::BytesPassthrough;
109pub use tls::tls_alert_server;
spate_test/lib.rs

spate_test/
lib.rs
