ferrokinesis 0.1.2

A local AWS Kinesis mock server for testing, written in 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

use ferrokinesis::store::StoreOptions;
use ferrokinesis::types::{ConsumerStatus, StreamStatus};

#[test]
fn stream_status_display_all_variants() {
    assert_eq!(StreamStatus::Active.to_string(), "ACTIVE");
    assert_eq!(StreamStatus::Creating.to_string(), "CREATING");
    assert_eq!(StreamStatus::Deleting.to_string(), "DELETING");
    assert_eq!(StreamStatus::Updating.to_string(), "UPDATING");
}

#[test]
fn consumer_status_display_all_variants() {
    assert_eq!(ConsumerStatus::Creating.to_string(), "CREATING");
    assert_eq!(ConsumerStatus::Deleting.to_string(), "DELETING");
    assert_eq!(ConsumerStatus::Active.to_string(), "ACTIVE");
}

#[test]
fn store_options_default_values() {
    let opts = StoreOptions::default();
    assert_eq!(opts.create_stream_ms, 500);
    assert_eq!(opts.delete_stream_ms, 500);
    assert_eq!(opts.update_stream_ms, 500);
    assert_eq!(opts.shard_limit, 10);
}

#[tokio::test]
async fn store_get_records_range_direct() {
    use ferrokinesis::store::Store;
    use ferrokinesis::types::StoredRecord;

    let store = Store::new(StoreOptions::default());
    store
        .put_record(
            "test-grr",
            "aabbccdd/seq001",
            StoredRecord {
                partition_key: "pk".to_string(),
                data: "AAAA".to_string(),
                approximate_arrival_timestamp: 1.0,
            },
        )
        .await;

    let records = store
        .get_records_range("test-grr", "aabbccdd/", "aabbccde/")
        .await;
    assert_eq!(records.len(), 1);

    let empty = store
        .get_records_range("test-grr", "ffffffff/", "ffffffff0/")
        .await;
    assert_eq!(empty.len(), 0);
}

#[tokio::test]
async fn store_with_streams_write_direct() {
    use ferrokinesis::store::Store;

    let store = Store::new(StoreOptions::default());
    let count = store
        .with_streams_write(|streams, _opts, _account, _region| streams.len())
        .await;
    assert_eq!(count, 0);
}

#[test]
fn error_server_error_with_some_args() {
    use ferrokinesis::error::KinesisErrorResponse;

    let err = KinesisErrorResponse::server_error(Some("CustomError"), Some("custom message"));
    assert_eq!(err.status_code, 500);
    assert_eq!(err.body.__type, "CustomError");
    assert_eq!(err.body.message.as_deref(), Some("custom message"));

    let _ = format!("{:?}", err.body);
    let _ = format!("{:?}", err);
    let cloned = err.clone();
    assert_eq!(cloned.status_code, 500);
}

#[test]
fn error_client_error_none_message() {
    use ferrokinesis::error::KinesisErrorResponse;

    let err = KinesisErrorResponse::client_error("SomeError", None);
    assert_eq!(err.status_code, 400);
    assert!(err.body.message.is_none());
    let _ = format!("{:?}", err);
}