ferrokinesis 0.7.0

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
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
137
138
139
140
141
142
143
144
145
146
147
148
149

// Case count rationale: 50 cases — each iteration does multiple PutRecord/PutRecords
// calls plus GetRecords, so lower count keeps total runtime reasonable.
mod common;
use common::*;

use ferrokinesis::store::StoreOptions;
use num_bigint::BigUint;
use proptest::prelude::*;
use proptest::test_runner::{Config, TestRunner};
use serde_json::json;

/// P11: Sequential PutRecord calls produce strictly increasing sequence numbers.
#[test]
fn prop_sequential_puts_increasing_sequence_numbers() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let server = rt.block_on(TestServer::with_options(StoreOptions {
        create_stream_ms: 0,
        delete_stream_ms: 0,
        update_stream_ms: 0,
        shard_limit: 200,
        ..Default::default()
    }));

    let mut runner = TestRunner::new(Config {
        cases: 50,
        ..Config::default()
    });

    runner
        .run(&(2u32..=20), |record_count| {
            let stream_name = unique_stream_name("prop-seq");

            let seq_nums: Vec<String> = rt.block_on(async {
                server.create_stream(&stream_name, 1).await;

                let mut seqs = Vec::with_capacity(record_count as usize);
                for _ in 0..record_count {
                    let res = server
                        .request(
                            "PutRecord",
                            &json!({
                                "StreamName": stream_name,
                                "Data": "dGVzdA==",
                                "PartitionKey": "same-key",
                            }),
                        )
                        .await;
                    assert_eq!(res.status(), 200);
                    let body: serde_json::Value = res.json().await.unwrap();
                    seqs.push(body["SequenceNumber"].as_str().unwrap().to_string());
                }
                seqs
            });

            let parsed: Vec<BigUint> = seq_nums
                .iter()
                .map(|s| s.parse::<BigUint>().unwrap())
                .collect();

            for i in 0..parsed.len() - 1 {
                prop_assert!(
                    parsed[i] < parsed[i + 1],
                    "sequence number ordering violated at index {}: {} >= {}",
                    i,
                    parsed[i],
                    parsed[i + 1]
                );
            }

            Ok(())
        })
        .unwrap();
}

/// P12: Within a PutRecords batch, sequence numbers on the same shard are
/// strictly increasing.
#[test]
fn prop_batch_sequence_numbers_increasing() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let server = rt.block_on(TestServer::with_options(StoreOptions {
        create_stream_ms: 0,
        delete_stream_ms: 0,
        update_stream_ms: 0,
        shard_limit: 200,
        ..Default::default()
    }));

    let mut runner = TestRunner::new(Config {
        cases: 50,
        ..Config::default()
    });

    runner
        .run(&(2u32..=50), |batch_size| {
            let stream_name = unique_stream_name("prop-seq");

            let seq_nums: Vec<String> = rt.block_on(async {
                server.create_stream(&stream_name, 1).await;

                let records: Vec<serde_json::Value> = (0..batch_size)
                    .map(|_| {
                        json!({
                            "Data": "dGVzdA==",
                            "PartitionKey": "same-key",
                        })
                    })
                    .collect();

                let res = server
                    .request(
                        "PutRecords",
                        &json!({
                            "StreamName": stream_name,
                            "Records": records,
                        }),
                    )
                    .await;
                assert_eq!(res.status(), 200);
                let body: serde_json::Value = res.json().await.unwrap();

                body["Records"]
                    .as_array()
                    .unwrap()
                    .iter()
                    .map(|r| r["SequenceNumber"].as_str().unwrap().to_string())
                    .collect()
            });

            prop_assert_eq!(seq_nums.len(), batch_size as usize);

            let parsed: Vec<BigUint> = seq_nums
                .iter()
                .map(|s| s.parse::<BigUint>().unwrap())
                .collect();

            for i in 0..parsed.len() - 1 {
                prop_assert!(
                    parsed[i] < parsed[i + 1],
                    "batch sequence ordering violated at index {}: {} >= {}",
                    i,
                    parsed[i],
                    parsed[i + 1]
                );
            }

            Ok(())
        })
        .unwrap();
}