#![cfg(feature = "postgres")]
use dbnexus::{DbConfig, DbPool};
use testcontainers::GenericImage;
use testcontainers::core::{ContainerAsync, ImageExt, WaitFor};
use testcontainers::runners::AsyncRunner;
async fn setup_postgres() -> (ContainerAsync<GenericImage>, String) {
let container = GenericImage::new("postgres", "16-alpine")
.with_wait_for(WaitFor::message_on_stderr(
"database system is ready to accept connections",
))
.with_wait_for(WaitFor::seconds(2))
.with_env_var("POSTGRES_USER", "dbnexus")
.with_env_var("POSTGRES_PASSWORD", "dbnexus")
.with_env_var("POSTGRES_DB", "dbnexus_test")
.start()
.await
.expect("Failed to start PostgreSQL container");
let host = container.get_host().await.expect("Failed to get host");
let port = container
.get_host_port_ipv4(5432)
.await
.expect("Failed to get host port");
let url = format!("postgres://dbnexus:dbnexus@{}:{}/dbnexus_test", host, port);
(container, url)
}
fn make_config(url: String) -> DbConfig {
DbConfig {
url,
admin_role: "admin".to_string(),
pool_config: dbnexus::foundation::PoolConfig {
max_connections: 5,
min_connections: 1,
idle_timeout: 300,
acquire_timeout: 5000,
},
..Default::default()
}
}
#[tokio::test]
async fn test_postgres_connection() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
assert_eq!(session.role(), "admin");
let status = pool.status();
assert!(
status.total >= 1,
"Pool should have at least one connection, got total={}",
status.total
);
assert_eq!(
status.total,
status.active + status.idle,
"Total should equal active + idle"
);
}
#[tokio::test]
async fn test_postgres_crud_insert() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl(
"CREATE TABLE users (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL UNIQUE
)",
)
.await
.expect("Failed to create table");
let result = session
.execute_raw("INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com')")
.await
.expect("Failed to insert");
assert_eq!(result.rows_affected(), 1, "Should insert 1 row");
}
#[tokio::test]
async fn test_postgres_crud_update_delete() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl(
"CREATE TABLE products (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
stock INTEGER NOT NULL DEFAULT 0
)",
)
.await
.expect("Failed to create table");
session
.execute_raw("INSERT INTO products (name, stock) VALUES ('Widget', 10)")
.await
.expect("Failed to insert");
let update_result = session
.execute_raw("UPDATE products SET stock = 5 WHERE name = 'Widget'")
.await
.expect("Failed to update");
assert_eq!(update_result.rows_affected(), 1, "Should update 1 row");
let delete_result = session
.execute_raw("DELETE FROM products WHERE name = 'Widget'")
.await
.expect("Failed to delete");
assert_eq!(delete_result.rows_affected(), 1, "Should delete 1 row");
}
#[tokio::test]
async fn test_postgres_transaction_rollback() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl(
"CREATE TABLE accounts (
id SERIAL PRIMARY KEY,
email VARCHAR(200) NOT NULL UNIQUE
)",
)
.await
.expect("Failed to create table");
session.begin_transaction().await.expect("Failed to begin transaction");
session
.execute_raw("INSERT INTO accounts (email) VALUES ('bob@example.com')")
.await
.expect("Failed to insert in transaction");
session.rollback().await.expect("Failed to rollback");
assert!(
!session.is_in_transaction().await,
"Should not be in transaction after rollback"
);
let result = session
.execute_raw("INSERT INTO accounts (email) VALUES ('bob@example.com')")
.await
.expect("Failed to insert after rollback");
assert_eq!(
result.rows_affected(),
1,
"Insert should succeed after rollback (data was not committed)"
);
}
#[tokio::test]
async fn test_postgres_transaction_commit() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl(
"CREATE TABLE orders (
id SERIAL PRIMARY KEY,
order_no VARCHAR(100) NOT NULL UNIQUE
)",
)
.await
.expect("Failed to create table");
session.begin_transaction().await.expect("Failed to begin transaction");
session
.execute_raw("INSERT INTO orders (order_no) VALUES ('ORD-001')")
.await
.expect("Failed to insert in transaction");
session.commit().await.expect("Failed to commit");
assert!(
!session.is_in_transaction().await,
"Should not be in transaction after commit"
);
let conflict_result = session
.execute_raw("INSERT INTO orders (order_no) VALUES ('ORD-001') ON CONFLICT (order_no) DO NOTHING")
.await
.expect("Failed to execute conflict insert");
assert_eq!(
conflict_result.rows_affected(),
0,
"Insert should conflict with committed data (rows_affected=0)"
);
}
#[tokio::test]
async fn test_postgres_data_types() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl(
"CREATE TABLE type_test (
id SERIAL PRIMARY KEY,
bool_col BOOLEAN,
int_col INTEGER,
bigint_col BIGINT,
float_col REAL,
double_col DOUBLE PRECISION,
text_col TEXT,
varchar_col VARCHAR(100),
uuid_col UUID,
json_col JSONB,
array_col INTEGER[]
)",
)
.await
.expect("Failed to create table");
session
.execute_raw(
"INSERT INTO type_test (bool_col, int_col, bigint_col, float_col, double_col,
text_col, varchar_col, uuid_col, json_col, array_col)
VALUES (true, 42, 9223372036854775807, 3.14, 2.718281828,
'text value', 'varchar value', 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11',
'{\"key\": \"value\"}', '{1,2,3}')",
)
.await
.expect("Failed to insert");
let result = session
.execute_raw("SELECT * FROM type_test WHERE id = 1")
.await
.expect("Failed to query");
assert_eq!(result.rows_affected(), 1, "Should return 1 row");
}
#[tokio::test]
async fn test_postgres_null_handling() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl("CREATE TABLE null_test (id SERIAL PRIMARY KEY, nullable_col VARCHAR(100))")
.await
.expect("Failed to create table");
session
.execute_raw("INSERT INTO null_test (nullable_col) VALUES (NULL)")
.await
.expect("Failed to insert NULL");
let result = session
.execute_raw("SELECT nullable_col FROM null_test WHERE id = 1")
.await
.expect("Failed to query");
assert_eq!(result.rows_affected(), 1, "Should return 1 row");
}
#[tokio::test]
async fn test_postgres_syntax_error_returns_error() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
let result = session.execute_raw("SELEC * FORM nonexistent").await;
assert!(result.is_err(), "Syntax error should return error");
}
#[tokio::test]
async fn test_postgres_table_not_exists_returns_error() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
let result = session.execute_raw("SELECT * FROM nonexistent_table").await;
assert!(result.is_err(), "Query on nonexistent table should return error");
}
#[tokio::test]
async fn test_postgres_duplicate_key_returns_error() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl("CREATE TABLE pk_test (id INTEGER PRIMARY KEY, value VARCHAR(50))")
.await
.expect("Failed to create table");
session
.execute_raw("INSERT INTO pk_test (id, value) VALUES (1, 'first')")
.await
.expect("First insert should succeed");
let result = session
.execute_raw("INSERT INTO pk_test (id, value) VALUES (1, 'duplicate')")
.await;
assert!(result.is_err(), "Duplicate primary key should return error");
}
#[tokio::test]
async fn test_postgres_aggregate_query() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl("CREATE TABLE sales (id SERIAL PRIMARY KEY, product VARCHAR(50), amount DECIMAL(10,2))")
.await
.expect("Failed to create table");
session
.execute_raw("INSERT INTO sales (product, amount) VALUES ('A', 100.50), ('A', 200.00), ('B', 50.00)")
.await
.expect("Failed to insert");
let result = session
.execute_raw(
"SELECT product, COUNT(*) as cnt, SUM(amount) as total FROM sales GROUP BY product ORDER BY product",
)
.await
.expect("Aggregate query should succeed");
assert_eq!(result.rows_affected(), 2, "Should return 2 groups");
}
#[tokio::test]
async fn test_postgres_join_query() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await.expect("Failed to get session");
session
.execute_raw_ddl("CREATE TABLE customers (id SERIAL PRIMARY KEY, name VARCHAR(100))")
.await
.expect("Failed to create customers table");
session
.execute_raw_ddl("CREATE TABLE orders (id SERIAL PRIMARY KEY, customer_id INTEGER REFERENCES customers(id), total DECIMAL(10,2))")
.await
.expect("Failed to create orders table");
session
.execute_raw("INSERT INTO customers (name) VALUES ('Alice'), ('Bob')")
.await
.expect("Failed to insert customers");
session
.execute_raw("INSERT INTO orders (customer_id, total) VALUES (1, 100.00), (1, 200.00), (2, 50.00)")
.await
.expect("Failed to insert orders");
let result = session
.execute_raw(
"SELECT c.name, COUNT(o.id) as order_count, SUM(o.total) as total_spent
FROM customers c LEFT JOIN orders o ON c.id = o.customer_id
GROUP BY c.name ORDER BY c.name",
)
.await
.expect("JOIN query should succeed");
assert_eq!(result.rows_affected(), 2, "Should return 2 customers");
}
#[tokio::test]
async fn test_postgres_health_check() {
let (_container, url) = setup_postgres().await;
let pool = DbPool::with_config(make_config(url))
.await
.expect("Failed to create pool");
let session = pool.get_session("admin").await;
assert!(session.is_ok(), "get_session should succeed with healthy connection");
let session = session.unwrap();
let result = session.execute_raw("SELECT 1").await;
assert!(result.is_ok(), "Health check query should succeed");
}
#[tokio::test(flavor = "multi_thread")]
async fn test_postgres_concurrent_access() {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
let (_container, url) = setup_postgres().await;
let config = make_config(url);
let pool = Arc::new(DbPool::with_config(config).await.expect("Failed to create pool"));
{
let session = pool.get_session("admin").await.expect("Failed to get setup session");
session
.execute_raw_ddl("CREATE TABLE concurrent_test (id INTEGER, value INTEGER)")
.await
.expect("CREATE TABLE should succeed");
}
let success_count = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for i in 0..4 {
let pool_clone = pool.clone();
let success_clone = success_count.clone();
handles.push(tokio::spawn(async move {
let session = match pool_clone.get_session("admin").await {
Ok(s) => s,
Err(_) => return,
};
let sql = format!("INSERT INTO concurrent_test VALUES ({}, {})", i, i * 10);
if session.execute_raw(&sql).await.is_ok() {
success_clone.fetch_add(1, Ordering::SeqCst);
}
}));
}
for handle in handles {
handle.await.expect("Task panicked");
}
assert_eq!(
success_count.load(Ordering::SeqCst),
4,
"All concurrent inserts should succeed"
);
let session = pool
.get_session("admin")
.await
.expect("Failed to get verification session");
let result = session
.execute_raw("SELECT COUNT(*) as cnt FROM concurrent_test")
.await
.expect("COUNT query should succeed");
assert_eq!(result.rows_affected(), 1, "Should return 1 row");
}