slave-framework 0.7.0

//! Migration engine tests against a real libSQL server: a dedicated Turso
//! Cloud test database or a throwaway local one (see `slave_framework::testing`).

use slave_framework::Error;
use slave_framework::migration::console::{format_rollback, format_run, format_status};
use slave_framework::migration::{Migration, Migrator, Registered, Schema, State, TABLE};
use slave_framework::testing::TestDatabase;

struct CreateTable(&'static str);

impl Migration for CreateTable {
    fn up(&self) -> Schema {
        Schema::new().execute(format!("CREATE TABLE {} (id INTEGER PRIMARY KEY)", self.0))
    }

    fn down(&self) -> Option<Schema> {
        Some(Schema::new().execute(format!("DROP TABLE {}", self.0)))
    }
}

/// Adds a column; reversible via DROP COLUMN.
struct AddColumn;

impl Migration for AddColumn {
    fn up(&self) -> Schema {
        Schema::new().execute("ALTER TABLE users ADD COLUMN email TEXT")
    }

    fn down(&self) -> Option<Schema> {
        Some(Schema::new().execute("ALTER TABLE users DROP COLUMN email"))
    }
}

/// Creates a table, then fails on its second statement.
struct Broken;

impl Migration for Broken {
    fn up(&self) -> Schema {
        Schema::new()
            .execute("CREATE TABLE half_done (id INTEGER PRIMARY KEY)")
            .execute("INSERT INTO table_that_does_not_exist VALUES (1)")
    }
}

/// Drops data, so it declares no `down`.
struct Irreversible;

impl Migration for Irreversible {
    fn up(&self) -> Schema {
        Schema::new().execute("DROP TABLE posts")
    }
}

static USERS: CreateTable = CreateTable("users");
static POSTS: CreateTable = CreateTable("posts");
static COMMENTS: CreateTable = CreateTable("comments");

fn users() -> Registered {
    Registered::new("20260101000000_create_users_table", &USERS)
}
fn posts() -> Registered {
    Registered::new("20260101000100_create_posts_table", &POSTS)
}
fn add_email() -> Registered {
    Registered::new("20260101000200_add_email_to_users", &AddColumn)
}
fn comments() -> Registered {
    Registered::new("20260101000300_create_comments_table", &COMMENTS)
}

/// An empty database for one test (see `slave_framework::testing`).
struct TempDb(TestDatabase);

impl TempDb {
    fn new() -> Self {
        Self(TestDatabase::start())
    }

    async fn migrator(&self, migrations: Vec<Registered>) -> Migrator {
        let conn = self.0.database().await.connect().await.unwrap();
        Migrator::new(conn, migrations).unwrap()
    }
}

async fn tables(migrator: &Migrator) -> Vec<String> {
    let mut rows = migrator
        .connection()
        .query(
            "SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'libsql_%' ORDER BY name",
            (),
        )
        .await
        .unwrap();
    let mut names = Vec::new();
    while let Some(row) = rows.next().await.unwrap() {
        names.push(row.get::<String>(0).unwrap());
    }
    names
}

async fn columns(migrator: &Migrator, table: &str) -> Vec<String> {
    let mut rows = migrator
        .connection()
        .query(
            &format!("SELECT name FROM pragma_table_info('{table}')"),
            (),
        )
        .await
        .unwrap();
    let mut names = Vec::new();
    while let Some(row) = rows.next().await.unwrap() {
        names.push(row.get::<String>(0).unwrap());
    }
    names
}

fn names(outcomes: &[slave_framework::migration::Outcome]) -> Vec<&str> {
    outcomes.iter().map(|o| o.name.as_str()).collect()
}

#[tokio::test]
async fn initial_run_applies_all_in_order_as_batch_one() {
    let db = TempDb::new();
    // Registered out of order on purpose: execution order comes from the names.
    let mut migrator = db.migrator(vec![add_email(), users(), posts()]).await;

    let report = migrator.run().await.unwrap();
    assert_eq!(report.batch, 1);
    assert_eq!(
        names(&report.applied),
        [users().name(), posts().name(), add_email().name()]
    );
    assert!(report.skipped.is_empty() && report.failed.is_none());
    assert_eq!(
        tables(&migrator).await,
        ["posts", "slave_migrations", "users"]
    );
    assert_eq!(columns(&migrator, "users").await, ["id", "email"]);
    assert!(format_run(&report).contains("Batch 1: 3 applied, 0 skipped"));
}

#[tokio::test]
async fn repeated_runs_skip_applied_migrations() {
    let db = TempDb::new();
    db.migrator(vec![users()]).await.run().await.unwrap();

    // A fresh connection, as a second `slave migrate` would have.
    let mut migrator = db.migrator(vec![users()]).await;
    let report = migrator.run().await.unwrap();
    assert!(report.applied.is_empty());
    assert_eq!(report.skipped, [users().name()]);
    assert_eq!(
        format_run(&report),
        "Nothing to migrate (1 already applied).\n"
    );

    // New migrations go into the next batch.
    let mut migrator = db.migrator(vec![users(), posts()]).await;
    let report = migrator.run().await.unwrap();
    assert_eq!(report.batch, 2);
    assert_eq!(names(&report.applied), [posts().name()]);
    assert_eq!(report.skipped, [users().name()]);
}

#[tokio::test]
async fn status_reports_applied_pending_and_missing() {
    let db = TempDb::new();
    db.migrator(vec![users(), posts()])
        .await
        .run()
        .await
        .unwrap();

    // `posts` was applied but is no longer registered; `add_email` is new.
    let mut migrator = db.migrator(vec![users(), add_email()]).await;
    let statuses = migrator.status().await.unwrap();
    let summary: Vec<(&str, &str)> = statuses
        .iter()
        .map(|s| {
            let state = match s.state {
                State::Applied { batch: 1, .. } => "applied",
                State::Pending => "pending",
                State::Missing { batch: 1, .. } => "missing",
                _ => "unexpected",
            };
            (s.name.as_str(), state)
        })
        .collect();
    assert_eq!(
        summary,
        [
            (users().name(), "applied"),
            (posts().name(), "missing"),
            (add_email().name(), "pending"),
        ]
    );

    let output = format_status(&statuses);
    assert!(output.starts_with("Status"), "{output}");
    assert!(output.contains("Missing: recorded as applied"), "{output}");
}

#[tokio::test]
async fn rollback_reverts_only_the_latest_batch_in_reverse_order() {
    let db = TempDb::new();
    db.migrator(vec![users()]).await.run().await.unwrap();
    let mut migrator = db.migrator(vec![users(), posts(), add_email()]).await;
    migrator.run().await.unwrap(); // batch 2: posts, add_email

    let report = migrator.rollback().await.unwrap();
    assert_eq!(report.batch, Some(2));
    assert_eq!(
        names(&report.rolled_back),
        [add_email().name(), posts().name()]
    );
    assert_eq!(tables(&migrator).await, ["slave_migrations", "users"]);
    assert_eq!(columns(&migrator, "users").await, ["id"]);
    assert!(format_rollback(&report).contains("Rolling back batch 2"));

    // Rolled-back migrations are pending again and can be re-applied.
    let report = migrator.run().await.unwrap();
    assert_eq!(report.batch, 2);
    assert_eq!(report.applied.len(), 2);

    migrator.rollback().await.unwrap();
    assert_eq!(migrator.rollback().await.unwrap().batch, Some(1));
    let report = migrator.rollback().await.unwrap();
    assert_eq!(report.batch, None);
    assert_eq!(format_rollback(&report), "Nothing to roll back.\n");
    assert_eq!(tables(&migrator).await, ["slave_migrations"]);
}

#[tokio::test]
async fn failed_migration_is_rolled_back_and_not_recorded() {
    let db = TempDb::new();
    let broken = Registered::new("20260101000150_broken", &Broken);
    let mut migrator = db.migrator(vec![users(), broken, add_email()]).await;

    let report = migrator.run().await.unwrap();
    assert_eq!(names(&report.applied), [users().name()]);
    let failure = report.failed.clone().expect("failure");
    assert_eq!(failure.name, broken.name());
    assert!(failure.error.contains("statement 2"), "{}", failure.error);
    assert!(failure.reverted);
    let output = format_run(&report);
    assert!(output.contains("FAILED  20260101000150_broken"), "{output}");
    assert!(
        output.contains("was rolled back and not recorded"),
        "{output}"
    );

    // The first statement of the broken migration was undone, and the run stopped.
    assert_eq!(tables(&migrator).await, ["slave_migrations", "users"]);
    assert_eq!(columns(&migrator, "users").await, ["id"]);
    let statuses = migrator.status().await.unwrap();
    assert_eq!(statuses[1].state, State::Pending);
    assert_eq!(statuses[2].state, State::Pending);

    // Running again retries the failed migration rather than skipping it.
    let report = migrator.run().await.unwrap();
    assert_eq!(
        report.failed.map(|f| f.name),
        Some(broken.name().to_owned())
    );
}

#[tokio::test]
async fn rollback_refuses_batches_with_irreversible_migrations() {
    let db = TempDb::new();
    let drop_posts = Registered::new("20260101000400_drop_posts_table", &Irreversible);
    let mut migrator = db
        .migrator(vec![users(), posts(), comments(), drop_posts])
        .await;
    migrator.run().await.unwrap();

    let err = migrator.rollback().await.unwrap_err();
    assert!(
        matches!(err, Error::RollbackRefused { batch: 1, .. }),
        "{err}"
    );
    assert!(err.to_string().contains("drop_posts_table"), "{err}");
    // Nothing was reverted, not even the reversible migrations of that batch.
    assert_eq!(
        tables(&migrator).await,
        ["comments", "slave_migrations", "users"]
    );
}

#[tokio::test]
async fn rollback_refuses_batches_with_unregistered_migrations() {
    let db = TempDb::new();
    db.migrator(vec![users(), posts()])
        .await
        .run()
        .await
        .unwrap();

    let mut migrator = db.migrator(vec![users()]).await;
    let err = migrator.rollback().await.unwrap_err();
    assert!(err.to_string().contains("not registered"), "{err}");
    assert_eq!(
        tables(&migrator).await,
        ["posts", "slave_migrations", "users"]
    );
}

#[tokio::test]
async fn duplicate_names_are_rejected() {
    let db = TestDatabase::start();
    let conn = db.database().await.connect().await.unwrap();
    let err = Migrator::new(conn, vec![users(), users()])
        .err()
        .expect("error");
    assert!(matches!(err, Error::DuplicateMigration(_)), "{err}");
}

#[tokio::test]
async fn tracking_table_records_batches_and_timestamps() {
    let db = TempDb::new();
    let mut migrator = db.migrator(vec![users()]).await;
    migrator.run().await.unwrap();

    let mut rows = migrator
        .connection()
        .query(&format!("SELECT name, batch, executed_at FROM {TABLE}"), ())
        .await
        .unwrap();
    let row = rows.next().await.unwrap().expect("tracking row");
    assert_eq!(row.get::<String>(0).unwrap(), users().name());
    assert_eq!(row.get::<i64>(1).unwrap(), 1);
    assert_eq!(
        row.get::<String>(2).unwrap().len(),
        "2026-01-01 00:00:00".len()
    );
}

#[test]
fn registered_from_module_strips_prefix() {
    let r = Registered::from_module("m20260101000000_create_users_table", &USERS);
    assert_eq!(r.name(), "20260101000000_create_users_table");
}