16810179837

Committed 07 Aug 2025 04:27PM UTC coverage: 55.941% (+0.02%) from 55.922%

Build # 16810179837

Build Type

Pull #1845

github

Committed by

web-flow

Commit Message

Merge 02139528b into 24cadc364

Pull Request Pull Request #1845: cargo fmt

Run Details

20 of 24 new or added lines in 3 files covered. (83.33%)

13 existing lines in 2 files now uncovered.

3889 of 6952 relevant lines covered (55.94%)

60.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.79

/migration/src/bin/migrate_to.rs

use clap::{Parser, ValueEnum, builder::PossibleValuesParser, command};
use sea_orm_migration::{
    MigratorTrait, SchemaManager,
    sea_orm::{ConnectOptions, Database, DatabaseConnection, DbErr, EntityTrait, QueryOrder},
    seaql_migrations,
};
use std::cmp::Ordering;
use tracing::{error, info};
use tracing_subscriber::EnvFilter;

use migration::Migrator;

#[derive(Copy, Clone, ValueEnum, Debug)]
enum Direction {
    Up,
    Down,
}

/// Wraps the SeaORM migration library to provide additional features. Lets
/// you bring migrations up or down to a particular version. Allows dry-run
/// of migrations.
#[derive(Parser, Debug)]
#[command(about, version)]
struct Args {
    #[arg(value_enum)]
    direction: Direction,
    #[arg(value_parser = available_migrations())]
    target_version: String,
    #[arg(short, long)]
    dry_run: bool,
    #[arg(short = 'u', long, env = "DATABASE_URL")]
    database_url: String,
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let args = Args::parse();
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .with_ansi(false)
        .init();

    let db = Database::connect(ConnectOptions::new(args.database_url)).await?;
    check_database_is_compatible::<Migrator>(&db).await?;
    match args.direction {
        Direction::Up => migrate_up::<Migrator>(&db, args.dry_run, &args.target_version).await?,
        Direction::Down => {
            migrate_down::<Migrator>(&db, args.dry_run, &args.target_version).await?
        }
    }
    Ok(())
}

/// Checks that the database is compatible with the given migrator. Applied
/// migrations must be a subset of the migrations available by the MigratorTrait
/// for this CLI to be operated safely.
async fn check_database_is_compatible<M: MigratorTrait>(
    db: &DatabaseConnection,
) -> Result<(), Error> {
    // If the database is uninitialized, we can continue. The migrator will
    // initialize the database for us.
    if !has_migrations_table(db).await? {
        return Ok(());
    }

    let migrations: Vec<String> = M::migrations().iter().map(|m| m.name().into()).collect();
    let applied_migrations = applied_migrations(db).await?;

    let print_error = || {
        error!("expected migrations: {:?}", migrations);
        error!("present migrations: {:?}", applied_migrations);
    };
    for (i, applied) in applied_migrations.iter().enumerate() {
        match migrations.get(i) {
            Some(migration) if migration != applied => {
                print_error();
                return Err(Error::DbNotCompatible);
            }
            None => {
                print_error();
                return Err(Error::DbNotCompatible);
            }
            _ => {}
        }
    }
    Ok(())
}

async fn migrate_up<M: MigratorTrait>(
    db: &DatabaseConnection,
    dry_run: bool,
    target: &str,
) -> Result<(), Error> {
    let target_index =
        migration_index::<M>(target).ok_or(Error::MigrationNotFound(target.to_string()))?;

    let (migrations_range, num_migrations) = match latest_applied_migration(db).await {
        Ok(latest_migration) => {
            let latest_index =
                migration_index::<M>(&latest_migration).ok_or(Error::DbMigrationNotFound)?;
            match target_index.cmp(&latest_index) {
                Ordering::Less => return Err(Error::VersionTooOld(target.to_string())),
                Ordering::Equal => {
                    info!("no action taken, already at desired version");
                    return Ok(());
                }
                Ordering::Greater => (
                    (latest_index + 1)..=target_index,
                    target_index - latest_index,
                ),
            }
        }
        Err(Error::DbNotInitialized) => (
            0usize..=target_index,
            // The migration API takes "number of migrations to apply". If we have an
            // uninitialized database, and we want to apply the first migration (index 0),
            // then we still have to apply at least one migration.
            target_index + 1,
        ),
        Err(err) => return Err(err),
    };

    info!(
        "executing {num_migrations} up migration(s) to reach {target}: {:?}",
        Migrator::migrations()[migrations_range]
            .iter()
            .map(|m| m.name())
            .collect::<Vec<_>>()
    );
    if !dry_run {
        M::up(db, Some(u32::try_from(num_migrations)?))
            .await
            .map_err(Error::from)?
    }
    Ok(())
}

