nusadb 0.1.0

//! Integration tests for the `nusadb` driver against a real `nusadb-server`.
//!
//! Boots the compiled `nusadb-server` binary (run `cargo build -p nusadb-server` first) on an
//! ephemeral port with a temporary data dir, then exercises the driver end-to-end. If the binary is
//! not found the test is skipped (prints a notice and returns) rather than failing.

#![allow(clippy::unwrap_used, clippy::expect_used)]

use std::net::{TcpListener, TcpStream};
use std::path::PathBuf;
use std::process::{Child, Command};
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{Duration, Instant};

use nusadb::orm::{Delete, Insert, Select, Update};
use nusadb::{Config, Connection, Pool, TypeTag, Value};

static UNIQ: AtomicU32 = AtomicU32::new(0);

fn server_binary() -> Option<PathBuf> {
    let exe = if cfg!(windows) {
        "nusadb-server.exe"
    } else {
        "nusadb-server"
    };
    let mut bases: Vec<PathBuf> = Vec::new();
    if let Ok(td) = std::env::var("CARGO_TARGET_DIR") {
        bases.push(PathBuf::from(td));
    }
    // Repo target/ relative to this crate (drivers/rust).
    bases.push(PathBuf::from("../../target"));
    for base in bases {
        for profile in ["debug", "release"] {
            let candidate = base.join(profile).join(exe);
            if candidate.exists() {
                return Some(candidate);
            }
        }
    }
    None
}

fn free_port() -> u16 {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    drop(listener);
    port
}

struct Server {
    child: Child,
    port: u16,
    data_dir: PathBuf,
}

impl Server {
    fn start(bin: &PathBuf) -> Self {
        Self::start_with(bin, &[])
    }

    /// Boot the server with extra CLI args (e.g. `--auth-user USER:PASSWORD` to require SCRAM).
    fn start_with(bin: &PathBuf, extra_args: &[&str]) -> Self {
        let port = free_port();
        let uniq = UNIQ.fetch_add(1, Ordering::SeqCst);
        let data_dir =
            std::env::temp_dir().join(format!("turbo_it_{}_{}", std::process::id(), uniq));
        std::fs::create_dir_all(&data_dir).unwrap();
        let child = Command::new(bin)
            .args([
                "--listen",
                &format!("127.0.0.1:{port}"),
                "--data-dir",
                data_dir.to_str().unwrap(),
            ])
            .args(extra_args)
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .expect("spawn nusadb-server");
        let server = Self {
            child,
            port,
            data_dir,
        };
        server.wait_ready();
        server
    }

    fn wait_ready(&self) {
        let deadline = Instant::now() + Duration::from_secs(15);
        while Instant::now() < deadline {
            if TcpStream::connect(("127.0.0.1", self.port)).is_ok() {
                // Give the listener a beat to finish binding before the first real frame.
                std::thread::sleep(Duration::from_millis(150));
                return;
            }
            std::thread::sleep(Duration::from_millis(100));
        }
        panic!("server on port {} did not become ready", self.port);
    }

    fn config(&self) -> Config {
        Config {
            host: "127.0.0.1".to_owned(),
            port: self.port,
            user: "turbo".to_owned(),
            password: None,
            database: "nusadb".to_owned(),
            connect_timeout: Some(Duration::from_secs(5)),
            tls: None,
        }
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
        let _ = std::fs::remove_dir_all(&self.data_dir);
    }
}

