flowfine 0.0.3

ScyllaDB and Cassandra migration tool
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
150
151
152
153
154
155
156
157
158
159
160
161

#[cfg(test)]
mod tests {
    use flowfine::config::VersionFormatting;
    use flowfine::config::VersionFormatting::Numeric;
    use flowfine::migration::parser::get_migrations;
    use flowfine::migration::version::MigrationVersionKey;
    use flowfine::migration::Migration;
    use flowfine::runner::{MigrationRunner, ScyllaMigrationRunner};
    use lazy_static::lazy_static;
    use rstest::{fixture, rstest};
    use scylla::{Session, SessionBuilder};
    use std::sync::Arc;

    lazy_static! {
        static ref KEYSPACE: &'static str = "flowfine";
        static ref PATH: &'static str = "./tests/data/int/numeric_migrations";
        static ref APPLY_MIGRATIONS_FAILED: &'static str = "Failed to apply migrations";
        static ref LOAD_MIGRATIONS_FAILED: &'static str = "Failed to load migrations files";
        static ref PARSE_MIGRATIONS_FAILED: &'static str = "Failed to parse migrations";
    }

    #[fixture]
    async fn session() -> Arc<Session> {
        let session = SessionBuilder::new()
            .known_node("[::1]:9042".to_string())
            .build()
            .await
            .unwrap();
        Arc::new(session)
    }

    #[fixture]
    async fn runner(#[future] session: Arc<Session>) -> ScyllaMigrationRunner {
        ScyllaMigrationRunner::new(session.await, *KEYSPACE)
    }

    async fn before_each(session: Arc<Session>) {
        let drop_query = format!("DROP KEYSPACE IF EXISTS {};", *KEYSPACE);
        let create_query = format!("CREATE KEYSPACE IF NOT EXISTS {} WITH replication = {{'class': 'SimpleStrategy', 'replication_factor' : 1}};", *KEYSPACE);
        session
            .query(drop_query, &[])
            .await
            .expect("Failed to drop keyspace");
        session
            .query(create_query, &[])
            .await
            .expect("Failed to create keyspace");
    }

    #[rstest]
    #[tokio::test]
    async fn test_apply_empty_migration(
        #[future] session: Arc<Session>,
        #[future] runner: ScyllaMigrationRunner,
    ) {
        let session = session.await;
        let runner = runner.await;
        before_each(session).await;

        // given
        let migrations = Vec::new();

        // when
        let applied_migrations = runner
            .run(migrations)
            .await
            .expect(*APPLY_MIGRATIONS_FAILED);

        // then
        assert!(applied_migrations.is_empty());
    }

    #[rstest]
    #[tokio::test]
    async fn test_apply_migrations(
        #[future] session: Arc<Session>,
        #[future] runner: ScyllaMigrationRunner,
    ) {
        let session = session.await;
        let runner = runner.await;
        before_each(session).await;

        // given
        let migrations = get_migrations(*PATH, &Numeric)
            .expect(*LOAD_MIGRATIONS_FAILED)
            .into_result()
            .expect(*PARSE_MIGRATIONS_FAILED);

        // when
        let applied_migrations = runner
            .run(migrations.clone())
            .await
            .expect(*APPLY_MIGRATIONS_FAILED);

        // then
        assert_eq!(migrations.len(), applied_migrations.len());
    }

    #[rstest]
    #[tokio::test]
    async fn test_ignore_already_applied_migrations(
        #[future] session: Arc<Session>,
        #[future] runner: ScyllaMigrationRunner,
    ) {
        let session = session.await;
        let runner = runner.await;
        before_each(session).await;

        // given
        let version_formatting = Numeric;
        let mut migrations = get_migrations(*PATH, &version_formatting)
            .expect(*LOAD_MIGRATIONS_FAILED)
            .into_result()
            .expect(*PARSE_MIGRATIONS_FAILED);
        let next_migration = new_migration(
            &version_formatting,
            "1.3",
            "delete_data",
            "DELETE FROM flowfine.test_table WHERE id = 1;",
        )
        .unwrap();

        // when
        runner
            .run(migrations.clone())
            .await
            .expect(*APPLY_MIGRATIONS_FAILED);

        migrations.push(next_migration.clone());
        let applied_migrations = runner
            .run(migrations.clone())
            .await
            .expect(*APPLY_MIGRATIONS_FAILED);

        // then
        assert_eq!(applied_migrations.len(), 1);
        assert_eq!(applied_migrations[0].version, next_migration.version);
        assert_eq!(applied_migrations[0].name, next_migration.name);
        assert_eq!(applied_migrations[0].filename, next_migration.filename);
        assert_eq!(applied_migrations[0].success, true);
    }

    fn new_migration(
        version_formatting: &VersionFormatting,
        version: &str,
        name: &str,
        content: &str,
    ) -> Option<Migration> {
        let version_key = MigrationVersionKey::new(version_formatting, version)?;
        let migration = Migration {
            filename: format!("V{}_{}.cql", version, name),
            version: version.to_string(),
            version_key,
            name: name.to_string(),
            content: content.to_string(),
            queries: vec![content.to_string()],
        };

        Some(migration)
    }
}