nifi-rust-client 0.12.0

Apache NiFi REST API client library
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

//! Regression tests for the wire-shape drift covered by
//! [`nifi_rust_client::FlexibleString`].
//!
//! NiFi 2.9.0+ returns `StatusSnapshotDTO.timestamp` as a JSON integer
//! (Unix milliseconds) even though the OpenAPI spec declares it as
//! `string + format: date-time`. Earlier versions return a formatted
//! string. Both shapes must deserialize through the generated DTO.

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

use nifi_rust_client::NifiClientBuilder;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

#[allow(clippy::unwrap_used)]
async fn dynamic_client_on(
    mock: &MockServer,
    version: &str,
) -> nifi_rust_client::dynamic::DynamicClient {
    Mock::given(method("GET"))
        .and(path("/nifi-api/flow/about"))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "about": { "title": "NiFi", "version": version }
        })))
        .mount(mock)
        .await;

    let client = NifiClientBuilder::new(&mock.uri())
        .unwrap()
        .build()
        .unwrap();
    nifi_rust_client::dynamic::DynamicClient::from_client(client)
        .await
        .unwrap()
}

#[tokio::test]
async fn status_snapshot_timestamp_accepts_integer_wire_shape() {
    let mock = MockServer::start().await;
    let dynamic = dynamic_client_on(&mock, "2.9.0").await;

    let processor_id = "d443858d-019d-1000-ffff-ffffc5e26250";

    Mock::given(method("GET"))
        .and(path(format!(
            "/nifi-api/flow/processors/{processor_id}/status/history"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "statusHistory": {
                "generated": "04/27/2026 10:00:00 UTC",
                "componentDetails": { "Type": "GenerateFlowFile" },
                "fieldDescriptors": [],
                "aggregateSnapshots": [
                    { "timestamp": 1777382685086_i64, "statusMetrics": { "bytesRead": 0 } },
                    { "timestamp": 1777382715086_i64, "statusMetrics": { "bytesRead": 4096 } }
                ]
            }
        })))
        .mount(&mock)
        .await;

    let history = dynamic
        .flow()
        .get_processor_status_history(processor_id)
        .await
        .expect("status history must deserialize integer-shape timestamps");

    let snapshots = history
        .status_history
        .as_ref()
        .and_then(|s| s.aggregate_snapshots.as_deref())
        .unwrap_or_default();
    assert_eq!(snapshots.len(), 2);
    assert_eq!(
        snapshots[0].timestamp.as_deref(),
        Some("1777382685086"),
        "integer wire value renders as decimal string"
    );
    assert_eq!(snapshots[1].timestamp.as_deref(), Some("1777382715086"));
}

#[tokio::test]
async fn status_snapshot_timestamp_accepts_string_wire_shape() {
    let mock = MockServer::start().await;
    let dynamic = dynamic_client_on(&mock, "2.6.0").await;

    let processor_id = "abc";

    Mock::given(method("GET"))
        .and(path(format!(
            "/nifi-api/flow/processors/{processor_id}/status/history"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "statusHistory": {
                "aggregateSnapshots": [
                    { "timestamp": "04/27/2026 10:00:00 UTC", "statusMetrics": {} }
                ]
            }
        })))
        .mount(&mock)
        .await;

    let history = dynamic
        .flow()
        .get_processor_status_history(processor_id)
        .await
        .unwrap();

    let snapshots = history
        .status_history
        .as_ref()
        .and_then(|s| s.aggregate_snapshots.as_deref())
        .unwrap_or_default();
    assert_eq!(
        snapshots[0].timestamp.as_deref(),
        Some("04/27/2026 10:00:00 UTC")
    );
}