#[test]
fn end_to_end() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start(&bin);
    let mut conn = Connection::connect_config(&server.config()).expect("connect");

    // --- DDL + simple query ---
    conn.execute("CREATE TABLE t (id INT NOT NULL, name TEXT, active BOOL, PRIMARY KEY (id))")
        .expect("create table");

    // --- parameterised insert (extended query) ---
    let n = conn
        .query_params(
            "INSERT INTO t VALUES ($1, $2, $3)",
            &[&1_i64, &"alice", &true],
        )
        .expect("insert 1")
        .affected();
    assert_eq!(n, 1, "one row inserted");
    conn.query_params(
        "INSERT INTO t VALUES ($1, $2, $3)",
        &[&2_i64, &"bob", &false],
    )
    .expect("insert 2");

    // --- typed metadata + native decoding (protocol 1.1) ---
    let result = conn
        .query("SELECT id, name, active FROM t ORDER BY id")
        .unwrap();
    assert_eq!(result.columns, ["id", "name", "active"]);
    assert_eq!(
        result.column_types,
        [TypeTag::Int, TypeTag::Text, TypeTag::Bool]
    );
    assert_eq!(result.rows.len(), 2);
    let id: i64 = result.rows[0].get(0).unwrap();
    let name: String = result.rows[0].get(1).unwrap();
    let active: bool = result.rows[0].get(2).unwrap();
    assert_eq!((id, name.as_str(), active), (1, "alice", true));

    // --- NULL round-trips as Option ---
    conn.query_params(
        "INSERT INTO t VALUES ($1, $2, $3)",
        &[&3_i64, &Option::<&str>::None, &true],
    )
    .expect("insert null");
    let r = conn.query("SELECT name FROM t WHERE id = 3").unwrap();
    let name: Option<String> = r.rows[0].get(0).unwrap();
    assert_eq!(name, None);

    // --- prepared statement, reused ---
    let stmt = conn.prepare("SELECT name FROM t WHERE id = $1").unwrap();
    let a = conn.query_prepared(&stmt, &[&1_i64]).unwrap();
    let b = conn.query_prepared(&stmt, &[&2_i64]).unwrap();
    assert_eq!(a.rows[0].get::<String>(0).unwrap(), "alice");
    assert_eq!(b.rows[0].get::<String>(0).unwrap(), "bob");

    // --- transaction: commit then rollback ---
    conn.transaction(|tx| {
        tx.query_params(
            "INSERT INTO t VALUES ($1, $2, $3)",
            &[&10_i64, &"dave", &true],
        )?;
        Ok(())
    })
    .unwrap();
    assert_eq!(
        conn.query("SELECT id FROM t WHERE id = 10")
            .unwrap()
            .rows
            .len(),
        1
    );

    let rolled: nusadb::Result<()> = conn.transaction(|tx| {
        tx.query_params(
            "INSERT INTO t VALUES ($1, $2, $3)",
            &[&11_i64, &"ghost", &false],
        )?;
        // A failing statement propagates its error, which makes `transaction` roll back.
        tx.query("SELECT * FROM definitely_not_a_table")?;
        Ok(())
    });
    assert!(rolled.is_err());
    assert_eq!(
        conn.query("SELECT id FROM t WHERE id = 11")
            .unwrap()
            .rows
            .len(),
        0
    );

    // --- savepoints: partial rollback inside a transaction ---
    conn.execute("CREATE TABLE sp (id INT NOT NULL)").unwrap();
    conn.begin().unwrap();
    conn.execute("INSERT INTO sp VALUES (1)").unwrap();
    conn.savepoint("sp1").unwrap();
    conn.execute("INSERT INTO sp VALUES (2)").unwrap();
    conn.rollback_to_savepoint("sp1").unwrap(); // undoes (2), keeps (1)
    conn.execute("INSERT INTO sp VALUES (3)").unwrap();
    conn.savepoint("sp2").unwrap();
    conn.execute("INSERT INTO sp VALUES (4)").unwrap();
    conn.release_savepoint("sp2").unwrap(); // keeps (4)
    conn.commit().unwrap();
    assert_eq!(
        conn.query("SELECT id FROM sp ORDER BY id")
            .unwrap()
            .rows
            .len(),
        3
    );

    // --- LISTEN/NOTIFY: self-delivery of an async notification ---
    conn.listen("turbo_chan").unwrap();
    conn.notify("turbo_chan", Some("hello")).unwrap();
    let note = conn
        .poll_notification(Some(Duration::from_secs(5)))
        .unwrap()
        .expect("expected a self-delivered notification");
    assert_eq!(note.channel, "turbo_chan");
    assert_eq!(note.payload, "hello");
    // After UNLISTEN, a further NOTIFY is not delivered (poll times out cleanly).
    conn.unlisten("turbo_chan").unwrap();
    conn.notify("turbo_chan", None).unwrap();
    assert!(conn
        .poll_notification(Some(Duration::from_millis(300)))
        .unwrap()
        .is_none());

    // --- server error surfaces and the connection survives ---
    let err = conn.query("SELECT * FROM no_such_table");
    assert!(matches!(err, Err(nusadb::Error::Server { .. })));
    assert_eq!(conn.query("SELECT 1").unwrap().rows.len(), 1);

    // --- advanced query shapes all transport + decode (driver is SQL-text) ---
    for sql in [
        "SELECT id, name FROM t t1 WHERE id IN (SELECT id FROM t WHERE active = true)",
        "WITH ids AS (SELECT id FROM t) SELECT count(*) FROM ids",
        "SELECT id, ROW_NUMBER() OVER (ORDER BY id) FROM t",
        "SELECT id FROM t WHERE id = 1 UNION SELECT id FROM t WHERE id = 2",
    ] {
        conn.query(sql)
            .unwrap_or_else(|e| panic!("advanced query failed: {sql}: {e}"));
    }

    // --- ORM builders ---
    Insert::into("t")
        .set("id", 20_i64)
        .set("name", "orm")
        .set("active", true)
        .run(&mut conn)
        .unwrap();
    let rows = Select::from("t")
        .columns(&["id", "name"])
        .filter("id", 20_i64)
        .limit(1)
        .fetch(&mut conn)
        .unwrap();
    assert_eq!(rows.rows[0].get::<String>(1).unwrap(), "orm");

    Update::table("t")
        .set("name", "orm2")
        .filter("id", 20_i64)
        .run(&mut conn)
        .unwrap();
    let v = Select::from("t")
        .filter("id", 20_i64)
        .fetch(&mut conn)
        .unwrap();
    assert_eq!(v.rows[0].get_by_name::<String>("name").unwrap(), "orm2");

    let deleted = Delete::from("t")
        .filter("id", 20_i64)
        .run(&mut conn)
        .unwrap();
    assert_eq!(deleted.tag, "DELETE 1");

    // --- ORM pagination, distinct, and aggregate terminals (t now holds ids 1, 2, 3, 10) ---
    let page = Select::from("t")
        .columns(&["id"])
        .order_by("id", false)
        .limit(2)
        .offset(1)
        .fetch(&mut conn)
        .unwrap();
    let page_ids: Vec<i64> = page.rows.iter().map(|r| r.get(0).unwrap()).collect();
    assert_eq!(page_ids, vec![2, 3], "LIMIT 2 OFFSET 1 over ordered ids");

    let actives = Select::from("t")
        .distinct()
        .columns(&["active"])
        .order_by("active", false)
        .fetch(&mut conn)
        .unwrap();
    let active_flags: Vec<bool> = actives.rows.iter().map(|r| r.get(0).unwrap()).collect();
    assert_eq!(active_flags, vec![false, true], "DISTINCT active");

    assert_eq!(Select::from("t").count(&mut conn).unwrap(), 4);
    assert_eq!(
        Select::from("t")
            .where_raw("active = true", &[])
            .count(&mut conn)
            .unwrap(),
        3,
        "count honours WHERE"
    );
    assert_eq!(
        Select::from("t")
            .min("id", &mut conn)
            .unwrap()
            .and_then(|v| v.as_i64()),
        Some(1)
    );
    assert_eq!(
        Select::from("t")
            .max("id", &mut conn)
            .unwrap()
            .and_then(|v| v.as_i64()),
        Some(10)
    );
    assert_eq!(
        Select::from("t")
            .sum("id", &mut conn)
            .unwrap()
            .and_then(|v| v.as_i64()),
        Some(16)
    );
    assert_eq!(Select::from("t").avg("id", &mut conn).unwrap(), Some(4.0));

    // --- ORM WHERE operators (t = {1:alice/true, 2:bob/false, 3:NULL/true, 10:dave/true}) ---
    let rows = |s: nusadb::orm::Select, c: &mut Connection| s.fetch(c).unwrap().rows.len();
    assert_eq!(
        rows(Select::from("t").gt("id", 2_i64), &mut conn),
        2,
        "id > 2 -> {{3,10}}"
    );
    assert_eq!(
        rows(Select::from("t").where_in("id", &[1_i64, 10]), &mut conn),
        2
    );
    assert_eq!(
        rows(
            Select::from("t").where_not_in("id", &[1_i64, 2, 3]),
            &mut conn
        ),
        1
    );
    assert_eq!(
        rows(Select::from("t").between("id", 2_i64, 3), &mut conn),
        2
    );
    assert_eq!(
        rows(Select::from("t").is_null("name"), &mut conn),
        1,
        "only id 3 has NULL name"
    );
    assert_eq!(rows(Select::from("t").is_not_null("name"), &mut conn), 3);
    assert_eq!(
        rows(Select::from("t").like("name", "a%"), &mut conn),
        1,
        "alice"
    );

    // --- GROUP BY + HAVING with a raw aggregate projection ---
    let grouped = Select::from("t")
        .select_raw(&["active", "count(*) AS n"])
        .group_by(&["active"])
        .having("count(*) >= $1", &[Value::Int(2)])
        .fetch(&mut conn)
        .unwrap();
    assert_eq!(
        grouped.rows.len(),
        1,
        "only the active=true group (3 rows) clears HAVING >= 2"
    );
    assert!(grouped.rows[0].get::<bool>(0).unwrap());
    assert_eq!(grouped.rows[0].get::<i64>(1).unwrap(), 3);

    // --- UNION combines two queries, renumbering the right operand's params ---
    let union = Select::from("t")
        .columns(&["id"])
        .filter("id", 1_i64)
        .union(Select::from("t").columns(&["id"]).filter("id", 2_i64))
        .order_by("id", false)
        .fetch(&mut conn)
        .unwrap();
    let union_ids: Vec<i64> = union.rows.iter().map(|r| r.get(0).unwrap()).collect();
    assert_eq!(union_ids, vec![1, 2]);

    // --- a sampling of value types decode to the right native form ---
    let typed = conn
        .query("SELECT true, 42, 1.5::float8, 'hi', DATE '2020-01-02', ARRAY[1,2,3]")
        .unwrap();
    let row = &typed.rows[0];
    assert!(row.get::<bool>(0).unwrap());
    assert_eq!(row.get::<i64>(1).unwrap(), 42);
    assert_eq!(row.get::<f64>(2).unwrap(), 1.5);
    assert_eq!(row.get::<String>(3).unwrap(), "hi");
    assert!(matches!(
        row.value(4),
        Some(Value::Typed {
            tag: TypeTag::Date,
            ..
        })
    ));
    assert!(matches!(
        row.value(5),
        Some(Value::Typed {
            tag: TypeTag::Array,
            ..
        })
    ));

    conn.close();

    // --- connection pool ---
    let pool = Pool::new(server.config(), 4).unwrap();
    let mut c1 = pool.get().unwrap();
    assert_eq!(c1.query("SELECT count(*) FROM t").unwrap().rows.len(), 1);
    let mut c2 = pool.get().unwrap();
    assert_eq!(c2.execute("SELECT 1").unwrap(), 1);
    drop(c1);
    drop(c2);
}

