ephemeral_postgres/
cluster.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use anyhow::Context;
5use anyhow::Result;
6use sqlx::PgPool;
7use sqlx::postgres::PgPoolOptions;
8use sqlx::query;
9use testcontainers_modules::postgres::Postgres;
10use testcontainers_modules::testcontainers::ContainerAsync;
11use testcontainers_modules::testcontainers::ImageExt;
12use testcontainers_modules::testcontainers::runners::AsyncRunner;
13use uuid::Uuid;
14
15use crate::cluster_params::ClusterParams;
16use crate::database::Database;
17use crate::wait_until_postgres_admin_pool_ready::wait_until_postgres_admin_pool_ready;
18
19const POSTGRES_HOST: &str = "127.0.0.1";
20const POSTGRES_CONTAINER_PORT: u16 = 5432;
21
22async fn start_postgres_container(image_tag: String) -> Result<ContainerAsync<Postgres>> {
23    Postgres::default()
24        .with_host_auth()
25        .with_tag(image_tag)
26        .start()
27        .await
28        .context("failed to start ephemeral postgres container")
29}
30
31pub struct Cluster {
32    admin_pool: PgPool,
33    base_url: String,
34    #[expect(
35        dead_code,
36        reason = "container handle keeps the postgres container alive until the cluster Arc is dropped"
37    )]
38    container: ContainerAsync<Postgres>,
39}
40
41impl Cluster {
42    pub async fn start() -> Result<Arc<Self>> {
43        Self::start_with_params(ClusterParams::default()).await
44    }
45
46    pub async fn start_with_params(params: ClusterParams) -> Result<Arc<Self>> {
47        let ClusterParams {
48            image_tag,
49            readiness_timeout,
50        } = params;
51
52        let container = start_postgres_container(image_tag).await?;
53
54        Self::from_started_container(container, readiness_timeout).await
55    }
56
57    async fn from_started_container(
58        container: ContainerAsync<Postgres>,
59        readiness_timeout: Duration,
60    ) -> Result<Arc<Self>> {
61        let port = container
62            .get_host_port_ipv4(POSTGRES_CONTAINER_PORT)
63            .await
64            .context("failed to resolve ephemeral postgres mapped port")?;
65
66        let base_url = format!("postgres://postgres@{POSTGRES_HOST}:{port}");
67        let admin_pool = wait_until_postgres_admin_pool_ready(
68            &format!("{base_url}/postgres"),
69            readiness_timeout,
70        )
71        .await
72        .context("postgres admin pool did not become ready before timeout")?;
73
74        Ok(Arc::new(Self {
75            admin_pool,
76            base_url,
77            container,
78        }))
79    }
80
81    pub async fn create_database(self: &Arc<Self>) -> Result<Database> {
82        self.create_database_with_id(Uuid::new_v4()).await
83    }
84
85    pub async fn create_database_with_id(self: &Arc<Self>, database_id: Uuid) -> Result<Database> {
86        let db_name = format!("test_{}", database_id.simple());
87
88        query(&format!("CREATE DATABASE \"{db_name}\""))
89            .execute(&self.admin_pool)
90            .await
91            .context("failed to create per-test ephemeral database")?;
92
93        let database_url = format!("{}/{db_name}", self.base_url);
94
95        PgPoolOptions::new()
96            .max_connections(5)
97            .connect(&database_url)
98            .await
99            .context("failed to open pool against per-test ephemeral database")
100            .map(|pool| Database::new(Arc::clone(self), database_url, db_name, pool))
101    }
102
103    #[must_use]
104    pub fn base_url(&self) -> &str {
105        &self.base_url
106    }
107}
108
109#[cfg(test)]
110mod tests {
111    use super::Cluster;
112    use super::start_postgres_container;
113    use crate::cluster_params::ClusterParams;
114
115    #[tokio::test]
116    async fn from_started_container_errors_when_mapped_port_unavailable() {
117        let ClusterParams {
118            image_tag,
119            readiness_timeout,
120        } = ClusterParams::default();
121
122        let container = start_postgres_container(image_tag)
123            .await
124            .expect("postgres container should start");
125        container
126            .stop_with_timeout(Some(0))
127            .await
128            .expect("postgres container should stop before finalizing the cluster");
129
130        let result = Cluster::from_started_container(container, readiness_timeout).await;
131
132        assert!(
133            result.is_err(),
134            "finalizing a cluster must fail when the container's mapped port is unavailable",
135        );
136    }
137}
ephemeral_postgres/cluster.rs

ephemeral_postgres/
cluster.rs
