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

//! E2E tests for HTTP Client to Console pipeline
//!
//! These tests verify real-world scenarios:
//! - API monitoring with data validation
//! - Multi-cycle polling stability
//!
//! Note: Since console output cannot be captured, these tests verify
//! the data flow by checking that the HTTP source successfully fetches
//! data from the mock server. The wiremock expectations serve as proof
//! that the pipeline is operating correctly.

#![cfg(feature = "http-client")]

mod common;

use common::{default_test_timeout, run_engine, wait_for_requests};
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

/// Scenario: API Monitoring
///
/// Simulates a typical API polling scenario where we fetch JSON data
/// from a REST endpoint. The wiremock server verifies that the pipeline
/// correctly triggers HTTP requests.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_api_monitoring_pipeline() {
    if common::skip_if_no_network() {
        return;
    }

    let mock_server = MockServer::start().await;

    // Simulate a real API response with nested data
    let api_response = serde_json::json!({
        "status": "healthy",
        "metrics": {
            "cpu_usage": 45.2,
            "memory_mb": 1024,
            "requests_per_sec": 150
        },
        "timestamp": "2026-01-07T12:00:00Z"
    });

    Mock::given(method("GET"))
        .and(path("/api/health"))
        .respond_with(ResponseTemplate::new(200).set_body_json(&api_response))
        .expect(1..) // Must receive at least 1 request
        .mount(&mock_server)
        .await;

    let yaml = format!(
        r#"
pipeline:
  sources:
    - id: health_monitor
      type: http_client
      config:
        url: "{}/api/health"
        interval: "100ms"
  transforms:
    - id: pass_through
      inputs: [health_monitor]
      outputs: [console_out]
  sinks:
    - id: console_out
      type: console
"#,
        mock_server.uri()
    );

    let (shutdown_tx, engine_handle) = run_engine(&yaml).await;

    // Wait for at least one poll cycle
    wait_for_requests(&mock_server, 1, default_test_timeout()).await;

    // Shutdown gracefully
    let _ = shutdown_tx.send(());
    let result = engine_handle.await.expect("Engine task panicked");

    assert!(result.is_ok(), "Engine should shutdown gracefully");
    // wiremock will verify at least 1 request was received on drop
}

/// Scenario: Multi-Cycle Polling Stability
///
/// Verifies that the pipeline remains stable over multiple polling cycles.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_multi_cycle_polling_stability() {
    if common::skip_if_no_network() {
        return;
    }

    let mock_server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/"))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"poll": "ok"})))
        .expect(3..) // Must receive at least 3 requests
        .mount(&mock_server)
        .await;

    let yaml = format!(
        r#"
pipeline:
  sources:
    - id: poller
      type: http_client
      config:
        url: "{}"
        interval: "100ms"
  transforms:
    - id: pass_through
      inputs: [poller]
      outputs: [console_out]
  sinks:
    - id: console_out
      type: console
"#,
        mock_server.uri()
    );

    let (shutdown_tx, engine_handle) = run_engine(&yaml).await;

    // Wait for at least 3 poll cycles
    wait_for_requests(&mock_server, 3, default_test_timeout()).await;

    let _ = shutdown_tx.send(());
    let result = engine_handle.await.expect("Engine task panicked");

    assert!(
        result.is_ok(),
        "Engine should remain stable over multiple cycles"
    );
    // wiremock verifies at least 3 requests were made
}