/// Proves the driver transports the full SELECT surface (ORDER BY / DISTINCT / LIMIT / windows /
/// CTE / subquery / set ops / every JOIN flavour / LATERAL) to a real server via the raw text query
/// API. The driver is a SQL-text transport, so any query the server accepts decodes back as typed
/// rows — this guards that none of those shapes regress at the wire/decode boundary.
#[test]
fn query_surface() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start(&bin);
    let mut conn = Connection::connect_config(&server.config()).expect("connect");

    // --- fixture ---
    conn.execute("CREATE TABLE surf_a (id INT NOT NULL, grp TEXT, v INT)")
        .expect("create surf_a");
    for (id, grp, v) in [
        (1, "a", 10),
        (2, "a", 30),
        (3, "b", 20),
        (4, "b", 20),
        (5, "a", 10),
    ] {
        conn.query_params(
            "INSERT INTO surf_a VALUES ($1, $2, $3)",
            &[&(id as i64), &grp, &(v as i64)],
        )
        .unwrap_or_else(|e| panic!("seed surf_a ({id}): {e}"));
    }
    conn.execute("CREATE TABLE surf_b (id INT NOT NULL, a_id INT, tag TEXT)")
        .expect("create surf_b");
    for (id, a_id, tag) in [(10, 1, "p"), (11, 1, "q"), (12, 2, "r")] {
        conn.query_params(
            "INSERT INTO surf_b VALUES ($1, $2, $3)",
            &[&(id as i64), &(a_id as i64), &tag],
        )
        .unwrap_or_else(|e| panic!("seed surf_b ({id}): {e}"));
    }

    // Each entry: (label, sql, expected row count). Run via the raw text query API.
    let cases: &[(&str, &str, usize)] = &[
        ("A ORDER BY", "SELECT id FROM surf_a ORDER BY v DESC, id", 5),
        ("B DISTINCT", "SELECT DISTINCT v FROM surf_a ORDER BY v", 3),
        (
            "C DISTINCT ON",
            "SELECT DISTINCT ON (grp) grp, v FROM surf_a ORDER BY grp, v",
            2,
        ),
        ("D LIMIT", "SELECT id FROM surf_a ORDER BY id LIMIT 2", 2),
        (
            "E OFFSET",
            "SELECT id FROM surf_a ORDER BY id LIMIT 2 OFFSET 3",
            2,
        ),
        (
            "F GROUP/HAVING",
            "SELECT grp, count(*) FROM surf_a GROUP BY grp HAVING count(*) > 1 ORDER BY grp",
            2,
        ),
        (
            "G window",
            "SELECT id, row_number() OVER (PARTITION BY grp ORDER BY v) FROM surf_a ORDER BY id",
            5,
        ),
        (
            "H CTE",
            "WITH g AS (SELECT grp, count(*) c FROM surf_a GROUP BY grp) SELECT count(*) FROM g",
            1,
        ),
        (
            "I subquery IN",
            "SELECT id FROM surf_a WHERE v IN (SELECT max(v) FROM surf_a) ORDER BY id",
            1,
        ),
        (
            "J UNION",
            "SELECT id FROM surf_a WHERE id = 1 UNION SELECT id FROM surf_a WHERE id = 2",
            2,
        ),
        (
            "K INNER JOIN",
            "SELECT surf_a.grp, surf_b.tag FROM surf_a JOIN surf_b ON surf_a.id = surf_b.a_id \
             ORDER BY surf_b.id",
            3,
        ),
        (
            "L LEFT JOIN",
            "SELECT surf_a.grp, surf_b.tag FROM surf_a LEFT JOIN surf_b ON surf_a.id = surf_b.a_id \
             ORDER BY surf_a.id, surf_b.id",
            6,
        ),
        (
            "M RIGHT JOIN",
            "SELECT surf_a.grp, surf_b.tag FROM surf_a RIGHT JOIN surf_b \
             ON surf_a.id = surf_b.a_id ORDER BY surf_b.id",
            3,
        ),
        (
            "N FULL JOIN",
            "SELECT surf_a.grp, surf_b.tag FROM surf_a FULL JOIN surf_b ON surf_a.id = surf_b.a_id \
             ORDER BY surf_a.id, surf_b.id",
            6,
        ),
        (
            "O CROSS JOIN",
            "SELECT surf_a.id, surf_b.id FROM surf_a CROSS JOIN surf_b",
            15,
        ),
        (
            "P LATERAL",
            "SELECT surf_a.id, l.tag FROM surf_a JOIN LATERAL \
             (SELECT tag FROM surf_b WHERE surf_b.a_id = surf_a.id LIMIT 1) l ON true \
             ORDER BY surf_a.id",
            2,
        ),
    ];

    for (label, sql, want) in cases {
        let r = conn
            .query(sql)
            .unwrap_or_else(|e| panic!("[{label}] query errored: {sql}\n  -> {e}"));
        assert_eq!(
            r.rows.len(),
            *want,
            "[{label}] row count mismatch for: {sql}"
        );
    }

    conn.close();
}

