tds-driver 0.2.0

use futures_util::io::{AsyncRead, AsyncWrite};
// use futures_util::stream::TryStreamExt;
use names::{Generator, Name};
use once_cell::sync::Lazy;
use std::cell::RefCell;
use std::env;
use std::sync::Once;

use tds_driver::{numeric::Numeric, Command, Result, TableValueRow};

use runtimes_macro::test_on_runtimes;

// This is used in the testing macro :)
#[allow(dead_code)]
static LOGGER_SETUP: Once = Once::new();

static CONN_STR: Lazy<String> = Lazy::new(|| {
    env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or_else(|_| {
        "server=tcp:localhost,1433;user=SA;password=<YourStrong@Passw0rd>;IntegratedSecurity=true;TrustServerCertificate=true".to_owned()
    })
});

thread_local! {
    static NAMES: RefCell<Option<Generator<'static>>> =
        const { RefCell::new(None) };
}

async fn random_table() -> String {
    NAMES.with(|maybe_generator| {
        maybe_generator
            .borrow_mut()
            .get_or_insert_with(|| Generator::with_naming(Name::Plain))
            .next()
            .unwrap()
            .replace('-', "")
    })
}

#[test_on_runtimes]
async fn basic_proc_exec<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let table = random_table().await;
    let proc = random_table().await;

    conn.simple_query(format!(
        r#"
        create table ##{} (
            id int identity(1,1),
            other varchar(50),
        )
        "#,
        table
    ))
    .await?;

    conn.simple_query(format!(
        r#"
        create or alter procedure {} 
          @Param1 varchar(50)
        as
            insert into ##{} (other)
            values (@Param1)

            return scope_identity()
        "#,
        proc, table,
    ))
    .await?;

    let mut ins_cmd = Command::new(&proc);
    ins_cmd.bind_param("@Param1", "some text");

    let result = ins_cmd.exec(&mut conn).await?.into_command_result().await?;
    assert_eq!(1, result.return_code());

    let mut ins_cmd = Command::new(&proc);
    ins_cmd.bind_param("@Param1", "another text");

    let result = ins_cmd.exec(&mut conn).await?.into_command_result().await?;
    assert_eq!(2, result.return_code());

    Ok(())
}

struct GeoTest {
    id: i32,
    lat: Numeric,
    lon: Numeric,
}

impl<'a> TableValueRow<'a> for GeoTest {
    fn bind_fields(&self, data_row: &mut tds_driver::SqlTableDataRow<'a>) {
        data_row.add_field(self.id);
        data_row.add_field(self.lat);
        data_row.add_field(self.lon);
    }

    fn get_db_type() -> &'static str {
        "GeoTest"
    }
}