async fn migrate_down<M: MigratorTrait>(
    db: &DatabaseConnection,
    dry_run: bool,
    target: &str,
) -> Result<(), Error> {
    let latest_index = migration_index::<M>(&latest_applied_migration(db).await?)
        .ok_or(Error::DbMigrationNotFound)?;
    let target_index =
        migration_index::<M>(target).ok_or(Error::MigrationNotFound(target.to_string()))?;

    let num_migrations = match latest_index.cmp(&target_index) {
        Ordering::Less => return Err(Error::VersionTooNew(target.to_string())),
        Ordering::Equal => {
            info!("no action taken, already at desired version");
            return Ok(());
        }
        Ordering::Greater => latest_index - target_index,
    };

    info!(
        "executing {num_migrations} down migration(s) to reach {target}: {:?}",
        Migrator::migrations()[(target_index + 1)..=(latest_index)]
            .iter()
            .rev()
            .map(|m| m.name())
            .collect::<Vec<_>>()
    );
    if !dry_run {
        M::down(db, Some(u32::try_from(num_migrations)?))
            .await
            .map_err(Error::from)?
    }
    Ok(())
}

fn migration_index<M: MigratorTrait>(version: &str) -> Option<usize> {
    M::migrations().iter().position(|m| m.name() == version)
}

async fn has_migrations_table(db: &DatabaseConnection) -> Result<bool, Error> {
    Ok(SchemaManager::new(db).has_table("seaql_migrations").await?)
}

async fn latest_applied_migration(db: &DatabaseConnection) -> Result<String, Error> {
    if !has_migrations_table(db).await? {
        return Err(Error::DbNotInitialized);
    }
    Ok(seaql_migrations::Entity::find()
        .order_by_desc(seaql_migrations::Column::Version)
        .one(db)
        .await?
        // The migrations table exists, but no migrations have been applied.
        .ok_or(Error::DbNotInitialized)?
        .version)
}

async fn applied_migrations(db: &DatabaseConnection) -> Result<Vec<String>, Error> {
    Ok(seaql_migrations::Entity::find()
        .order_by_asc(seaql_migrations::Column::Version)
        .all(db)
        .await?
        .into_iter()
        .map(|m| m.version)
        .collect())
}

#[derive(Debug, thiserror::Error)]
enum Error {
    #[error("DB error: {0}")]
    Db(#[from] DbErr),
    #[error("DB is not initialized with migrations table")]
    DbNotInitialized,
    #[error("migration applied to DB is not found in available migrations")]
    DbMigrationNotFound,
    #[error("migration version {0} not found in avaliable migrations")]
    MigrationNotFound(String),
    #[error("migration version {0} is older than the latest applied migration")]
    VersionTooOld(String),
    #[error("migration version {0} is newer than the latest applied migration")]
    VersionTooNew(String),
    #[error("error calculating number of migrations, too many migrations?: {0}")]
    Overflow(#[from] std::num::TryFromIntError),
    #[error("applied migrations do not match migrations present in this tool")]
    DbNotCompatible,
}

fn available_migrations() -> PossibleValuesParser {
    PossibleValuesParser::new(
        // Leak memory to give migration names 'static lifetime, so clap can
        // use them.
        Migrator::migrations()
            .into_iter()
            .map(|m| Box::leak(Box::new(m.name().to_owned())) as &'static str)
            .collect::<Vec<_>>(),
    )
}

#[cfg(test)]
mod tests {
    use std::{sync::Once, time::SystemTime};

    use super::*;
    use sea_orm::{ActiveModelTrait, ActiveValue};
    use sea_orm_migration::prelude::*;

    macro_rules! test_migration {
        ($name:ident, $table_name:ident) => {
            #[allow(non_camel_case_types)]
            struct $name;

            impl MigrationName for $name {
                fn name(&self) -> &str {
                    stringify!($name)
                }
            }

            #[async_trait::async_trait]
            impl MigrationTrait for $name {
                async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
                    manager
                        .create_table(
                            Table::create()
                                .table($table_name::Table)
                                .col(ColumnDef::new($table_name::Id).uuid().primary_key())
                                .to_owned(),
                        )
                        .await?;
                    Ok(())
                }
                async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
                    manager
                        .drop_table(Table::drop().table($table_name::Table).to_owned())
                        .await
                }
            }

            #[derive(Iden)]
            enum $table_name {
                Table,
                Id,
            }
        };
    }
    test_migration!(m20230101_000000_test_migration_1, TestTable1);
    test_migration!(m20230201_000000_test_migration_2, TestTable2);
    test_migration!(m20230301_000000_test_migration_3, TestTable3);
    test_migration!(m20230401_000000_test_migration_4, TestTable4);
    test_migration!(m20230501_000000_test_migration_5, TestTable5);

    struct TestMigrator;

    #[async_trait::async_trait]
    impl MigratorTrait for TestMigrator {
        fn migrations() -> Vec<Box<dyn MigrationTrait>> {
            vec![
                Box::new(m20230101_000000_test_migration_1),
                Box::new(m20230201_000000_test_migration_2),
                Box::new(m20230301_000000_test_migration_3),
                Box::new(m20230401_000000_test_migration_4),
                Box::new(m20230501_000000_test_migration_5),
            ]
        }
    }

    fn all_migrations() -> Vec<&'static str> {
        vec![
            "m20230101_000000_test_migration_1",
            "m20230201_000000_test_migration_2",
            "m20230301_000000_test_migration_3",
            "m20230401_000000_test_migration_4",
            "m20230501_000000_test_migration_5",
        ]
    }