/// Regression for the SCRAM auth loop dropping out one frame early (reported by R&I as user-QA
/// D2): the auth loop must `break` once `scram` completes the SASL exchange, rather than reading
/// the post-auth `BackendKeyData` ('K') as another auth message. The trust-mode tests above never
/// exercise SCRAM, so this boots a server that *requires* it.
#[test]
fn scram_auth() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start_with(&bin, &["--auth-user", "turbo:s3cret"]);

    // Correct password authenticates and the connection is immediately usable — this is the path
    // that failed with "unexpected message 'K' during auth" before the fix.
    let mut good = server.config();
    good.password = Some("s3cret".to_owned());
    let mut conn = Connection::connect_config(&good).expect("SCRAM auth should succeed");
    assert_eq!(conn.query("SELECT 1").unwrap().rows.len(), 1);
    conn.close();

    // Wrong password is rejected.
    let mut bad = server.config();
    bad.password = Some("wrong".to_owned());
    assert!(
        Connection::connect_config(&bad).is_err(),
        "wrong password must not authenticate"
    );

    // A server requiring auth rejects a passwordless connection rather than hanging.
    assert!(
        Connection::connect_config(&server.config()).is_err(),
        "missing password must not authenticate"
    );
}

#[test]
fn config_url_parsing() {
    let cfg = Config::from_url("nusadb://alice:secret@db.example:6000/shop").unwrap();
    assert_eq!(cfg.user, "alice");
    assert_eq!(cfg.password.as_deref(), Some("secret"));
    assert_eq!(cfg.host, "db.example");
    assert_eq!(cfg.port, 6000);
    assert_eq!(cfg.database, "shop");

    let bare = Config::from_url("nusadb://127.0.0.1:5678/nusadb").unwrap();
    assert_eq!(bare.user, "nusa-root"); // default
    assert_eq!(bare.password.as_deref(), Some("nusa-root")); // default
    assert_eq!(bare.host, "127.0.0.1");
    assert_eq!(bare.port, 5678);

    assert!(Config::from_url("other://x").is_err());
}