#[test_on_runtimes]
async fn tvp_proc_exec<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let table = random_table().await;
    let proc_tvp = random_table().await;
    let proc_ins = random_table().await;
    let proc_get = random_table().await;
    let db_type = random_table().await;

    conn.simple_query(format!(
        r#"
            create type dbo.[{}] as table
            (
                [ID] int not null,
                [lat] decimal(9,6),
                [lon] decimal(9,6)
            )
        "#,
        db_type
    ))
    .await?;

    conn.simple_query(format!(
        r#"
        create table ##{} (
            id int not null,
            lat decimal(9,6),
            lon decimal(9,6),
            n varchar(50)
        )
    "#,
        table
    ))
    .await?;

    conn.simple_query(format!(
        r#"
        create or alter procedure {} 
          @id int,
          @geo dbo.[{}] readonly
        as
        update t set
            [lat] = g.[lat],
            [lon] = g.[lon]
        from
            ##{} t
            inner join @geo g on g.id = t.id

        "#,
        proc_tvp, db_type, table,
    ))
    .await?;

    conn.simple_query(format!(
        r#"
        create or alter procedure {} 
          @id int,
          @name varchar(50)
        as
            insert into ##{} (id, n)
            values (@id, @name)

        "#,
        proc_ins, table,
    ))
    .await?;

    conn.simple_query(format!(
        r#"
        create or alter procedure {} 
          @id int,
          @count int out
        as
          set @count = (select count(*) from ##{})
          select * from ##{}
        "#,
        proc_get, table, table
    ))
    .await?;

    let mut ins_cmd = Command::new(&proc_ins);
    ins_cmd.bind_param("@id", 23);
    ins_cmd.bind_param("@name", "the twenty three");

    let result = ins_cmd.exec(&mut conn).await?.into_command_result().await?;
    assert_eq!(0, result.return_code());

    let g1 = GeoTest {
        id: 23,
        lon: Numeric::new_with_scale(141, 6),
        lat: Numeric::new_with_scale(192, 6),
    };
    let g2 = GeoTest {
        id: 78,
        lon: Numeric::new_with_scale(1141, 6),
        lat: Numeric::new_with_scale(8192, 6),
    };
    let tbl = vec![g1, g2];
    let mut tvp_cmd = Command::new(&proc_tvp);
    tvp_cmd.bind_param("@id", 23);
    tvp_cmd.bind_table_with_dbtype("@geo", &db_type, tbl);

    let result = tvp_cmd.exec(&mut conn).await?.into_command_result().await?;
    assert_eq!(0, result.return_code());

    let count = 0;
    let mut get_cmd = Command::new(&proc_get);
    get_cmd.bind_param("@id", 23);
    get_cmd.bind_out_param("@count", count);

    let result = get_cmd.exec(&mut conn).await?.into_command_result().await?;
    assert_eq!(0, result.return_code());
    let count: i32 = result.try_return_value("@count")?.unwrap();
    assert_eq!(1, count);

    let rows = result.to_query_result(0).unwrap();
    let lat: Numeric = rows[0].get("lat").unwrap();
    let lon: Numeric = rows[0].get("lon").unwrap();
    assert_eq!(Numeric::new_with_scale(141, 6), lon);
    assert_eq!(Numeric::new_with_scale(192, 6), lat);

    Ok(())
}

#[test_on_runtimes]
async fn command_result_keeps_empty_result_sets<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let proc = format!("#{}", random_table().await);
    conn.simple_query(format!(
        "CREATE PROCEDURE {} AS BEGIN \
           SELECT 1 AS a; SELECT 2 AS b WHERE 1 = 0; SELECT 3 AS c WHERE 1 = 0; SELECT 4 AS d \
         END",
        proc
    ))
    .await?
    .into_results()
    .await?;

    let res = Command::new(&proc)
        .exec(&mut conn)
        .await?
        .into_command_result()
        .await?;

    let sets: Vec<Vec<Option<i32>>> = res
        .into_iter()
        .map(|rows| rows.iter().map(|r| r.get::<i32, _>(0)).collect())
        .collect();
    assert_eq!(sets, vec![vec![Some(1)], vec![], vec![], vec![Some(4)]]);

    Ok(())
}

#[test_on_runtimes]
async fn command_negative_return_status<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let proc = format!("#{}", random_table().await);
    conn.simple_query(format!("CREATE PROCEDURE {} AS RETURN -1", proc))
        .await?
        .into_results()
        .await?;

    let res = Command::new(&proc)
        .exec(&mut conn)
        .await?
        .into_command_result()
        .await?;
    assert_eq!(-1, res.return_code());

    Ok(())
}

