orbit-orm 0.3.0

//! Runs the blocking API against a real database: MySQL or MariaDB, PostgreSQL or SQLite. Set
//! `ORBIT_DATABASE_URL` to a test database, such as `mysql://user@127.0.0.1:3306/orbit_test`. The
//! tests are skipped with a notice when it is not set. Each test uses its own table and drops it
//! at the end.

mod common;

use std::panic::{self, AssertUnwindSafe};
use std::sync::Arc;

use orbit_orm::blocking::Database;
use orbit_orm::{Error, Model, OrbitModel};

fn connect() -> Option<Database> {
    let url = common::url()?;
    Some(Database::connect(&url).expect("connects"))
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_notes")]
struct Note {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_tx_commit")]
struct Committed {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_tx_rollback")]
struct RolledBack {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_tx_panic")]
struct Panicked {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_tx_own")]
struct Own {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_tx_spawn_panic")]
struct SpawnPanicked {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_spawn")]
struct Spawned {
    id: i64,
    #[orbit(fillable)]
    title: String,
}

#[derive(Debug, PartialEq, OrbitModel)]
#[orbit(table = "orbit_blocking_codes")]
struct Code {
    id: i64,
    #[orbit(fillable)]
    code: String,
    #[orbit(fillable)]
    title: String,
}

/// Runs a statement through an async connection to the same database. The blocking handle does
/// not expose its async pool, so the setup opens its own for each statement.
fn run_sql(db: &Database, sql: &str) {
    let url = common::url().expect("ORBIT_DATABASE_URL is set");
    db.run(async {
        let admin = orbit_orm::Database::connect(&url)
            .await
            .expect("connects for setup");
        common::exec(&admin, sql).await;
        admin.close().await;
    })
    .expect("runs on the blocking database");
}

/// Recreates `table` with an id and a title, as the models above expect.
fn setup(db: &Database, table: &str) {
    run_sql(db, &format!("DROP TABLE IF EXISTS `{table}`"));
    run_sql(
        db,
        &format!(
            "CREATE TABLE `{table}` (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                title VARCHAR(255) NOT NULL
            )"
        ),
    );
}

fn teardown(db: &Database, table: &str) {
    run_sql(db, &format!("DROP TABLE IF EXISTS `{table}`"));
}

#[test]
fn connects_and_answers_a_ping() {
    let Some(db) = connect() else { return };
    db.ping().expect("answers SELECT 1");
    db.close().expect("closes");
}

#[test]
fn a_wrong_password_is_an_error() {
    let Ok(url) = std::env::var("ORBIT_DATABASE_URL") else {
        return;
    };
    // SQLite has no users or passwords, so there is no login to get wrong.
    if url.starts_with("sqlite:") {
        return;
    }
    let wrong = url.replacen("://", "://orbit-no-such-user:wrong@", 1);
    assert!(Database::connect(&wrong).is_err());
}

#[test]
fn finds_a_row_and_inserts_one() {
    let Some(db) = connect() else { return };
    setup(&db, Note::TABLE);

    let id = Note::query()
        .blocking()
        .insert(&db, &[("title", "hello".into())])
        .expect("inserts");
    let found = Note::query().blocking().find(&db, id).expect("finds");
    assert_eq!(
        found,
        Some(Note {
            id: id as i64,
            title: "hello".to_owned()
        })
    );
    assert_eq!(
        Note::query()
            .blocking()
            .find(&db, 999)
            .expect("finds nothing"),
        None
    );
    assert!(matches!(
        Note::query().blocking().find_or_fail(&db, 999),
        Err(Error::NotFound { .. })
    ));

    teardown(&db, Note::TABLE);
    db.close().expect("closes");
}

#[test]
fn transaction_commits_when_the_closure_returns_ok() {
    let Some(db) = connect() else { return };
    setup(&db, Committed::TABLE);

    let id = db
        .transaction(|tx| {
            let id = Committed::query()
                .blocking()
                .insert(tx, &[("title", "kept".into())])?;
            Ok::<_, Error>(id)
        })
        .expect("commits");

    let saved = Committed::query().blocking().find(&db, id).expect("finds");
    assert_eq!(saved.map(|row| row.title), Some("kept".to_owned()));

    teardown(&db, Committed::TABLE);
    db.close().expect("closes");
}

#[test]
fn transaction_rolls_back_when_the_closure_returns_err() {
    let Some(db) = connect() else { return };
    setup(&db, RolledBack::TABLE);

    let result = db.transaction(|tx| {
        RolledBack::query()
            .blocking()
            .insert(tx, &[("title", "dropped".into())])?;
        Err::<(), _>(Error::NotFound {
            table: RolledBack::TABLE,
        })
    });
    assert!(matches!(result, Err(Error::NotFound { .. })));
    assert_eq!(
        RolledBack::query().blocking().count(&db).expect("counts"),
        0
    );

    teardown(&db, RolledBack::TABLE);
    db.close().expect("closes");
}

#[test]
fn replace_overwrites_on_a_unique_key() {
    let Some(db) = connect() else { return };
    run_sql(&db, "DROP TABLE IF EXISTS orbit_blocking_codes");
    run_sql(
        &db,
        "CREATE TABLE orbit_blocking_codes (
            id BIGINT AUTO_INCREMENT PRIMARY KEY,
            code VARCHAR(32) NOT NULL,
            title VARCHAR(255) NOT NULL,
            UNIQUE KEY `orbit_blocking_codes_code` (code)
        )",
    );

    Code::query()
        .blocking()
        .insert(&db, &[("code", "a".into()), ("title", "first".into())])
        .expect("inserts");
    let result = Code::query()
        .blocking()
        .replace(&db, &[("code", "a".into()), ("title", "second".into())]);
    if common::on_postgres(db.driver()) {
        // PostgreSQL has no REPLACE, so replace is refused there.
        assert!(
            matches!(
                result,
                Err(Error::Unsupported {
                    operation: "replace",
                    driver: "PostgreSQL"
                })
            ),
            "got {result:?}"
        );
        run_sql(&db, "DROP TABLE IF EXISTS orbit_blocking_codes");
        db.close().expect("closes");
        return;
    }
    let affected = result.expect("replaces");
    // MySQL deletes and inserts, so two rows are affected; SQLite reports one.
    if common::on_mysql(db.driver()) {
        assert_eq!(affected, 2);
    } else {
        assert_eq!(affected, 1);
    }

    let row = Code::query()
        .where_eq("code", "a")
        .blocking()
        .first_or_fail(&db)
        .expect("finds the replaced row");
    assert_eq!(row.title, "second");

    run_sql(&db, "DROP TABLE IF EXISTS orbit_blocking_codes");
    db.close().expect("closes");
}

#[test]
fn a_blocking_call_on_an_async_thread_is_an_error() {
    let Ok(url) = std::env::var("ORBIT_DATABASE_URL") else {
        eprintln!("skipped: set ORBIT_DATABASE_URL to run the database tests");
        return;
    };
    let Some(db) = connect() else { return };

    let runtime = tokio::runtime::Runtime::new().expect("builds a runtime");
    runtime.block_on(async {
        assert!(matches!(db.ping(), Err(Error::BlockingInRuntime)));
        assert!(matches!(
            Note::query().blocking().count(&db),
            Err(Error::BlockingInRuntime)
        ));
        assert!(matches!(
            db.transaction(|_| Ok::<(), Error>(())),
            Err(Error::BlockingInRuntime)
        ));
        assert!(matches!(
            Database::connect(&url),
            Err(Error::BlockingInRuntime)
        ));
    });

    db.close().expect("closes");
}

#[test]
fn spawn_blocking_runs_blocking_calls() {
    let Some(db) = connect() else { return };
    setup(&db, Spawned::TABLE);
    let runtime = tokio::runtime::Runtime::new().expect("builds a runtime");

    let (db, id) = runtime.block_on(async {
        tokio::task::spawn_blocking(move || {
            let id = Spawned::query()
                .blocking()
                .insert(&db, &[("title", "from a blocking thread".into())])
                .expect("inserts on a blocking thread");
            (db, id)
        })
        .await
        .expect("the blocking task finishes")
    });

    let (db, found) = runtime.block_on(async {
        tokio::task::spawn_blocking(move || {
            let found = Spawned::query().blocking().find(&db, id).expect("finds");
            (db, found)
        })
        .await
        .expect("the blocking task finishes")
    });
    assert_eq!(
        found.map(|row| row.title),
        Some("from a blocking thread".to_owned())
    );

    teardown(&db, Spawned::TABLE);
}

#[test]
fn transaction_rolls_back_when_the_closure_panics() {
    let Some(db) = connect() else { return };
    setup(&db, Panicked::TABLE);

    let outcome = panic::catch_unwind(AssertUnwindSafe(|| {
        db.transaction(|tx| {
            Panicked::query()
                .blocking()
                .insert(tx, &[("title", "never".into())])?;
            panic!("panic inside the transaction");
            #[allow(unreachable_code)]
            Ok::<(), Error>(())
        })
    }));
    let payload = outcome.expect_err("the panic reaches the caller");
    assert_eq!(
        payload.downcast_ref::<&str>().copied(),
        Some("panic inside the transaction")
    );

    assert_eq!(Panicked::query().blocking().count(&db).expect("counts"), 0);

    teardown(&db, Panicked::TABLE);
    db.close().expect("closes");
}

#[test]
fn queries_in_the_transaction_see_their_own_writes() {
    let Some(db) = connect() else { return };
    setup(&db, Own::TABLE);

    db.transaction(|tx| {
        Own::query()
            .blocking()
            .insert(&mut *tx, &[("title", "inside".into())])?;

        let inside = Own::query().blocking().count(&mut *tx)?;
        assert_eq!(inside, 1, "the transaction sees its own insert");

        let from_pool = Own::query().blocking().count(&db)?;
        assert_eq!(from_pool, 0, "the pool does not see the uncommitted insert");
        Ok::<(), Error>(())
    })
    .expect("commits");

    assert_eq!(Own::query().blocking().count(&db).expect("counts"), 1);

    teardown(&db, Own::TABLE);
    db.close().expect("closes");
}

#[test]
fn a_panic_on_a_blocking_thread_rolls_back_and_the_pool_still_works() {
    let Some(db) = connect() else { return };
    setup(&db, SpawnPanicked::TABLE);
    let db = Arc::new(db);
    let runtime = tokio::runtime::Runtime::new().expect("builds a runtime");

    let shared = Arc::clone(&db);
    let joined = runtime.block_on(async move {
        tokio::task::spawn_blocking(move || {
            shared.transaction(|tx| {
                SpawnPanicked::query()
                    .blocking()
                    .insert(tx, &[("title", "never".into())])?;
                panic!("panic on a blocking thread");
                #[allow(unreachable_code)]
                Ok::<(), Error>(())
            })
        })
        .await
    });
    assert!(
        joined.expect_err("the blocking task panicked").is_panic(),
        "the panic is observed by the caller"
    );

    assert_eq!(
        SpawnPanicked::query()
            .blocking()
            .count(&*db)
            .expect("counts after the panic"),
        0
    );
    SpawnPanicked::query()
        .blocking()
        .insert(&*db, &[("title", "after".into())])
        .expect("the pool still inserts");
    assert_eq!(
        SpawnPanicked::query()
            .blocking()
            .count(&*db)
            .expect("counts"),
        1
    );

    teardown(&db, SpawnPanicked::TABLE);
    db.close().expect("closes");
}