itsscb/sqlx
1
0
Fork 0
mirror of https://github.com/launchbadge/sqlx.git synced 2025-12-30 05:11:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: tests

Browse Source
This commit is contained in:
Austin Bonander 2024-11-08 21:32:51 -08:00
parent eb478cec03
commit 928b25691e
2 changed files with 42 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
sqlx-core/src/pool/connect.rs
View File

@ -33,10 +33,11 @@ use std::io;
/// use sqlx::PgConnection;
/// use sqlx::postgres::PgPoolOptions;
/// use sqlx::Connection;
/// use sqlx::pool::PoolConnectMetadata;
///
/// # async fn _example() -> sqlx::Result<()> {
/// // `PoolConnector` is implemented for closures but has restrictions on returning borrows
/// // due to current language limitations.
/// async fn _example() -> sqlx::Result<()> {
/// // `PoolConnector` is implemented for closures but this has restrictions on returning borrows
/// // due to current language limitations. Custom implementations are not subject to this.
/// //
/// // This example shows how to get around this using `Arc`.
/// let database_url: Arc<str> = "postgres://...".into();
@ -44,7 +45,8 @@ use std::io;
/// let pool = PgPoolOptions::new()
/// .min_connections(5)
/// .max_connections(30)
/// .connect_with_connector(move |meta| {
/// // Type annotation on the argument is required for the trait impl to reseolve.
/// .connect_with_connector(move |meta: PoolConnectMetadata| {
/// let database_url = database_url.clone();
/// async move {
/// println!(
@ -57,7 +59,9 @@ use std::io;
/// let mut conn = PgConnection::connect(&database_url).await?;
///
/// // Override the time zone of the connection.
/// sqlx::raw_sql("SET TIME ZONE 'Europe/Berlin'").await?;
/// sqlx::raw_sql("SET TIME ZONE 'Europe/Berlin'")
/// .execute(&mut conn)
/// .await?;
///
/// Ok(conn)
/// }
@ -76,13 +80,14 @@ use std::io;
///
/// ```rust,no_run
/// use std::sync::Arc;
/// use tokio::sync::{Mutex, RwLock};
/// use tokio::sync::RwLock;
/// use sqlx::PgConnection;
/// use sqlx::postgres::PgConnectOptions;
/// use sqlx::postgres::PgPoolOptions;
/// use sqlx::ConnectOptions;
/// use sqlx::pool::PoolConnectMetadata;
///
/// # async fn _example() -> sqlx::Result<()> {
/// async fn _example() -> sqlx::Result<()> {
/// // If you do not wish to hold the lock during the connection attempt,
/// // you could use `Arc<PgConnectOptions>` instead.
/// let connect_opts: Arc<RwLock<PgConnectOptions>> = Arc::new(RwLock::new("postgres://...".parse()?));
@ -90,7 +95,7 @@ use std::io;
/// let connect_opts_ = connect_opts.clone();
///
/// let pool = PgPoolOptions::new()
/// .connect_with_connector(move |meta| {
/// .connect_with_connector(move |meta: PoolConnectMetadata| {
/// let connect_opts_ = connect_opts.clone();
/// async move {
/// println!(

90
tests/any/pool.rs
View File

@ -1,45 +1,14 @@
use sqlx::any::{AnyConnectOptions, AnyPoolOptions};
use sqlx::Executor;
use sqlx_core::connection::ConnectOptions;
use sqlx_core::pool::PoolConnectMetadata;
use sqlx_core::sql_str::AssertSqlSafe;
use std::sync::{
atomic::{AtomicI32, AtomicUsize, Ordering},
atomic::{AtomicI32, Ordering},
Arc, Mutex,
};
use std::time::Duration;
#[sqlx_macros::test]
async fn pool_should_invoke_after_connect() -> anyhow::Result<()> {
sqlx::any::install_default_drivers();
let counter = Arc::new(AtomicUsize::new(0));
let pool = AnyPoolOptions::new()
.after_connect({
let counter = counter.clone();
move |_conn, _meta| {
let counter = counter.clone();
Box::pin(async move {
counter.fetch_add(1, Ordering::SeqCst);
Ok(())
})
}
})
.connect(&dotenvy::var("DATABASE_URL")?)
.await?;
let _ = pool.acquire().await?;
let _ = pool.acquire().await?;
let _ = pool.acquire().await?;
let _ = pool.acquire().await?;
// since connections are released asynchronously,
// `.after_connect()` may be called more than once
assert!(counter.load(Ordering::SeqCst) >= 1);
Ok(())
}
// https://github.com/launchbadge/sqlx/issues/527
#[sqlx_macros::test]
async fn pool_should_be_returned_failed_transactions() -> anyhow::Result<()> {
@ -84,38 +53,13 @@ async fn test_pool_callbacks() -> anyhow::Result<()> {
sqlx_test::setup_if_needed();
let conn_options: AnyConnectOptions = std::env::var("DATABASE_URL")?.parse()?;
let conn_options: Arc<AnyConnectOptions> = Arc::new(std::env::var("DATABASE_URL")?.parse()?);
let current_id = AtomicI32::new(0);
let pool = AnyPoolOptions::new()
.max_connections(1)
.acquire_timeout(Duration::from_secs(5))
.after_connect(move |conn, meta| {
assert_eq!(meta.age, Duration::ZERO);
assert_eq!(meta.idle_for, Duration::ZERO);
let id = current_id.fetch_add(1, Ordering::AcqRel);
Box::pin(async move {
let statement = format!(
// language=SQL
r#"
CREATE TEMPORARY TABLE conn_stats(
id int primary key,
before_acquire_calls int default 0,
after_release_calls int default 0
);
INSERT INTO conn_stats(id) VALUES ({});
"#,
// Until we have generalized bind parameters
id
);
conn.execute(AssertSqlSafe(statement)).await?;
Ok(())
})
})
.before_acquire(|conn, meta| {
// `age` and `idle_for` should both be nonzero
assert_ne!(meta.age, Duration::ZERO);
@ -166,7 +110,31 @@ async fn test_pool_callbacks() -> anyhow::Result<()> {
})
})
// Don't establish a connection yet.
.connect_lazy_with(conn_options);
.connect_lazy_with_connector(move |_meta: PoolConnectMetadata| {
let connect_opts = Arc::clone(&conn_options);
let id = current_id.fetch_add(1, Ordering::AcqRel);
async move {
let mut conn = connect_opts.connect().await?;
let statement = format!(
// language=SQL
r#"
CREATE TEMPORARY TABLE conn_stats(
id int primary key,
before_acquire_calls int default 0,
after_release_calls int default 0
);
INSERT INTO conn_stats(id) VALUES ({});
"#,
// Until we have generalized bind parameters
id
);
conn.execute(&statement[..]).await?;
Ok(conn)
}
});
// Expected pattern of (id, before_acquire_calls, after_release_calls)
let pattern = [