dms-cdc-operator 0.1.26

The dms-cdc-operator is a Rust-based utility for comparing the state of a list of tables in an Amazon RDS database with data stored in Parquet files on Amazon S3, particularly useful for change data capture (CDC) scenarios
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

#[cfg(test)]
mod tests {
    use crate::s3::s3_operator::LoadParquetFilesPayload;
    use crate::s3::s3_operator::MockS3Operator;
    use crate::s3::s3_operator::S3Operator;
    use crate::s3::s3_operator::S3ParquetFile;
    use aws_sdk_s3::primitives::{DateTime, DateTimeFormat};

    #[tokio::test]
    async fn test_get_list_of_parquet_files_from_s3() {
        let mut s3_operator = MockS3Operator::new();

        let bucket_name = "bucket_name".to_string();
        let s3_prefix = "s3_prefix".to_string();
        let database_name = "database_name".to_string();
        let schema_name = "database_schema".to_string();
        let table_name = "table_name".to_string();
        let start_date = "2021-01-01T00:00:00Z".to_string();
        let stop_date = Some("2021-01-01T00:00:00Z".to_string());

        s3_operator
            .expect_get_list_of_parquet_files_from_s3()
            .returning(|_| {
                Ok(vec![S3ParquetFile::new(
                    "bucket_name/s3_prefix/file.parquet",
                )])
            });

        let load_parquet_files_payload = LoadParquetFilesPayload::DateAware {
            bucket_name,
            s3_prefix,
            database_name,
            schema_name,
            table_name,
            start_date,
            stop_date,
        };

        let files = s3_operator
            .get_list_of_parquet_files_from_s3(&load_parquet_files_payload)
            .await
            .unwrap();

        assert_eq!(files.len(), 1);
    }

    #[tokio::test]
    async fn test_get_files_from_s3_based_on_date() {
        let mut s3_operator = MockS3Operator::new();

        s3_operator
            .expect_get_files_from_s3_based_on_date()
            .returning(|_, _, _, _, _, _| Ok(vec![S3ParquetFile::new("file1")]));

        let bucket_name = "bucket_name".to_string();
        let table_name = "table_name".to_string();
        let start_date_path = "start_date_path".to_string();
        let prefix_path = "prefix_path".to_string();
        let start_date =
            DateTime::from_str("2021-01-01T00:00:00Z", DateTimeFormat::DateTimeWithOffset).unwrap();
        let stop_date = Some(
            DateTime::from_str("2021-01-01T00:00:00Z", DateTimeFormat::DateTimeWithOffset).unwrap(),
        );

        let files = s3_operator
            .get_files_from_s3_based_on_date(
                &bucket_name,
                &table_name,
                &start_date_path,
                &prefix_path,
                &start_date,
                stop_date,
            )
            .await
            .unwrap();

        assert_eq!(files.len(), 1);
    }
}