#[test]
fn copy_in_and_out() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start(&bin);
    let mut conn = Connection::connect_config(&server.config()).expect("connect");

    conn.execute("CREATE TABLE copy_t (id INT NOT NULL, name TEXT, PRIMARY KEY (id))")
        .expect("create table");

    // COPY FROM STDIN: tab-delimited rows, `\N` for NULL (the server's default text format).
    let data = "1\talice\n2\t\\N\n3\tcarol\n";
    let mut src = std::io::Cursor::new(data.as_bytes().to_vec());
    let loaded = conn
        .copy_in("COPY copy_t FROM STDIN", &mut src)
        .expect("copy in");
    assert_eq!(loaded, 3, "COPY FROM loaded three rows");

    // The rows landed and the NULL round-tripped.
    let rows = conn
        .query("SELECT id, name FROM copy_t ORDER BY id")
        .expect("select");
    assert_eq!(rows.rows.len(), 3);
    assert_eq!(rows.rows[0].get::<String>(1).unwrap(), "alice");
    assert_eq!(
        rows.rows[1].get::<Option<String>>(1).unwrap(),
        None,
        "row 2's name is NULL"
    );
    assert_eq!(rows.rows[2].get::<String>(1).unwrap(), "carol");

    // COPY TO STDOUT: bytes come back in the same text format.
    let mut out = Vec::new();
    let exported = conn
        .copy_out("COPY copy_t TO STDOUT", &mut out)
        .expect("copy out");
    assert_eq!(exported, 3, "COPY TO exported three rows");
    let text = String::from_utf8(out).expect("utf8");
    let lines: Vec<&str> = text.lines().collect();
    assert_eq!(lines, vec!["1\talice", "2\t\\N", "3\tcarol"]);

    // A COPY the server refuses surfaces as an error, and the connection stays usable afterwards.
    let mut empty = std::io::Cursor::new(Vec::new());
    assert!(
        conn.copy_in("COPY no_such_table FROM STDIN", &mut empty)
            .is_err(),
        "COPY into a missing table errors"
    );
    let again = conn
        .query("SELECT count(*) FROM copy_t")
        .expect("connection usable after a refused COPY");
    assert_eq!(again.rows.len(), 1);

    conn.close();
}