    async fn test_database() -> DatabaseConnection {
        Database::connect(ConnectOptions::new("sqlite::memory:".to_string()))
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn migrate_up_latest() {
        install_tracing_subscriber();
        let db = test_database().await;

        // To latest
        migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

        // Ensure no-op
        migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());
    }

    #[tokio::test]
    async fn migrate_up_works() {
        install_tracing_subscriber();
        let db = test_database().await;

        // To first
        migrate_up::<TestMigrator>(&db, false, "m20230101_000000_test_migration_1")
            .await
            .unwrap();
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            vec!["m20230101_000000_test_migration_1"]
        );

        // Dry run
        migrate_up::<TestMigrator>(&db, true, "m20230101_000000_test_migration_1")
            .await
            .unwrap();
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            vec!["m20230101_000000_test_migration_1"]
        );

        // To third
        migrate_up::<TestMigrator>(&db, false, "m20230301_000000_test_migration_3")
            .await
            .unwrap();
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );

        // To non-existent
        let result = migrate_up::<TestMigrator>(&db, false, "foobar").await;
        assert!(matches!(result, Err(Error::MigrationNotFound(_))));
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );

        // To old version
        let result =
            migrate_up::<TestMigrator>(&db, false, "m20230101_000000_test_migration_1").await;
        assert!(matches!(result, Err(Error::VersionTooOld(_))));
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );
    }

    #[tokio::test]
    async fn migrate_down_works() {
        install_tracing_subscriber();
        let db = test_database().await;
        let result =
            migrate_down::<TestMigrator>(&db, false, "m20230401_000000_test_migration_4").await;
        assert!(matches!(result, Err(Error::DbNotInitialized)));

        // Fail if DB not initialized

        // To latest
        migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

        // Ensure no-op
        migrate_down::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

        // Dry-run
        migrate_down::<TestMigrator>(&db, true, "m20230301_000000_test_migration_3")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

        // To third
        migrate_down::<TestMigrator>(&db, false, "m20230301_000000_test_migration_3")
            .await
            .unwrap();
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );

        // To newer version
        let result =
            migrate_down::<TestMigrator>(&db, false, "m20230401_000000_test_migration_4").await;
        assert!(matches!(result, Err(Error::VersionTooNew(_))));
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );

        // To non-existent version
        let result = migrate_down::<TestMigrator>(&db, false, "foobar").await;
        assert!(matches!(result, Err(Error::MigrationNotFound(_))));
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );

        // Upgrade back to fourth, ensure we can still upgrade again.
        migrate_up::<TestMigrator>(&db, false, "m20230401_000000_test_migration_4")
            .await
            .unwrap();
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..4]
        );
    }

    #[tokio::test]
    async fn accept_compatible_db() {
        install_tracing_subscriber();
        let db = test_database().await;
        check_database_is_compatible::<TestMigrator>(&db)
            .await
            .unwrap();

        // To third
        migrate_up::<TestMigrator>(&db, false, "m20230301_000000_test_migration_3")
            .await
            .unwrap();
        assert_eq!(
            applied_migrations(&db).await.unwrap(),
            all_migrations()[..3]
        );
        check_database_is_compatible::<TestMigrator>(&db)
            .await
            .unwrap();

        // To latest
        migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());
        check_database_is_compatible::<TestMigrator>(&db)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn reject_incompatible_db_using_wrong_migrator() {
        install_tracing_subscriber();

        struct AnotherTestMigrator;
        test_migration!(m20240101_000000_test_migration_1, AnotherTestTable1);
        test_migration!(m20240201_000000_test_migration_2, AnotherTestTable2);
        #[async_trait::async_trait]
        impl MigratorTrait for AnotherTestMigrator {
            fn migrations() -> Vec<Box<dyn MigrationTrait>> {
                vec![
                    Box::new(m20240101_000000_test_migration_1),
                    Box::new(m20240201_000000_test_migration_2),
                ]
            }
        }

        // DB brought up with AnotherTestMigrator. Simulates database brought
        // up on an entirely different schema.
        {
            let db = test_database().await;
            migrate_up::<AnotherTestMigrator>(&db, false, "m20240201_000000_test_migration_2")
                .await
                .unwrap();
            assert_eq!(
                applied_migrations(&db).await.unwrap(),
                vec![
                    "m20240101_000000_test_migration_1",
                    "m20240201_000000_test_migration_2",
                ]
            );

            // Use wrong TestMigrator, result should be incompatible.
            assert!(matches!(
                check_database_is_compatible::<TestMigrator>(&db).await,
                Err(Error::DbNotCompatible),
            ));
        }
        {
            let db = test_database().await;
            migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
                .await
                .unwrap();
            assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

            // Use wrong TestMigrator, result should be incompatible.
            assert!(matches!(
                check_database_is_compatible::<AnotherTestMigrator>(&db).await,
                Err(Error::DbNotCompatible),
            ));
        }
    }

    #[tokio::test]
    async fn reject_incompatible_db_using_outdated_migrator() {
        install_tracing_subscriber();

        // To latest
        let db = test_database().await;
        migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

        // Insert an additional fake migration, to simulate the database having
        // a newer schema than this tool supports.
        seaql_migrations::ActiveModel {
            version: ActiveValue::Set("m20230601_000000_test_migration_6".to_owned()),
            applied_at: ActiveValue::Set(
                SystemTime::now()
                    .duration_since(SystemTime::UNIX_EPOCH)
                    .unwrap()
                    .as_secs() as i64,
            ),
        }
        .insert(&db)
        .await
        .unwrap();

        assert!(matches!(
            check_database_is_compatible::<TestMigrator>(&db).await,
            Err(Error::DbNotCompatible),
        ));
    }

    #[tokio::test]
    async fn reject_incompatible_db_tampered() {
        install_tracing_subscriber();

        let db = test_database().await;

        // To latest
        migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")
            .await
            .unwrap();
        assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());

        // Tamper with schema table.
        seaql_migrations::Entity::delete_by_id("m20230301_000000_test_migration_3")
            .exec(&db)
            .await
            .unwrap();
        assert!(matches!(
            check_database_is_compatible::<TestMigrator>(&db).await,
            Err(Error::DbNotCompatible),
        ));
    }

    #[test]
    fn ensure_migrations_are_sorted() {
        // Migrations in Migrator must be in lexicographic order, otherwise
        // this CLI will not work correctly.
        assert!(
            Migrator::migrations()
                .windows(2)
                .all(|window| window[0].name() <= window[1].name())
        )
    }

    fn install_tracing_subscriber() {
        static INSTALL_TRACE_SUBSCRIBER: Once = Once::new();
        INSTALL_TRACE_SUBSCRIBER.call_once(|| {
            tracing_subscriber::fmt()
                .with_env_filter(EnvFilter::from_default_env())
                .with_test_writer()
                .init();
        });
    }
}

