dora_message/integration_testing_format.rs
1//! Use these types for integration testing nodes.
2
3use std::{
4 collections::{BTreeMap, BTreeSet},
5 path::PathBuf,
6};
7
8use crate::{
9 config::Input,
10 descriptor::EnvValue,
11 id::{DataId, NodeId},
12 metadata::MetadataParameters,
13};
14
15/// Defines the input data and events for integration testing a node.
16///
17/// Most of the fields are similar to the fields defined in the [`Node`](crate::descriptor::Node)
18/// struct, which is used to define nodes in a dataflow YAML file.
19///
20/// For integration testing, the most important field is the [`events`](Self::events) field, which
21/// specifies the events that should be sent to the node during the test.
22#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
23pub struct IntegrationTestInput {
24 /// Unique node identifier. Must not contain `/` characters.
25 ///
26 /// Node IDs can be arbitrary strings with the following limitations:
27 ///
28 /// - They must not contain any `/` characters (slashes).
29 /// - We do not recommend using whitespace characters (e.g. spaces) in IDs
30 ///
31 /// Each node must have an ID field.
32 ///
33 /// ## Example
34 ///
35 /// ```yaml
36 /// nodes:
37 /// - id: camera_node
38 /// - id: some_other_node
39 /// ```
40 pub id: NodeId,
41
42 /// Human-readable node name for documentation.
43 ///
44 /// This optional field can be used to define a more descriptive name in addition to a short
45 /// [`id`](Self::id).
46 ///
47 /// ## Example
48 ///
49 /// ```yaml
50 /// nodes:
51 /// - id: camera_node
52 /// name: "Camera Input Handler"
53 pub name: Option<String>,
54
55 /// Detailed description of the node's functionality.
56 ///
57 /// ## Example
58 ///
59 /// ```yaml
60 /// nodes:
61 /// - id: camera_node
62 /// description: "Captures video frames from webcam"
63 /// ```
64 pub description: Option<String>,
65
66 /// Command-line arguments passed to the executable.
67 ///
68 /// The command-line arguments that should be passed to the executable/script specified in `path`.
69 /// The arguments should be separated by space.
70 /// This field is optional and defaults to an empty argument list.
71 ///
72 /// ## Example
73 /// ```yaml
74 /// nodes:
75 /// - id: example
76 /// path: example-node
77 /// args: -v --some-flag foo
78 /// ```
79 #[serde(default, skip_serializing_if = "Option::is_none")]
80 pub args: Option<String>,
81
82 /// Environment variables for node builds and execution.
83 ///
84 /// Key-value map of environment variables that should be set for both the
85 /// [`build`](crate::descriptor::Node::build) operation and the node execution (i.e. when the node is spawned
86 /// through [`path`](crate::descriptor::Node::path)).
87 ///
88 /// Supports strings, numbers, and booleans.
89 ///
90 /// ## Example
91 ///
92 /// ```yaml
93 /// nodes:
94 /// - id: example-node
95 /// path: path/to/node
96 /// env:
97 /// DEBUG: true
98 /// PORT: 8080
99 /// API_KEY: "secret-key"
100 /// ```
101 pub env: Option<BTreeMap<String, EnvValue>>,
102
103 /// Output data identifiers produced by this node.
104 ///
105 /// List of output identifiers that the node sends.
106 /// Must contain all `output_id` values that the node uses when sending output, e.g. through the
107 /// [`send_output`](https://docs.rs/dora-node-api/latest/dora_node_api/struct.DoraNode.html#method.send_output)
108 /// function.
109 ///
110 /// ## Example
111 ///
112 /// ```yaml
113 /// nodes:
114 /// - id: example-node
115 /// outputs:
116 /// - processed_image
117 /// - metadata
118 /// ```
119 #[serde(default)]
120 pub outputs: BTreeSet<DataId>,
121
122 /// Input data connections from other nodes.
123 ///
124 /// Defines the inputs that this node is subscribing to.
125 ///
126 /// The `inputs` field should be a key-value map of the following format:
127 ///
128 /// `input_id: source_node_id/source_node_output_id`
129 ///
130 /// The components are defined as follows:
131 ///
132 /// - `input_id` is the local identifier that should be used for this input.
133 ///
134 /// This will map to the `id` field of
135 /// [`Event::Input`](https://docs.rs/dora-node-api/latest/dora_node_api/enum.Event.html#variant.Input)
136 /// events sent to the node event loop.
137 /// - `source_node_id` should be the `id` field of the node that sends the output that we want
138 /// to subscribe to
139 /// - `source_node_output_id` should be the identifier of the output that that we want
140 /// to subscribe to
141 ///
142 /// ## Example
143 ///
144 /// ```yaml
145 /// nodes:
146 /// - id: example-node
147 /// outputs:
148 /// - one
149 /// - two
150 /// - id: receiver
151 /// inputs:
152 /// my_input: example-node/two
153 /// ```
154 #[serde(default)]
155 pub inputs: BTreeMap<DataId, Input>,
156
157 /// Redirect stdout/stderr to a data output.
158 ///
159 /// This field can be used to send all stdout and stderr output of the node as a Dora output.
160 /// Each output line is sent as a separate message.
161 ///
162 ///
163 /// ## Example
164 ///
165 /// ```yaml
166 /// nodes:
167 /// - id: example
168 /// send_stdout_as: stdout_output
169 /// - id: logger
170 /// inputs:
171 /// example_output: example/stdout_output
172 /// ```
173 #[serde(skip_serializing_if = "Option::is_none")]
174 pub send_stdout_as: Option<String>,
175
176 /// List of incoming events for the integration test.
177 ///
178 /// The node event stream will yield these events during the test. Once the list is exhausted,
179 /// the event stream will close itself.
180 pub events: Vec<TimedIncomingEvent>,
181
182 /// Status of the recording that produced this file. Emitted by the
183 /// `DORA_WRITE_EVENTS_TO` recorder; `None` for files produced by
184 /// older versions of dora (before #1857) or hand-authored fixtures.
185 ///
186 /// Consumers should refuse to replay (or at minimum warn loudly)
187 /// when this is `Some(RecordingStatus::Poisoned { .. })`, because
188 /// the `events` array is incomplete relative to the original run.
189 ///
190 /// Boxed so the (large) `RecordingStatus::Poisoned` variant doesn't
191 /// inflate `IntegrationTestInput`'s size for the common-case
192 /// `Clean`-or-`None` recording. This also avoids
193 /// `large_enum_variant` lints on `TestingInput::Input(...)` (32
194 /// call sites; widening that enum would be intrusive).
195 #[serde(default, skip_serializing_if = "Option::is_none")]
196 pub recording_status: Option<Box<RecordingStatus>>,
197}
198
199/// State of a `DORA_WRITE_EVENTS_TO` recording (#1857).
200///
201/// `Clean` means the recorder observed every event it was asked to
202/// record. `Poisoned` means at least one `record_event()` call failed
203/// (Arrow→JSON conversion error, file I/O error, etc.) — the `events`
204/// array is missing at least one event, and replay results will not
205/// match the original run.
206#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
207#[serde(tag = "state", rename_all = "lowercase")]
208pub enum RecordingStatus {
209 Clean,
210 Poisoned {
211 /// Index in the `events` array where the first missed event
212 /// would have been. Equals the number of events that WERE
213 /// successfully recorded before the first failure.
214 first_failure_event_index: usize,
215 /// Seconds since the EventStream's start timestamp at the
216 /// moment of the first failure.
217 first_failure_time_offset_secs: f64,
218 /// `format!("{:?}", err)` of the first `record_event()` error.
219 first_failure_error: String,
220 /// Count of subsequent failures after the first one.
221 additional_failures: u64,
222 },
223}
224
225impl IntegrationTestInput {
226 pub fn new(id: NodeId, events: Vec<TimedIncomingEvent>) -> Self {
227 Self {
228 id,
229 name: None,
230 description: None,
231 args: None,
232 env: None,
233 outputs: BTreeSet::new(),
234 inputs: BTreeMap::new(),
235 send_stdout_as: None,
236 events,
237 recording_status: None,
238 }
239 }
240}
241
242/// An incoming event with a time offset.
243#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
244pub struct TimedIncomingEvent {
245 /// The time offset in seconds from the start of the node.
246 pub time_offset_secs: f64,
247 /// The incoming event.
248 #[serde(flatten)]
249 pub event: IncomingEvent,
250}
251
252/// An event that is sent to a node during an integration test.
253///
254/// This struct is very similar to the `Event` enum used during normal node operation.
255#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
256#[serde(tag = "type")]
257pub enum IncomingEvent {
258 Stop,
259 Input {
260 id: DataId,
261 metadata: Option<MetadataParameters>,
262 #[serde(flatten)]
263 data: Option<Box<InputData>>,
264 },
265 InputClosed {
266 id: DataId,
267 },
268 AllInputsClosed,
269}
270
271/// Represents the data of an incoming input event for integration testing.
272#[derive(Debug, PartialEq, Eq, Clone, serde::Deserialize, serde::Serialize)]
273#[serde(untagged)]
274pub enum InputData {
275 /// Converts the given JSON object to the closest Arrow representation.
276 ///
277 /// An optional data type can be provided to guide the conversion.
278 JsonObject {
279 /// The input data as JSON.
280 ///
281 /// This can be a JSON array, object, string, number, boolean, etc. Dora automatically
282 /// converts the JSON to an Apache Arrow array, wrapping the data if needed (e.g. wrap
283 /// bare integers into an array because Arrow requires all data to be in array form).
284 data: serde_json::Value,
285 /// Specifies the arrow `DataType` of the `data` field.
286 ///
287 /// This field is optional. If not set, Dora will try to infer the data type automatically.
288 ///
289 /// Use this field if the exact data type is important (e.g. to distinguish between
290 /// different integer sizes).
291 data_type: Option<serde_json::Value>,
292 },
293 /// Load data from an Arrow IPC file.
294 ///
295 /// The data must be in the
296 /// [Arrow IPC file format](https://arrow.apache.org/docs/python/ipc.html#writing-and-reading-random-access-files)
297 ArrowFile {
298 /// The path to the Arrow IPC file.
299 path: PathBuf,
300 /// The optional batch index to read from the file.
301 ///
302 /// Arrow IPC files can contain multiple record batches. Only one batch is read per input
303 /// event. This field specifies which batch to read. Defaults to `0`.
304 #[serde(default)]
305 batch_index: usize,
306 /// Optional column name to read from the record batch.
307 ///
308 /// If not set, the entire record batch is read and converted to an Arrow `StructArray`.
309 column: Option<String>,
310 },
311}
312
313#[cfg(test)]
314mod tests {
315 use super::*;
316
317 /// Pre-#1857 recordings and hand-authored fixtures have no
318 /// `recording_status` field. They must continue to deserialize
319 /// cleanly with `recording_status: None`, otherwise the new
320 /// schema breaks the entire existing fixture corpus.
321 #[test]
322 fn integration_test_input_without_recording_status_deserializes() {
323 let json = r#"{
324 "id": "test-node",
325 "events": []
326 }"#;
327 let v: IntegrationTestInput = serde_json::from_str(json).expect("deserialize legacy form");
328 assert!(v.recording_status.is_none());
329 }
330
331 /// Clean recordings emit `{"recording_status": {"state": "clean"}}`.
332 /// Deserialization should land on `Some(RecordingStatus::Clean)`.
333 #[test]
334 fn integration_test_input_with_clean_recording_status_deserializes() {
335 let json = r#"{
336 "id": "test-node",
337 "events": [],
338 "recording_status": { "state": "clean" }
339 }"#;
340 let v: IntegrationTestInput = serde_json::from_str(json).expect("deserialize clean form");
341 assert_eq!(v.recording_status.as_deref(), Some(&RecordingStatus::Clean));
342 }
343
344 /// Poisoned recordings carry the failure details; the deserialized
345 /// enum variant must preserve every field. The exact JSON shape
346 /// here is what `event_stream::WriteEventsTo::write_out` produces
347 /// in dora-node-api — pinning both sides of the wire contract.
348 #[test]
349 fn integration_test_input_with_poisoned_recording_status_deserializes() {
350 let json = r#"{
351 "id": "test-node",
352 "events": [],
353 "recording_status": {
354 "state": "poisoned",
355 "first_failure_event_index": 7,
356 "first_failure_time_offset_secs": 1.25,
357 "first_failure_error": "Arrow conversion failed: bad type",
358 "additional_failures": 3
359 }
360 }"#;
361 let v: IntegrationTestInput =
362 serde_json::from_str(json).expect("deserialize poisoned form");
363 match v.recording_status.map(|b| *b) {
364 Some(RecordingStatus::Poisoned {
365 first_failure_event_index,
366 first_failure_time_offset_secs,
367 first_failure_error,
368 additional_failures,
369 }) => {
370 assert_eq!(first_failure_event_index, 7);
371 assert_eq!(first_failure_time_offset_secs, 1.25);
372 assert_eq!(first_failure_error, "Arrow conversion failed: bad type");
373 assert_eq!(additional_failures, 3);
374 }
375 other => panic!("expected Poisoned, got {other:?}"),
376 }
377 }
378}