#[test_on_runtimes]
async fn command_rows_affected_skips_selects<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let table = random_table().await;
    let proc = format!("#{}", random_table().await);
    conn.simple_query(format!("CREATE TABLE ##{} (id int)", table))
        .await?
        .into_results()
        .await?;
    conn.simple_query(format!(
        "CREATE PROCEDURE {p} AS BEGIN \
           INSERT INTO ##{t} VALUES (5), (6); SELECT id FROM ##{t}; \
           DECLARE @y int; SELECT @y = MAX(id) FROM ##{t}; \
           UPDATE ##{t} SET id = id WHERE id = 5 \
         END",
        p = proc,
        t = table
    ))
    .await?
    .into_results()
    .await?;

    let res = Command::new(&proc)
        .exec(&mut conn)
        .await?
        .into_command_result()
        .await?;
    // ExecuteNonQuery on this procedure returns 3.
    assert_eq!(res.rows_affected(), [2, 1]);
    assert_eq!(res.to_query_result(0).map(Vec::len), Some(2));

    Ok(())
}

struct MoneyRow<T> {
    m: T,
}

impl<'a, T: tds_driver::IntoSql<'a> + Copy + 'a> TableValueRow<'a> for MoneyRow<T> {
    fn bind_fields(&self, data_row: &mut tds_driver::SqlTableDataRow<'a>) {
        data_row.add_field(self.m);
    }

    fn get_db_type() -> &'static str {
        "MoneyRow"
    }
}

/// Creates a one-column TVP type `column_type` and a procedure reading it
/// back as text; returns (type, procedure).
async fn money_tvp<S>(
    conn: &mut tds_driver::Client<S>,
    column_type: &str,
) -> Result<(String, String)>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let db_type = random_table().await;
    let proc = random_table().await;

    conn.simple_query(format!(
        "create type dbo.[{db_type}] as table (m {column_type})"
    ))
    .await?;
    conn.simple_query(format!(
        "create or alter procedure {proc} @t dbo.[{db_type}] readonly \
         as select convert(varchar(40), m, 2) from @t"
    ))
    .await?;

    Ok((db_type, proc))
}

#[test_on_runtimes]
async fn tvp_money_is_exact_from_numeric<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let (db_type, proc) = money_tvp(&mut conn, "money").await?;

    // (sent, stored): f64 would turn .5806 into .5807, the saturated maximum.
    let cases = [
        (
            Numeric::new_with_scale(9223372036854775806, 4),
            "922337203685477.5806",
        ),
        (Numeric::new_with_scale(12345, 5), "0.1235"),
        (Numeric::new_with_scale(-12345, 5), "-0.1235"),
    ];

    for (sent, stored) in cases {
        let mut cmd = Command::new(&proc);
        cmd.bind_table_with_dbtype("@t", &db_type, vec![MoneyRow { m: sent }]);
        let result = cmd.exec(&mut conn).await?.into_command_result().await?;
        let rows = result.to_query_result(0).unwrap();
        assert_eq!(Some(stored), rows[0].get::<&str, _>(0), "{sent}");
    }

    // The rounding matches the server's own decimal-to-money conversion.
    let row = conn
        .simple_query(
            "select convert(varchar(40), cast(cast(0.12345 as decimal(10,5)) as money), 2), \
                    convert(varchar(40), cast(cast(-0.12345 as decimal(10,5)) as money), 2)",
        )
        .await?
        .into_row()
        .await?
        .unwrap();
    assert_eq!(Some("0.1235"), row.get::<&str, _>(0));
    assert_eq!(Some("-0.1235"), row.get::<&str, _>(1));

    conn.simple_query(format!("drop procedure {proc}; drop type dbo.[{db_type}]"))
        .await?;

    Ok(())
}

#[test_on_runtimes]
async fn tvp_smallmoney_out_of_range_is_an_error<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let (db_type, proc) = money_tvp(&mut conn, "smallmoney").await?;
    // Cleaned up first: the failed request leaves this connection unusable.
    conn.simple_query(format!("drop procedure {proc}")).await?;

    // Used to be stored as 214748.3647.
    let mut cmd = Command::new(&proc);
    cmd.bind_table_with_dbtype("@t", &db_type, vec![MoneyRow { m: 300000.0f64 }]);
    let err = cmd.exec(&mut conn).await.unwrap_err();
    assert!(
        matches!(err, tds_driver::error::Error::Conversion(_)),
        "{err:?}"
    );

    Ok(())
}

