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

//! E2E tests for system sink defaults.

mod common;

use std::time::Duration;

use pipeflow::common::types::{Audit, Event};
use pipeflow::config::Config;
use pipeflow::engine::Engine;

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

    let config = Config::from_yaml(&format!(
        r#"
system:
  sinks:
    dir: "{}"
pipeline: {{}}
"#,
        temp_dir.path().display()
    ))
    .expect("Valid config");
    let mut engine = Engine::from_config(config).expect("Engine creation");
    engine.build().await.expect("Engine build");

    let channels = engine
        .system_channels()
        .expect("System channels should exist")
        .clone();

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let engine_handle = tokio::spawn(async move {
        let _ = engine
            .run_with_signal(async move {
                let _ = shutdown_rx.await;
            })
            .await;
    });

    let event = Event::new("test_event", serde_json::json!({"data": "hello"}));
    channels.event.send(event).await.expect("Send event");

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

    let _ = shutdown_tx.send(());
    let _ = tokio::time::timeout(Duration::from_secs(5), engine_handle).await;

    assert!(!file_content.is_empty(), "Event file should not be empty");
}

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

    let config = Config::from_yaml(&format!(
        r#"
system:
  sinks:
    dir: "{}"
pipeline: {{}}
"#,
        temp_dir.path().display()
    ))
    .expect("Valid config");
    let mut engine = Engine::from_config(config).expect("Engine creation");
    engine.build().await.expect("Engine build");

    let channels = engine
        .system_channels()
        .expect("System channels should exist")
        .clone();

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let engine_handle = tokio::spawn(async move {
        let _ = engine
            .run_with_signal(async move {
                let _ = shutdown_rx.await;
            })
            .await;
    });

    let audit = Audit::new("test_sink", "sink", "write")
        .with_message_id("msg-test")
        .success(50);
    channels.audit.send(audit).await.expect("Send audit");

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

    let _ = shutdown_tx.send(());
    let _ = tokio::time::timeout(Duration::from_secs(5), engine_handle).await;

    assert!(!file_content.is_empty(), "Audit file should not be empty");
}