tdsql 0.9.0

//! Blocking-client integration tests.
//!
//! ```text
//! cargo test --features blocking --test blocking_test -- --ignored
//! ```

#![cfg(feature = "blocking")]

use tdsql::blocking::Client;
use tdsql::{Command, Config, Error, IsolationLevel};

fn test_config() -> Config {
    Config::new()
        .host("localhost")
        .port(1433)
        .database("master")
        .auth("sa", "YourStrong!Passw0rd")
        .trust_cert()
}

fn connect() -> Client {
    Client::connect(&test_config())
        .expect("failed to connect; is SQL Server running on localhost:1433?")
}

fn fresh_table(client: &mut Client, name: &str) {
    client
        .batch(&format!(
            "IF OBJECT_ID('dbo.{name}', 'U') IS NOT NULL DROP TABLE dbo.{name};
             CREATE TABLE dbo.{name} (id INT PRIMARY KEY)"
        ))
        .unwrap();
}

fn ids(client: &mut Client, name: &str) -> Vec<i32> {
    client
        .query(&format!("SELECT id FROM dbo.{name} ORDER BY id"), &[])
        .unwrap()
        .iter()
        .map(|r| r.get::<i32, _>("id"))
        .collect()
}

fn drop_table(client: &mut Client, name: &str) {
    client
        .batch(&format!("DROP TABLE IF EXISTS dbo.{name}"))
        .unwrap();
}

// No #[tokio::test] here: these are ordinary synchronous tests, which is the
// entire point of the blocking client.
#[test]
#[ignore]
fn basic_query() {
    let mut client = connect();
    let rows = client.query("SELECT 1 AS value", &[]).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].get::<i32, _>("value"), 1);
}

#[test]
#[ignore]
fn positional_params_and_typed_reads() {
    let mut client = connect();
    let row = client
        .query_one("SELECT @P1 AS n, @P2 AS s", &[&42i32, &"hello"])
        .unwrap();
    assert_eq!(row.get::<i32, _>("n"), 42);
    assert_eq!(row.get::<String, _>("s"), "hello");
}

#[test]
#[ignore]
fn scalar_and_execute() {
    let mut client = connect();
    let n: i32 = client.query_scalar("SELECT 7", &[]).unwrap();
    assert_eq!(n, 7);

    fresh_table(&mut client, "blk_exec");
    let affected = client
        .execute("INSERT INTO dbo.blk_exec (id) VALUES (1), (2), (3)", &[])
        .unwrap();
    assert_eq!(affected, 3);
    assert_eq!(ids(&mut client, "blk_exec"), vec![1, 2, 3]);
    drop_table(&mut client, "blk_exec");
}

#[test]
#[ignore]
fn query_one_rejects_wrong_row_counts() {
    let mut client = connect();
    let err = client
        .query_one("SELECT 1 AS v WHERE 1 = 0", &[])
        .unwrap_err();
    assert!(
        matches!(err, Error::UnexpectedRowCount { found: 0 }),
        "{err}"
    );
}

#[test]
#[ignore]
fn multiple_result_sets() {
    let mut client = connect();
    let ds = client
        .query_dataset(&Command::query("SELECT 1 AS a; SELECT 2 AS b"))
        .unwrap();
    assert_eq!(ds.len(), 2);
    assert_eq!(ds[0][0].get::<i32, _>("a"), 1);
    assert_eq!(ds[1][0].get::<i32, _>("b"), 2);
}

#[test]
#[ignore]
fn stored_procedure() {
    let mut client = connect();
    client
        .batch("IF OBJECT_ID('dbo.blk_sp', 'P') IS NOT NULL DROP PROCEDURE dbo.blk_sp")
        .unwrap();
    client
        .batch("CREATE PROCEDURE dbo.blk_sp @x INT AS BEGIN SELECT @x + 1 AS result; END")
        .unwrap();

    let ds = client
        .query_dataset(&Command::stored_procedure("dbo.blk_sp").param("x", 5))
        .unwrap();
    assert_eq!(ds[0][0].get::<i32, _>("result"), 6);

    client.batch("DROP PROCEDURE dbo.blk_sp").unwrap();
}

#[test]
#[ignore]
fn transaction_commits() {
    let mut client = connect();
    fresh_table(&mut client, "blk_tx_commit");

    let mut tx = client.transaction().unwrap();
    tx.execute("INSERT INTO dbo.blk_tx_commit (id) VALUES (@P1)", &[&1i32])
        .unwrap();
    tx.commit().unwrap();

    assert_eq!(ids(&mut client, "blk_tx_commit"), vec![1]);
    drop_table(&mut client, "blk_tx_commit");
}

/// The same guarantee as the async client: dropping never commits.
#[test]
#[ignore]
fn dropping_a_transaction_rolls_back() {
    let mut client = connect();
    fresh_table(&mut client, "blk_tx_drop");

    {
        let mut tx = client.transaction().unwrap();
        tx.execute("INSERT INTO dbo.blk_tx_drop (id) VALUES (@P1)", &[&1i32])
            .unwrap();
    }

    assert!(
        ids(&mut client, "blk_tx_drop").is_empty(),
        "dropping a blocking transaction must not commit it"
    );
    drop_table(&mut client, "blk_tx_drop");
}

#[test]
#[ignore]
fn error_path_rolls_back() {
    let mut client = connect();
    fresh_table(&mut client, "blk_tx_err");

    fn unit_of_work(client: &mut Client) -> tdsql::Result<()> {
        let mut tx = client.transaction()?;
        tx.execute("INSERT INTO dbo.blk_tx_err (id) VALUES (@P1)", &[&1i32])?;
        tx.execute("INSERT INTO dbo.blk_tx_err (id) VALUES (@P1)", &[&1i32])?;
        tx.commit()
    }

    assert!(unit_of_work(&mut client).is_err());
    assert!(ids(&mut client, "blk_tx_err").is_empty());
    drop_table(&mut client, "blk_tx_err");
}