/// Runs `proc_` with `value` bound as its OUT parameter `@x`.
async fn out_param<'a, S>(
    conn: &mut tds_driver::Client<S>,
    proc_: &str,
    value: impl tds_driver::IntoSql<'a> + 'a,
) -> Result<tds_driver::CommandResult>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    let mut cmd = Command::new(proc_.to_string());
    cmd.bind_out_param("@x", value);
    cmd.exec(conn).await?.into_command_result().await
}

#[test_on_runtimes]
async fn out_params_round_trip_null_and_max<S>(mut conn: tds_driver::Client<S>) -> Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    // (declaration, value the procedure sets)
    let procs = [
        (
            "nvarchar(max)",
            "replicate(cast(N'z' as nvarchar(max)), 10000)",
        ),
        ("nvarchar(50)", "replicate(N'q', 40)"),
        (
            "varbinary(max)",
            "cast(replicate(cast('a' as varchar(max)), 9000) as varbinary(max))",
        ),
        ("decimal(10,4)", "12.3456"),
        ("datetime2(7)", "'2024-01-02T03:04:05.1234567'"),
        ("int", "-7"),
        ("xml", "'<a>1</a>'"),
    ];

    let mut names = Vec::new();
    for (decl, value) in procs {
        let name = random_table().await;
        conn.simple_query(format!(
            "create or alter procedure {name} @x {decl} output as set @x = {value}"
        ))
        .await?;
        names.push(name);
    }

    let z = "z".repeat(10000);
    let q = "q".repeat(40);

    // (max) output is not cut to the bound value's declared length.
    for bound in [None, Some("a")] {
        let res = out_param(&mut conn, &names[0], bound).await?;
        assert_eq!(Some(z.as_str()), res.try_return_value::<&str>("@x")?);
        let res = out_param(&mut conn, &names[1], bound).await?;
        assert_eq!(Some(q.as_str()), res.try_return_value::<&str>("@x")?);
    }
    for bound in [None, Some(&[1u8][..])] {
        let res = out_param(&mut conn, &names[2], bound).await?;
        let bytes: Option<&[u8]> = res.try_return_value("@x")?;
        assert_eq!(Some(&[b'a'; 9000][..]), bytes);
    }

    // The scale of a bound Numeric sets the output's scale; NULL gives scale 0.
    let res = out_param(&mut conn, &names[3], Numeric::new_with_scale(0, 4)).await?;
    assert_eq!(
        Some(Numeric::new_with_scale(123456, 4)),
        res.try_return_value("@x")?
    );
    let res = out_param(&mut conn, &names[3], None::<Numeric>).await?;
    assert_eq!(
        Some(Numeric::new_with_scale(12, 0)),
        res.try_return_value("@x")?
    );

    #[cfg(all(feature = "tds73", feature = "chrono"))]
    {
        let res = out_param(&mut conn, &names[4], None::<chrono::NaiveDateTime>).await?;
        let expected = chrono::NaiveDate::from_ymd_opt(2024, 1, 2)
            .unwrap()
            .and_hms_nano_opt(3, 4, 5, 123_456_700)
            .unwrap();
        assert_eq!(Some(expected), res.try_return_value("@x")?);
    }

    let res = out_param(&mut conn, &names[5], None::<i32>).await?;
    assert_eq!(Some(-7i32), res.try_return_value("@x")?);

    let res = out_param(&mut conn, &names[6], None::<tds_driver::xml::XmlData>).await?;
    let xml: Option<&tds_driver::xml::XmlData> = res.try_return_value("@x")?;
    assert_eq!(Some("<a>1</a>"), xml.map(|x| x.as_ref()));

    for name in names {
        conn.simple_query(format!("drop procedure {name}")).await?;
    }

    Ok(())
}