warp10 3.0.0

Warp10 client for rust
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

use testcontainers::{
    GenericImage, ImageExt,
    core::{
        ContainerPort, IntoContainerPort, WaitFor,
        logs::consumer::logging_consumer::LoggingConsumer, wait::HttpWaitStrategy,
    },
    runners::AsyncRunner,
};
use warp10::{Client, Data, Label, Value};

async fn start_warp10() -> (testcontainers::ContainerAsync<GenericImage>, Client) {
    let _ = env_logger::builder().is_test(true).try_init();

    let container = GenericImage::new("warp10io/warp10", "3.5.0-ubuntu-ci")
        .with_exposed_port(8080.tcp())
        .with_wait_for(WaitFor::http(
            HttpWaitStrategy::new("/api/v0/check")
                .with_port(ContainerPort::Tcp(8080))
                .with_expected_status_code(200u16),
        ))
        .with_startup_timeout(std::time::Duration::from_secs(60))
        .with_log_consumer(
            LoggingConsumer::new()
                .with_stdout_level(log::Level::Info)
                .with_stderr_level(log::Level::Warn)
                .with_prefix("[warp10]"),
        )
        .start()
        .await
        .expect("failed to start warp10 container");

    let host = container.get_host().await.expect("failed to get host");
    let port = container
        .get_host_port_ipv4(8080.tcp())
        .await
        .expect("failed to get port");

    let url = format!("http://{}:{}", host, port);

    let client = Client::builder()
        .url(&url)
        .read_token("readTokenCI")
        .write_token("writeTokenCI")
        .build()
        .expect("failed to build client");

    (container, client)
}

#[tokio::test]
#[ignore]
async fn exec_simple_arithmetic() {
    let (_container, client) = start_warp10().await;

    let response = client.exec("1 2 +").await.unwrap();

    assert_eq!(response.body, serde_json::json!([3]));
    assert!(response.meta.elapsed.is_some());
    assert!(response.meta.ops.is_some());
}

#[tokio::test]
#[ignore]
async fn exec_error_returns_detail() {
    let (_container, client) = start_warp10().await;

    let err = client.exec("INVALID_FUNCTION_THAT_DOES_NOT_EXIST").await;

    match err {
        Err(warp10::Error::Exec(detail)) => {
            assert!(!detail.message.is_empty());
        }
        other => panic!("expected Exec error, got: {:?}", other),
    }
}

#[tokio::test]
#[ignore]
async fn write_and_exec_roundtrip() {
    let (_container, client) = start_warp10().await;
    let writer = client.get_writer();

    let data = vec![Data::new(
        time::OffsetDateTime::now_utc(),
        None,
        "test.integration.roundtrip".to_string(),
        vec![Label::new("source", "rust-test")],
        Value::Long(42),
    )];

    writer.post(data).await.unwrap();

    let script = format!(
        "[ '{}' 'test.integration.roundtrip' {{}} NOW -1 ] FETCH",
        "readTokenCI"
    );
    let response = client.exec(&script).await.unwrap();

    assert!(response.body.is_array());
    assert!(!response.body.as_array().unwrap().is_empty());
}

#[tokio::test]
#[ignore]
async fn find_series() {
    let (_container, client) = start_warp10().await;
    let writer = client.get_writer();

    let data = vec![Data::new(
        time::OffsetDateTime::now_utc(),
        None,
        "test.integration.find".to_string(),
        vec![Label::new("source", "rust-test")],
        Value::Long(1),
    )];

    writer.post(data).await.unwrap();

    let response = client
        .find("test.integration.find{source=rust-test}")
        .await
        .unwrap();

    assert!(!response.series().is_empty());
}

#[tokio::test]
#[ignore]
async fn exec_meta_headers_present() {
    let (_container, client) = start_warp10().await;

    let response = client.exec("1 2 + 3 4 +").await.unwrap();

    assert!(response.meta.elapsed.is_some());
    assert!(response.meta.ops.is_some());
}