1	use clap::{Parser, ValueEnum, builder::PossibleValuesParser, command};
2	use sea_orm_migration::{
3	MigratorTrait, SchemaManager,
4	sea_orm::{ConnectOptions, Database, DatabaseConnection, DbErr, EntityTrait, QueryOrder},
5	seaql_migrations,
6	};
7	use std::cmp::Ordering;
8	use tracing::{error, info};
9	use tracing_subscriber::EnvFilter;
10
11	use migration::Migrator;
12
13	#[derive(Copy, Clone, ValueEnum, Debug)]
14	enum Direction {
15	Up,
16	Down,
17	}
18
19	/// Wraps the SeaORM migration library to provide additional features. Lets
20	/// you bring migrations up or down to a particular version. Allows dry-run
21	/// of migrations.
22	#[derive(Parser, Debug)]
23	#[command(about, version)]
24	struct Args {
25	#[arg(value_enum)]
26	direction: Direction,
27	#[arg(value_parser = available_migrations())]
28	target_version: String,
29	#[arg(short, long)]
30	dry_run: bool,
31	#[arg(short = 'u', long, env = "DATABASE_URL")]
32	database_url: String,
33	}
34
35	#[tokio::main]
36	async fn main() -> Result<(), Error> {	×
37	let args = Args::parse();	×
38	tracing_subscriber::fmt()	×
39	.with_env_filter(EnvFilter::from_default_env())	×
40	.with_ansi(false)	×
UNCOV 41	.init();	×
42
43	let db = Database::connect(ConnectOptions::new(args.database_url)).await?;	×
44	check_database_is_compatible::<Migrator>(&db).await?;	×
45	match args.direction {	×
46	Direction::Up => migrate_up::<Migrator>(&db, args.dry_run, &args.target_version).await?,	×
47	Direction::Down => {	×
48	migrate_down::<Migrator>(&db, args.dry_run, &args.target_version).await?	×
49	}	×
50	}	×
51	Ok(())	×
UNCOV 52	}	×
53
54	/// Checks that the database is compatible with the given migrator. Applied
55	/// migrations must be a subset of the migrations available by the MigratorTrait
56	/// for this CLI to be operated safely.
57	async fn check_database_is_compatible<M: MigratorTrait>(	7✔
58	db: &DatabaseConnection,	7✔
59	) -> Result<(), Error> {	7✔
60	// If the database is uninitialized, we can continue. The migrator will
61	// initialize the database for us.
62	if !has_migrations_table(db).await? {	7✔
63	return Ok(());	1✔
64	}	6✔
65
66	let migrations: Vec<String> = M::migrations().iter().map(\|m\| m.name().into()).collect();	27✔
67	let applied_migrations = applied_migrations(db).await?;	6✔
68
69	let print_error = \|\| {	6✔
70	error!("expected migrations: {:?}", migrations);	4✔
71	error!("present migrations: {:?}", applied_migrations);	4✔
72	};	4✔
73	for (i, applied) in applied_migrations.iter().enumerate() {	19✔
74	match migrations.get(i) {	19✔
75	Some(migration) if migration != applied => {	18✔
76	print_error();	3✔
77	return Err(Error::DbNotCompatible);	3✔
78	}
79	None => {
80	print_error();	1✔
81	return Err(Error::DbNotCompatible);	1✔
82	}
83	_ => {}	15✔
84	}
85	}
86	Ok(())	2✔
87	}	7✔
88
89	async fn migrate_up<M: MigratorTrait>(	15✔
90	db: &DatabaseConnection,	15✔
91	dry_run: bool,	15✔
92	target: &str,	15✔
93	) -> Result<(), Error> {	15✔
94	let target_index =	14✔
95	migration_index::<M>(target).ok_or(Error::MigrationNotFound(target.to_string()))?;	15✔
96
97	let (migrations_range, num_migrations) = match latest_applied_migration(db).await {	14✔
98	Ok(latest_migration) => {	6✔
99	let latest_index =	6✔
100	migration_index::<M>(&latest_migration).ok_or(Error::DbMigrationNotFound)?;	6✔
101	match target_index.cmp(&latest_index) {	6✔
102	Ordering::Less => return Err(Error::VersionTooOld(target.to_string())),	1✔
103	Ordering::Equal => {
104	info!("no action taken, already at desired version");	2✔
105	return Ok(());	2✔
106	}
107	Ordering::Greater => (	3✔
108	(latest_index + 1)..=target_index,	3✔
109	target_index - latest_index,	3✔
110	),	3✔
111	}
112	}
113	Err(Error::DbNotInitialized) => (	8✔
114	0usize..=target_index,	8✔
115	// The migration API takes "number of migrations to apply". If we have an	8✔
116	// uninitialized database, and we want to apply the first migration (index 0),	8✔
117	// then we still have to apply at least one migration.	8✔
118	target_index + 1,	8✔
119	),	8✔
UNCOV 120	Err(err) => return Err(err),	×
121	};
122
123	info!(	11✔
124	"executing {num_migrations} up migration(s) to reach {target}: {:?}",	×
125	Migrator::migrations()[migrations_range]	×
126	.iter()	×
127	.map(\|m\| m.name())	×
UNCOV 128	.collect::<Vec<_>>()	×
129	);
130	if !dry_run {	11✔
131	M::up(db, Some(u32::try_from(num_migrations)?))	11✔
132	.await	11✔
133	.map_err(Error::from)?	11✔
UNCOV 134	}	×
135	Ok(())	11✔
136	}	15✔
137
138	async fn migrate_down<M: MigratorTrait>(	6✔
139	db: &DatabaseConnection,	6✔
140	dry_run: bool,	6✔
141	target: &str,	6✔
142	) -> Result<(), Error> {	6✔
143	let latest_index = migration_index::<M>(&latest_applied_migration(db).await?)	6✔
144	.ok_or(Error::DbMigrationNotFound)?;	5✔
145	let target_index =	4✔
146	migration_index::<M>(target).ok_or(Error::MigrationNotFound(target.to_string()))?;	5✔
147
148	let num_migrations = match latest_index.cmp(&target_index) {	4✔
149	Ordering::Less => return Err(Error::VersionTooNew(target.to_string())),	1✔
150	Ordering::Equal => {
151	info!("no action taken, already at desired version");	1✔
152	return Ok(());	1✔
153	}
154	Ordering::Greater => latest_index - target_index,	2✔
155	};
156
157	info!(	2✔
158	"executing {num_migrations} down migration(s) to reach {target}: {:?}",	×
159	Migrator::migrations()[(target_index + 1)..=(latest_index)]	×
160	.iter()	×
161	.rev()	×
162	.map(\|m\| m.name())	×
UNCOV 163	.collect::<Vec<_>>()	×
164	);
165	if !dry_run {	2✔
166	M::down(db, Some(u32::try_from(num_migrations)?))	1✔
167	.await	1✔
168	.map_err(Error::from)?	1✔
169	}	1✔
170	Ok(())	2✔
171	}	6✔
172
173	fn migration_index<M: MigratorTrait>(version: &str) -> Option<usize> {	31✔
174	M::migrations().iter().position(\|m\| m.name() == version)	112✔
175	}	31✔
176
177	async fn has_migrations_table(db: &DatabaseConnection) -> Result<bool, Error> {	27✔
178	Ok(SchemaManager::new(db).has_table("seaql_migrations").await?)	27✔
179	}	27✔
180
181	async fn latest_applied_migration(db: &DatabaseConnection) -> Result<String, Error> {	20✔
182	if !has_migrations_table(db).await? {	20✔
183	return Err(Error::DbNotInitialized);	9✔
184	}	11✔
185	Ok(seaql_migrations::Entity::find()	11✔
186	.order_by_desc(seaql_migrations::Column::Version)	11✔
187	.one(db)	11✔
188	.await?	11✔
189	// The migrations table exists, but no migrations have been applied.
190	.ok_or(Error::DbNotInitialized)?	11✔
191	.version)
192	}	20✔
193
194	async fn applied_migrations(db: &DatabaseConnection) -> Result<Vec<String>, Error> {	26✔
195	Ok(seaql_migrations::Entity::find()	26✔
196	.order_by_asc(seaql_migrations::Column::Version)	26✔
197	.all(db)	26✔
198	.await?	26✔
199	.into_iter()	26✔
200	.map(\|m\| m.version)	26✔
201	.collect())	26✔
202	}	26✔
203
204	#[derive(Debug, thiserror::Error)]
205	enum Error {
206	#[error("DB error: {0}")]
207	Db(#[from] DbErr),
208	#[error("DB is not initialized with migrations table")]
209	DbNotInitialized,
210	#[error("migration applied to DB is not found in available migrations")]
211	DbMigrationNotFound,
212	#[error("migration version {0} not found in avaliable migrations")]
213	MigrationNotFound(String),
214	#[error("migration version {0} is older than the latest applied migration")]
215	VersionTooOld(String),
216	#[error("migration version {0} is newer than the latest applied migration")]
217	VersionTooNew(String),
218	#[error("error calculating number of migrations, too many migrations?: {0}")]
219	Overflow(#[from] std::num::TryFromIntError),
220	#[error("applied migrations do not match migrations present in this tool")]
221	DbNotCompatible,
222	}
223
224	fn available_migrations() -> PossibleValuesParser {	×
UNCOV 225	PossibleValuesParser::new(	×
226	// Leak memory to give migration names 'static lifetime, so clap can
227	// use them.
228	Migrator::migrations()	×
229	.into_iter()	×
230	.map(\|m\| Box::leak(Box::new(m.name().to_owned())) as &'static str)	×
UNCOV 231	.collect::<Vec<_>>(),	×
232	)
UNCOV 233	}	×
234
235	#[cfg(test)]
236	mod tests {
237	use std::{sync::Once, time::SystemTime};
238
239	use super::*;
240	use sea_orm::{ActiveModelTrait, ActiveValue};
241	use sea_orm_migration::prelude::*;
242
243	macro_rules! test_migration {
244	($name:ident, $table_name:ident) => {
245	#[allow(non_camel_case_types)]
246	struct $name;
247
248	impl MigrationName for $name {
249	fn name(&self) -> &str {	291✔
250	stringify!($name)	291✔
251	}	291✔
252	}
253
254	#[async_trait::async_trait]
255	impl MigrationTrait for $name {
256	async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {	36✔
257	manager	36✔
258	.create_table(	36✔
259	Table::create()	36✔
260	.table($table_name::Table)	36✔
261	.col(ColumnDef::new($table_name::Id).uuid().primary_key())	36✔
262	.to_owned(),	36✔
263	)	36✔
264	.await?;	36✔
265	Ok(())	36✔
266	}	72✔
267	async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {	2✔
268	manager	2✔
269	.drop_table(Table::drop().table($table_name::Table).to_owned())	2✔
270	.await	2✔
271	}	4✔
272	}
273
274	#[derive(Iden)]
275	enum $table_name {
276	Table,
277	Id,
278	}
279	};
280	}
281	test_migration!(m20230101_000000_test_migration_1, TestTable1);
282	test_migration!(m20230201_000000_test_migration_2, TestTable2);
283	test_migration!(m20230301_000000_test_migration_3, TestTable3);
284	test_migration!(m20230401_000000_test_migration_4, TestTable4);
285	test_migration!(m20230501_000000_test_migration_5, TestTable5);
286
287	struct TestMigrator;
288
289	#[async_trait::async_trait]
290	impl MigratorTrait for TestMigrator {
291	fn migrations() -> Vec<Box<dyn MigrationTrait>> {	46✔
292	vec![	46✔
293	Box::new(m20230101_000000_test_migration_1),	46✔
294	Box::new(m20230201_000000_test_migration_2),	46✔
295	Box::new(m20230301_000000_test_migration_3),	46✔
296	Box::new(m20230401_000000_test_migration_4),	46✔
297	Box::new(m20230501_000000_test_migration_5),	46✔
298	]
299	}	46✔
300	}
301
302	fn all_migrations() -> Vec<&'static str> {	17✔
303	vec![	17✔
304	"m20230101_000000_test_migration_1",	17✔
305	"m20230201_000000_test_migration_2",	17✔
306	"m20230301_000000_test_migration_3",	17✔
307	"m20230401_000000_test_migration_4",	17✔
308	"m20230501_000000_test_migration_5",	17✔
309	]
310	}	17✔
311
312	async fn test_database() -> DatabaseConnection {	8✔
313	Database::connect(ConnectOptions::new("sqlite::memory:".to_string()))	8✔
314	.await	8✔
315	.unwrap()	8✔
316	}	8✔
317
318	#[tokio::test]
319	async fn migrate_up_latest() {	1✔
320	install_tracing_subscriber();	1✔
321	let db = test_database().await;	1✔
322
323	// To latest
324	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
325	.await	1✔
326	.unwrap();	1✔
327	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
328
329	// Ensure no-op
330	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
331	.await	1✔
332	.unwrap();	1✔
333	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
334	}	1✔
335
336	#[tokio::test]
337	async fn migrate_up_works() {	1✔
338	install_tracing_subscriber();	1✔
339	let db = test_database().await;	1✔
340
341	// To first
342	migrate_up::<TestMigrator>(&db, false, "m20230101_000000_test_migration_1")	1✔
343	.await	1✔
344	.unwrap();	1✔
345	assert_eq!(	1✔
346	applied_migrations(&db).await.unwrap(),	1✔
347	vec!["m20230101_000000_test_migration_1"]	1✔
348	);
349
350	// Dry run
351	migrate_up::<TestMigrator>(&db, true, "m20230101_000000_test_migration_1")	1✔
352	.await	1✔
353	.unwrap();	1✔
354	assert_eq!(	1✔
355	applied_migrations(&db).await.unwrap(),	1✔
356	vec!["m20230101_000000_test_migration_1"]	1✔
357	);
358
359	// To third
360	migrate_up::<TestMigrator>(&db, false, "m20230301_000000_test_migration_3")	1✔
361	.await	1✔
362	.unwrap();	1✔
363	assert_eq!(	1✔
364	applied_migrations(&db).await.unwrap(),	1✔
365	all_migrations()[..3]	1✔
366	);
367
368	// To non-existent
369	let result = migrate_up::<TestMigrator>(&db, false, "foobar").await;	1✔
370	assert!(matches!(result, Err(Error::MigrationNotFound(_))));	1✔
371	assert_eq!(	1✔
372	applied_migrations(&db).await.unwrap(),	1✔
373	all_migrations()[..3]	1✔
374	);
375
376	// To old version
377	let result =	1✔
378	migrate_up::<TestMigrator>(&db, false, "m20230101_000000_test_migration_1").await;	1✔
379	assert!(matches!(result, Err(Error::VersionTooOld(_))));	1✔
380	assert_eq!(	1✔
381	applied_migrations(&db).await.unwrap(),	1✔
382	all_migrations()[..3]	1✔
383	);	1✔
384	}	1✔
385
386	#[tokio::test]
387	async fn migrate_down_works() {	1✔
388	install_tracing_subscriber();	1✔
389	let db = test_database().await;	1✔
390	let result =	1✔
391	migrate_down::<TestMigrator>(&db, false, "m20230401_000000_test_migration_4").await;	1✔
392	assert!(matches!(result, Err(Error::DbNotInitialized)));	1✔
393
394	// Fail if DB not initialized
395
396	// To latest
397	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
398	.await	1✔
399	.unwrap();	1✔
400	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
401
402	// Ensure no-op
403	migrate_down::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
404	.await	1✔
405	.unwrap();	1✔
406	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
407
408	// Dry-run
409	migrate_down::<TestMigrator>(&db, true, "m20230301_000000_test_migration_3")	1✔
410	.await	1✔
411	.unwrap();	1✔
412	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
413
414	// To third
415	migrate_down::<TestMigrator>(&db, false, "m20230301_000000_test_migration_3")	1✔
416	.await	1✔
417	.unwrap();	1✔
418	assert_eq!(	1✔
419	applied_migrations(&db).await.unwrap(),	1✔
420	all_migrations()[..3]	1✔
421	);
422
423	// To newer version
424	let result =	1✔
425	migrate_down::<TestMigrator>(&db, false, "m20230401_000000_test_migration_4").await;	1✔
426	assert!(matches!(result, Err(Error::VersionTooNew(_))));	1✔
427	assert_eq!(	1✔
428	applied_migrations(&db).await.unwrap(),	1✔
429	all_migrations()[..3]	1✔
430	);
431
432	// To non-existent version
433	let result = migrate_down::<TestMigrator>(&db, false, "foobar").await;	1✔
434	assert!(matches!(result, Err(Error::MigrationNotFound(_))));	1✔
435	assert_eq!(	1✔
436	applied_migrations(&db).await.unwrap(),	1✔
437	all_migrations()[..3]	1✔
438	);
439
440	// Upgrade back to fourth, ensure we can still upgrade again.
441	migrate_up::<TestMigrator>(&db, false, "m20230401_000000_test_migration_4")	1✔
442	.await	1✔
443	.unwrap();	1✔
444	assert_eq!(	1✔
445	applied_migrations(&db).await.unwrap(),	1✔
446	all_migrations()[..4]	1✔
447	);	1✔
448	}	1✔
449
450	#[tokio::test]
451	async fn accept_compatible_db() {	1✔
452	install_tracing_subscriber();	1✔
453	let db = test_database().await;	1✔
454	check_database_is_compatible::<TestMigrator>(&db)	1✔
455	.await	1✔
456	.unwrap();	1✔
457
458	// To third
459	migrate_up::<TestMigrator>(&db, false, "m20230301_000000_test_migration_3")	1✔
460	.await	1✔
461	.unwrap();	1✔
462	assert_eq!(	1✔
463	applied_migrations(&db).await.unwrap(),	1✔
464	all_migrations()[..3]	1✔
465	);
466	check_database_is_compatible::<TestMigrator>(&db)	1✔
467	.await	1✔
468	.unwrap();	1✔
469
470	// To latest
471	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
472	.await	1✔
473	.unwrap();	1✔
474	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
475	check_database_is_compatible::<TestMigrator>(&db)	1✔
476	.await	1✔
477	.unwrap();	1✔
478	}	1✔
479
480	#[tokio::test]
481	async fn reject_incompatible_db_using_wrong_migrator() {	1✔
482	install_tracing_subscriber();	1✔
483
484	struct AnotherTestMigrator;
485	test_migration!(m20240101_000000_test_migration_1, AnotherTestTable1);
486	test_migration!(m20240201_000000_test_migration_2, AnotherTestTable2);
487	#[async_trait::async_trait]	1✔
488	impl MigratorTrait for AnotherTestMigrator {	1✔
489	fn migrations() -> Vec<Box<dyn MigrationTrait>> {	3✔
490	vec![	3✔
491	Box::new(m20240101_000000_test_migration_1),	3✔
492	Box::new(m20240201_000000_test_migration_2),	3✔
493	]	1✔
494	}	3✔
495	}	1✔
496		1✔
497	// DB brought up with AnotherTestMigrator. Simulates database brought	1✔
498	// up on an entirely different schema.	1✔
499	{	1✔
500	let db = test_database().await;	1✔
501	migrate_up::<AnotherTestMigrator>(&db, false, "m20240201_000000_test_migration_2")	1✔
502	.await	1✔
503	.unwrap();	1✔
504	assert_eq!(	1✔
505	applied_migrations(&db).await.unwrap(),	1✔
506	vec![	1✔
507	"m20240101_000000_test_migration_1",	1✔
508	"m20240201_000000_test_migration_2",	1✔
509	]	1✔
510	);	1✔
511		1✔
512	// Use wrong TestMigrator, result should be incompatible.	1✔
513	assert!(matches!(	1✔
514	check_database_is_compatible::<TestMigrator>(&db).await,	1✔
515	Err(Error::DbNotCompatible),	1✔
516	));	1✔
517	}	1✔
518	{	1✔
519	let db = test_database().await;	1✔
520	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
521	.await	1✔
522	.unwrap();	1✔
523	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
524		1✔
525	// Use wrong TestMigrator, result should be incompatible.	1✔
526	assert!(matches!(	1✔
527	check_database_is_compatible::<AnotherTestMigrator>(&db).await,	1✔
528	Err(Error::DbNotCompatible),	1✔
529	));	1✔
530	}	1✔
531	}	1✔
532
533	#[tokio::test]
534	async fn reject_incompatible_db_using_outdated_migrator() {	1✔
535	install_tracing_subscriber();	1✔
536
537	// To latest
538	let db = test_database().await;	1✔
539	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
540	.await	1✔
541	.unwrap();	1✔
542	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
543
544	// Insert an additional fake migration, to simulate the database having
545	// a newer schema than this tool supports.
546	seaql_migrations::ActiveModel {	1✔
547	version: ActiveValue::Set("m20230601_000000_test_migration_6".to_owned()),	1✔
548	applied_at: ActiveValue::Set(	1✔
549	SystemTime::now()	1✔
550	.duration_since(SystemTime::UNIX_EPOCH)	1✔
551	.unwrap()	1✔
552	.as_secs() as i64,	1✔
553	),	1✔
554	}	1✔
555	.insert(&db)	1✔
556	.await	1✔
557	.unwrap();	1✔
558
559	assert!(matches!(	1✔
560	check_database_is_compatible::<TestMigrator>(&db).await,	1✔
561	Err(Error::DbNotCompatible),	1✔
562	));	1✔
563	}	1✔
564
565	#[tokio::test]
566	async fn reject_incompatible_db_tampered() {	1✔
567	install_tracing_subscriber();	1✔
568
569	let db = test_database().await;	1✔
570
571	// To latest
572	migrate_up::<TestMigrator>(&db, false, "m20230501_000000_test_migration_5")	1✔
573	.await	1✔
574	.unwrap();	1✔
575	assert_eq!(applied_migrations(&db).await.unwrap(), all_migrations());	1✔
576
577	// Tamper with schema table.
578	seaql_migrations::Entity::delete_by_id("m20230301_000000_test_migration_3")	1✔
579	.exec(&db)	1✔
580	.await	1✔
581	.unwrap();	1✔
582	assert!(matches!(	1✔
583	check_database_is_compatible::<TestMigrator>(&db).await,	1✔
584	Err(Error::DbNotCompatible),	1✔
585	));	1✔
586	}	1✔
587
588	#[test]
589	fn ensure_migrations_are_sorted() {	1✔
590	// Migrations in Migrator must be in lexicographic order, otherwise
591	// this CLI will not work correctly.
592	assert!(	1✔
593	Migrator::migrations()	1✔
594	.windows(2)	1✔
595	.all(\|window\| window[0].name() <= window[1].name())	25✔
596	)
597	}	1✔
598
599	fn install_tracing_subscriber() {	7✔
600	static INSTALL_TRACE_SUBSCRIBER: Once = Once::new();
601	INSTALL_TRACE_SUBSCRIBER.call_once(\|\| {	7✔
602	tracing_subscriber::fmt()	7✔
603	.with_env_filter(EnvFilter::from_default_env())	7✔
604	.with_test_writer()	7✔
605	.init();	7✔
606	});	7✔
607	}	7✔
608	}

divviup / divviup-api / 16810179837

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous