pipeflow 0.0.4

A lightweight, configuration-driven data pipeline framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253

//! E2E tests for system sources (DLQ, Event, Audit)

#![cfg(feature = "file")]

mod common;

use std::fs;
use std::time::Duration;

use pipeflow::Message;
use pipeflow::common::types::{Audit, Event};

#[tokio::test]
async fn test_dlq_source_to_file() {
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let output_path = temp_dir.path().join("dlq_output.jsonl");

    let config_yaml = format!(
        r#"
pipeline:
  transforms:
    - id: dlq_to_file
      inputs: [source::system::dlq]
      outputs: [dlq_file]

  sinks:
    - id: dlq_file
      type: file
      config:
        path: "{}"
"#,
        output_path.display()
    );

    let harness = common::SystemSourceHarness::new(&config_yaml).await;

    // Send a message directly to DLQ (original message that failed)
    let failed_msg = Message::new("test_source", serde_json::json!({"data": "test_failed"}));

    harness
        .channels
        .dlq
        .send(failed_msg)
        .await
        .expect("Failed to send to DLQ");

    // Wait for file content
    let file_content = common::wait_for_file_content(&output_path, Duration::from_secs(5)).await;

    // Shutdown pipeline
    harness.shutdown(Duration::from_secs(5)).await;

    // Verify output file contains the message
    assert!(!file_content.is_empty(), "Output file should not be empty");

    // Parse the JSONL line
    let line = file_content
        .lines()
        .next()
        .expect("Should have at least one line");
    let parsed: serde_json::Value = serde_json::from_str(line).expect("Valid JSON");

    // Verify payload contains the original message
    // DLQ source wraps the received Message, so structure is:
    // { meta: ..., payload: { meta: ..., payload: { data: "test_failed" } } }
    let inner_msg = &parsed["payload"];
    assert!(
        inner_msg["meta"].is_object(),
        "Should contain original message meta"
    );
    assert_eq!(
        inner_msg["payload"]["data"], "test_failed",
        "Should contain original message payload"
    );
}

#[tokio::test]
async fn test_event_source_processing() {
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let output_path = temp_dir.path().join("event_output.jsonl");

    let config_yaml = format!(
        r#"
pipeline:
  transforms:
    - id: event_to_file
      inputs: [source::system::event]
      outputs: [event_file]

  sinks:
    - id: event_file
      type: file
      config:
        path: "{}"
"#,
        output_path.display()
    );

    let harness = common::SystemSourceHarness::new(&config_yaml).await;

    // Send an event
    let event = Event::new("pipeline_started", serde_json::json!({"version": "1.0.0"}))
        .with_label("env", "test");

    harness
        .channels
        .event
        .send(event)
        .await
        .expect("Failed to send event");

    let file_content = common::wait_for_file_content(&output_path, Duration::from_secs(5)).await;

    harness.shutdown(Duration::from_secs(5)).await;

    // Verify output
    let line = file_content
        .lines()
        .next()
        .expect("Should have at least one line");
    let parsed: serde_json::Value = serde_json::from_str(line).expect("Valid JSON");

    let payload = &parsed["payload"];
    assert_eq!(payload["name"], "pipeline_started");
    assert!(payload["labels"]["env"].as_str().is_some());
}

#[tokio::test]
async fn test_audit_source_processing() {
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let output_path = temp_dir.path().join("audit_output.jsonl");

    let config_yaml = format!(
        r#"
pipeline:
  transforms:
    - id: audit_to_file
      inputs: [source::system::audit]
      outputs: [audit_file]

  sinks:
    - id: audit_file
      type: file
      config:
        path: "{}"
"#,
        output_path.display()
    );

    let harness = common::SystemSourceHarness::new(&config_yaml).await;

    // Send an audit record
    let audit = Audit::new("test_sink", "sink", "write")
        .with_message_id("msg-123")
        .with_label("batch", "1")
        .success(150);

    harness
        .channels
        .audit
        .send(audit)
        .await
        .expect("Failed to send audit");

    let file_content = common::wait_for_file_content(&output_path, Duration::from_secs(5)).await;

    harness.shutdown(Duration::from_secs(5)).await;

    // Verify output
    let line = file_content
        .lines()
        .next()
        .expect("Should have at least one line");
    let parsed: serde_json::Value = serde_json::from_str(line).expect("Valid JSON");

    let payload = &parsed["payload"];
    assert_eq!(payload["node_id"], "test_sink");
    assert_eq!(payload["node_type"], "sink");
    assert_eq!(payload["operation"], "write");
    assert_eq!(payload["status"], "success");
    assert_eq!(payload["duration_ms"], 150);
    assert_eq!(payload["message_id"], "msg-123");
}

#[tokio::test]
async fn test_all_system_sources_together() {
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let dlq_path = temp_dir.path().join("dlq.jsonl");
    let event_path = temp_dir.path().join("events.jsonl");
    let audit_path = temp_dir.path().join("audits.jsonl");

    let config_yaml = format!(
        r#"
pipeline:
  transforms:
    - id: dlq_to_file
      inputs: [source::system::dlq]
      outputs: [dlq_file]
    - id: event_to_file
      inputs: [source::system::event]
      outputs: [event_file]
    - id: audit_to_file
      inputs: [source::system::audit]
      outputs: [audit_file]

  sinks:
    - id: dlq_file
      type: file
      config:
        path: "{}"
    - id: event_file
      type: file
      config:
        path: "{}"
    - id: audit_file
      type: file
      config:
        path: "{}"
"#,
        dlq_path.display(),
        event_path.display(),
        audit_path.display()
    );

    let harness = common::SystemSourceHarness::new(&config_yaml).await;

    // Send one message to each channel
    let failed_msg = Message::new("src", serde_json::json!({"failed": true}));
    let event = Event::new("test", serde_json::json!({}));
    let audit = Audit::new("test_node", "transform", "process").success(10);

    harness.channels.dlq.send(failed_msg).await.unwrap();
    harness.channels.event.send(event).await.unwrap();
    harness.channels.audit.send(audit).await.unwrap();

    // Wait for all files to be populated
    common::wait_for_file_content(&dlq_path, Duration::from_secs(5)).await;
    common::wait_for_file_content(&event_path, Duration::from_secs(5)).await;
    common::wait_for_file_content(&audit_path, Duration::from_secs(5)).await;

    harness.shutdown(Duration::from_secs(5)).await;

    // Verify all output files exist and have content
    for (name, path) in [
        ("DLQ", &dlq_path),
        ("Event", &event_path),
        ("Audit", &audit_path),
    ] {
        assert!(path.exists(), "{} output file should exist", name);
        let content = fs::read_to_string(path).unwrap_or_else(|_| panic!("Read {} file", name));
        assert!(!content.is_empty(), "{} file should not be empty", name);
    }
}