#[test]
#[ignore]
fn savepoints_and_isolation() {
    let mut client = connect();
    fresh_table(&mut client, "blk_tx_sp");

    let mut tx = client
        .transaction_with_isolation(IsolationLevel::Serializable)
        .unwrap();
    tx.execute("INSERT INTO dbo.blk_tx_sp (id) VALUES (@P1)", &[&1i32])
        .unwrap();
    {
        let mut sp = tx.savepoint("inner").unwrap();
        sp.execute("INSERT INTO dbo.blk_tx_sp (id) VALUES (@P1)", &[&2i32])
            .unwrap();
        sp.rollback().unwrap();
    }
    tx.commit().unwrap();

    assert_eq!(ids(&mut client, "blk_tx_sp"), vec![1]);
    drop_table(&mut client, "blk_tx_sp");
}

#[test]
#[ignore]
fn client_survives_a_dropped_transaction() {
    let mut client = connect();
    {
        let mut tx = client.transaction().unwrap();
        tx.query_scalar::<i32>("SELECT 1", &[]).unwrap();
    }
    let n: i32 = client.query_scalar("SELECT 42", &[]).unwrap();
    assert_eq!(n, 42);
}

#[test]
#[ignore]
fn connects_from_a_connection_string() {
    let mut client = Client::connect_str(
        "Server=tcp:localhost,1433;Database=master;User Id=sa;\
         Password=YourStrong!Passw0rd;TrustServerCertificate=true",
    )
    .unwrap();
    let n: i32 = client.query_scalar("SELECT 1", &[]).unwrap();
    assert_eq!(n, 1);
}

#[test]
#[ignore]
fn streams_rows_as_an_iterator() {
    let mut client = connect();
    let rows = client
        .query_stream(
            "SELECT n FROM (VALUES (1), (2), (3)) AS v (n) ORDER BY n",
            &[],
        )
        .unwrap();

    assert_eq!(rows.columns()[0].name(), "n");
    let ns: Vec<i32> = rows.map(|row| row.unwrap().get("n")).collect();
    assert_eq!(ns, [1, 2, 3]);
}

#[test]
#[ignore]
fn dropping_a_row_iterator_early_leaves_the_connection_clean() {
    let mut client = connect();
    {
        let mut rows = client
            .query_stream(
                "SELECT TOP (5000) 1 AS n FROM sys.all_columns a CROSS JOIN sys.all_columns b",
                &[],
            )
            .unwrap();
        rows.next().unwrap().unwrap();
    }
    let n: i32 = client.query_scalar("SELECT 42", &[]).unwrap();
    assert_eq!(n, 42);
}

#[test]
#[ignore]
fn dropping_a_row_iterator_early_cancels_its_query() {
    let mut client = connect();
    let started = std::time::Instant::now();
    {
        // Minutes' worth of rows if it were read to the end instead.
        let mut rows = client
            .query_stream(
                "SELECT TOP (1000000000) 1 AS n \
                 FROM sys.all_columns a CROSS JOIN sys.all_columns b CROSS JOIN sys.all_columns c",
                &[],
            )
            .unwrap();
        rows.next().unwrap().unwrap();
    }
    let n: i32 = client.query_scalar("SELECT 42", &[]).unwrap();
    assert_eq!(n, 42);

    let took = started.elapsed();
    assert!(
        took < std::time::Duration::from_secs(10),
        "the abandoned query held the connection for {took:?}"
    );
}

#[test]
#[ignore]
fn streams_through_the_blocking_executor_trait() {
    use tdsql::blocking::Executor;

    fn count(db: &mut impl Executor) -> usize {
        db.query_stream("SELECT n FROM (VALUES (1), (2)) AS v (n)", &[])
            .unwrap()
            .count()
    }

    let mut client = connect();
    assert_eq!(count(&mut client), 2);
    let mut tx = client.transaction().unwrap();
    assert_eq!(count(&mut tx), 2);
    tx.rollback().unwrap();
}

/// Creating a blocking client inside an async runtime must fail cleanly rather
/// than panic later with "Cannot start a runtime from within a runtime".
#[tokio::test]
async fn refuses_to_run_inside_an_async_runtime() {
    let err = Client::connect(&test_config())
        .expect_err("should refuse to build a runtime inside a runtime");
    assert!(matches!(err, Error::BlockingInAsync), "{err}");
}

/// A client connected on a plain thread and dropped inside async code must
/// not panic with "Cannot drop a runtime in a context where blocking is not
/// allowed".
#[tokio::test]
#[ignore]
async fn dropping_inside_an_async_runtime_is_fine() {
    let client = std::thread::spawn(|| Client::connect(&test_config()).unwrap())
        .join()
        .unwrap();
    drop(client);
}

/// Inside `spawn_blocking` a client connected elsewhere works, transaction
/// and drop included.
#[tokio::test]
#[ignore]
async fn works_inside_spawn_blocking() {
    let mut client = std::thread::spawn(|| Client::connect(&test_config()).unwrap())
        .join()
        .unwrap();
    let n = tokio::task::spawn_blocking(move || {
        let mut tx = client.transaction().unwrap();
        let n: i32 = tx.query_scalar("SELECT 7", &[]).unwrap();
        tx.commit().unwrap();
        n
    })
    .await
    .unwrap();
    assert_eq!(n, 7);
}