#[test]
fn bytea_round_trip() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start(&bin);
    let mut conn = Connection::connect_config(&server.config()).expect("connect");

    conn.execute("CREATE TABLE bt (id INT NOT NULL, data BYTEA, PRIMARY KEY (id))")
        .expect("create table");

    // Bind raw bytes (sent as the BYTEA text form \x<hex>); read them back losslessly.
    let payload: Vec<u8> = vec![0xde, 0xad, 0xbe, 0xef, 0x00, 0x7f];
    conn.query_params("INSERT INTO bt VALUES ($1, $2)", &[&1_i64, &payload])
        .expect("insert bytes");
    conn.query_params(
        "INSERT INTO bt VALUES ($1, $2)",
        &[&2_i64, &Vec::<u8>::new()],
    )
    .expect("insert empty");
    conn.query_params(
        "INSERT INTO bt VALUES ($1, $2)",
        &[&3_i64, &Option::<Vec<u8>>::None],
    )
    .expect("insert null");

    let rows = conn
        .query("SELECT id, data FROM bt ORDER BY id")
        .expect("select");
    assert_eq!(rows.rows[0].get::<Vec<u8>>(1).unwrap(), payload);
    assert_eq!(rows.rows[1].get::<Vec<u8>>(1).unwrap(), Vec::<u8>::new());
    assert_eq!(rows.rows[2].get::<Option<Vec<u8>>>(1).unwrap(), None);

    conn.close();
}