25660589290

Committed 11 May 2026 09:00AM UTC coverage: 76.533% (+0.006%) from 76.527%

Build # 25660589290

Build Type

push

github

Committed by

web-flow

Commit Message

bump(vector): Bump vector image to our own ECR registry (#729)

Coverage Stats

29737 of 38855 relevant lines covered (76.53%)

896.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.67

/etl-postgres/src/sqlx/test_utils.rs

use etl_config::shared::{IntoConnectOptions, PgConnectionConfig};
use pg_escape::{quote_identifier, quote_literal};
use sqlx::{AssertSqlSafe, Connection, PgConnection, PgPool};

/// Creates a new Postgres database and returns a connection pool.
///
/// Connects to Postgres server, creates a new database, and returns a
/// [`PgPool`] connected to the newly created database.
///
/// # Panics
/// Panics if connection or database creation fails.
pub async fn create_pg_database(config: &PgConnectionConfig) -> PgPool {
    // Create the database via a single connection.
    let mut connection = PgConnection::connect_with(&config.without_db(None))
        .await
        .expect("Failed to connect to Postgres");
    let create_database_query = format!("create database {};", quote_identifier(&config.name));
    sqlx::query(AssertSqlSafe(create_database_query))
        .execute(&mut connection)
        .await
        .expect("Failed to create database");

    // Create a connection pool to the database.
    PgPool::connect_with(config.with_db(None)).await.expect("Failed to connect to Postgres")
}

/// Drops a Postgres database and terminates all connections.
///
/// Connects to Postgres server, forcefully terminates active connections
/// to the target database, and drops it if it exists. Used for test cleanup.
///
/// This function will not panic on errors - it logs them and continues.
/// This ensures test cleanup doesn't fail when databases are already gone
/// or connections can't be established.
pub async fn drop_pg_database(config: &PgConnectionConfig) {
    // Connect to the default database.
    let mut connection = match PgConnection::connect_with(&config.without_db(None)).await {
        Ok(conn) => conn,
        Err(e) => {
            eprintln!("warning: failed to connect to Postgres for cleanup: {e}");
            return;
        }
    };

    // Forcefully terminate any remaining connections to the database.
    let terminate_connections_query = format!(
        r#"
            select pg_terminate_backend(pg_stat_activity.pid)
            from pg_stat_activity
            where pg_stat_activity.datname = {}
            and pid <> pg_backend_pid();"#,
        quote_literal(&config.name)
    );
    if let Err(e) =
        sqlx::query(AssertSqlSafe(terminate_connections_query)).execute(&mut connection).await
    {
        eprintln!("warning: failed to terminate connections for database {}: {}", config.name, e);
    }

    // Drop the database.
    let drop_database_query =
        format!("drop database if exists {};", quote_identifier(&config.name));
    if let Err(e) = sqlx::query(AssertSqlSafe(drop_database_query)).execute(&mut connection).await {
        eprintln!("warning: failed to drop database {}: {}", config.name, e);
    }
}

1	use etl_config::shared::{IntoConnectOptions, PgConnectionConfig};
2	use pg_escape::{quote_identifier, quote_literal};
3	use sqlx::{AssertSqlSafe, Connection, PgConnection, PgPool};
4
5	/// Creates a new Postgres database and returns a connection pool.
6	///
7	/// Connects to Postgres server, creates a new database, and returns a
8	/// [`PgPool`] connected to the newly created database.
9	///
10	/// # Panics
11	/// Panics if connection or database creation fails.
12	pub async fn create_pg_database(config: &PgConnectionConfig) -> PgPool {	160✔
13	// Create the database via a single connection.
14	let mut connection = PgConnection::connect_with(&config.without_db(None))	160✔
15	.await	160✔
16	.expect("Failed to connect to Postgres");	160✔
17	let create_database_query = format!("create database {};", quote_identifier(&config.name));	160✔
18	sqlx::query(AssertSqlSafe(create_database_query))	160✔
19	.execute(&mut connection)	160✔
20	.await	160✔
21	.expect("Failed to create database");	160✔
22
23	// Create a connection pool to the database.
24	PgPool::connect_with(config.with_db(None)).await.expect("Failed to connect to Postgres")	160✔
25	}	160✔
26
27	/// Drops a Postgres database and terminates all connections.
28	///
29	/// Connects to Postgres server, forcefully terminates active connections
30	/// to the target database, and drops it if it exists. Used for test cleanup.
31	///
32	/// This function will not panic on errors - it logs them and continues.
33	/// This ensures test cleanup doesn't fail when databases are already gone
34	/// or connections can't be established.
35	pub async fn drop_pg_database(config: &PgConnectionConfig) {	160✔
36	// Connect to the default database.
37	let mut connection = match PgConnection::connect_with(&config.without_db(None)).await {	160✔
38	Ok(conn) => conn,	160✔
39	Err(e) => {	×
40	eprintln!("warning: failed to connect to Postgres for cleanup: {e}");	×
41	return;	×
42	}
43	};
44
45	// Forcefully terminate any remaining connections to the database.
46	let terminate_connections_query = format!(	160✔
47	r#"
48	select pg_terminate_backend(pg_stat_activity.pid)
49	from pg_stat_activity
50	where pg_stat_activity.datname = {}
51	and pid <> pg_backend_pid();"#,
52	quote_literal(&config.name)	160✔
53	);
54	if let Err(e) =	×
55	sqlx::query(AssertSqlSafe(terminate_connections_query)).execute(&mut connection).await	160✔
56	{	×
57	eprintln!("warning: failed to terminate connections for database {}: {}", config.name, e);	×
58	}	160✔
59
60	// Drop the database.
61	let drop_database_query =	160✔
62	format!("drop database if exists {};", quote_identifier(&config.name));	160✔
63	if let Err(e) = sqlx::query(AssertSqlSafe(drop_database_query)).execute(&mut connection).await {	160✔
64	eprintln!("warning: failed to drop database {}: {}", config.name, e);	×
65	}	160✔
66	}	160✔

supabase / etl / 25660589290

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous