tmp_postgrust/
lib.rs

1/*!
2`tmp-postgrust` provides temporary postgresql processes that are cleaned up
3after being dropped.
4
5
6# Inspiration / Similar Projects
7- [tmp-postgres](https://github.com/jfischoff/tmp-postgres)
8- [testing.postgresql](https://github.com/tk0miya/testing.postgresql)
9*/
10#![deny(missing_docs)]
11#![warn(clippy::all, clippy::pedantic)]
12
13/// Methods for Asynchronous API
14#[cfg(feature = "tokio-process")]
15pub mod asynchronous;
16/// Common Errors
17pub mod errors;
18mod search;
19/// Methods for Synchronous API
20pub mod synchronous;
21
22use std::fmt::Write as FmtWrite;
23use std::fs::{metadata, set_permissions};
24use std::io::{BufRead, BufReader};
25use std::path::{Path, PathBuf};
26use std::sync::atomic::AtomicU32;
27use std::sync::{Arc, OnceLock, RwLock};
28use std::{fs::File, io::Write};
29
30use ctor::dtor;
31use nix::unistd::{Gid, Uid};
32use tempfile::{Builder, TempDir};
33use tracing::{debug, info, instrument};
34
35use crate::errors::{TmpPostgrustError, TmpPostgrustResult};
36use crate::search::PostgresBinaries;
37
38const TMP_POSTGRUST_DB_NAME: &str = "tmp-postgrust";
39const TMP_POSTGRUST_USER_NAME: &str = "tmp-postgrust-user";
40
41pub(crate) static POSTGRES_UID_GID: OnceLock<(Uid, Gid)> = OnceLock::new();
42
43/// As the static variables declared by this crate contain values that
44/// need to be dropped at program exit to clean up resources, we use a
45/// `#[dtor]` hack to drop the variables if they have been initialized.
46#[dtor]
47fn cleanup_static() {
48    #[cfg(feature = "tokio-process")]
49    if let Some(factory_mutex) = TOKIO_POSTGRES_FACTORY.get() {
50        let mut guard = factory_mutex.blocking_write();
51        drop(guard.take());
52    }
53
54    if let Some(factory_mutex) = DEFAULT_POSTGRES_FACTORY.get() {
55        let mut guard = factory_mutex
56            .write()
57            .expect("Failed to lock default factory mutex.");
58        drop(guard.take());
59    }
60}
61
62static DEFAULT_POSTGRES_FACTORY: OnceLock<RwLock<Option<TmpPostgrustFactory>>> = OnceLock::new();
63
64/// Create a new default instance, initializing the `DEFAULT_POSTGRES_FACTORY` if it
65/// does not already exist.
66///
67/// fsync is disabled in the default process.
68///
69/// # Errors
70///
71/// Will return `Err` if postgres is not installed on system
72///
73/// # Panics
74///
75/// Will panic if a `TmpPostgrustFactory::try_new` returns an error the first time the function
76/// is called.
77pub fn new_default_process() -> TmpPostgrustResult<synchronous::ProcessGuard> {
78    let factory_mutex = DEFAULT_POSTGRES_FACTORY.get_or_init(|| {
79        RwLock::new(Some(
80            TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
81                disable_fsync: true,
82                ..Default::default()
83            })
84            .expect("Failed to initialize default postgres factory."),
85        ))
86    });
87    let guard = factory_mutex
88        .read()
89        .expect("Failed to lock default factory mutex.");
90    let factory = guard
91        .as_ref()
92        .expect("Default factory is uninitialized or has been dropped.");
93    factory.new_instance()
94}
95
96/// Create a new default instance, initializing the `DEFAULT_POSTGRES_FACTORY` if it
97/// does not already exist. The function passed as the `migrate` parameters
98/// will be run the first time the factory is initialised.
99///
100/// fsync is disabled in the default process.
101///
102/// # Errors
103///
104/// Will return `Err` if postgres is not installed on system
105///
106/// # Panics
107///
108/// Will panic if a `TmpPostgrustFactory::try_new` returns an error the first time the function
109/// is called.
110pub fn new_default_process_with_migrations(
111    migrate: impl Fn(&str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>,
112) -> TmpPostgrustResult<synchronous::ProcessGuard> {
113    let factory_mutex = DEFAULT_POSTGRES_FACTORY.get_or_init(|| {
114        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
115            disable_fsync: true,
116            ..Default::default()
117        })
118        .expect("Failed to initialize default postgres factory.");
119        factory
120            .run_migrations(migrate)
121            .expect("Failed to run migrations");
122
123        RwLock::new(Some(factory))
124    });
125    let guard = factory_mutex
126        .read()
127        .expect("Failed to lock default factory mutex.");
128    let factory = guard
129        .as_ref()
130        .expect("Default factory is uninitialized or has been dropped.");
131    factory.new_instance()
132}
133
134/// Static factory that can be re-used between tests.
135#[cfg(feature = "tokio-process")]
136static TOKIO_POSTGRES_FACTORY: tokio::sync::OnceCell<
137    tokio::sync::RwLock<Option<TmpPostgrustFactory>>,
138> = tokio::sync::OnceCell::const_new();
139
140/// Create a new default instance, initializing the `TOKIO_POSTGRES_FACTORY` if it
141/// does not already exist.
142///
143/// fsync is disabled in the default process.
144///
145/// # Errors
146///
147/// Will return `Err` if postgres is not installed on system
148///
149/// # Panics
150///
151/// Will panic if a `TmpPostgrustFactory::try_new_async` returns an error the first time the function
152/// is called.
153#[cfg(feature = "tokio-process")]
154pub async fn new_default_process_async() -> TmpPostgrustResult<asynchronous::ProcessGuard> {
155    let factory_mutex = TOKIO_POSTGRES_FACTORY
156        .get_or_try_init(|| async {
157            TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig {
158                disable_fsync: true,
159                ..Default::default()
160            })
161            .await
162            .map(|factory| tokio::sync::RwLock::new(Some(factory)))
163        })
164        .await?;
165    let guard = factory_mutex.read().await;
166    let factory = guard
167        .as_ref()
168        .expect("Default tokio factory is uninitialized or has been dropped.");
169    factory.new_instance_async().await
170}
171
172/// Create a new default instance, initializing the `TOKIO_POSTGRES_FACTORY` if it
173/// does not already exist. The function passed as the `migrate` parameters
174/// will be run the first time the factory is initialised.
175///
176/// fsync is disabled in the default process.
177///
178/// # Errors
179///
180/// Will return `Err` if postgres is not installed on system
181///
182/// # Panics
183///
184/// Will panic if a `TmpPostgrustFactory::try_new_async` returns an error the first time the function
185/// is called.
186#[cfg(feature = "tokio-process")]
187pub async fn new_default_process_async_with_migrations<F>(
188    migrate: F,
189) -> TmpPostgrustResult<asynchronous::ProcessGuard>
190where
191    F: for<'r> Fn(
192        &'r str,
193    ) -> futures_util::future::BoxFuture<
194        'r,
195        Result<(), Box<dyn std::error::Error + Send + Sync>>,
196    >,
197{
198    let factory_mutex = TOKIO_POSTGRES_FACTORY
199        .get_or_try_init(|| async {
200            let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig {
201                disable_fsync: true,
202                ..Default::default()
203            })
204            .await?;
205            factory.run_migrations_async(migrate).await?;
206            Ok(tokio::sync::RwLock::new(Some(factory)))
207        })
208        .await?;
209    let guard = factory_mutex.read().await;
210    let factory = guard
211        .as_ref()
212        .expect("Default tokio factory is uninitialized or has been dropped.");
213    factory.new_instance_async().await
214}
215
216/// Factory for creating new temporary postgresql processes.
217#[derive(Debug)]
218pub struct TmpPostgrustFactory {
219    socket_dir: Arc<TempDir>,
220    cache_dir: Arc<TempDir>,
221    config: String,
222    next_port: AtomicU32,
223    binaries: PostgresBinaries,
224}
225
226/// Configuration for the `TmpPostgrustFactory`.
227#[derive(Default, Debug)]
228#[non_exhaustive]
229pub struct TmpPostgrustFactoryConfig {
230    /// Disable fsync this will speed up unit tests in exchange for
231    /// not guaranteeing that files will be written if postgresql
232    /// crashes.
233    pub disable_fsync: bool,
234    /// Directory containing the postgres binaries (`postgres`, `initdb`,
235    /// `createdb`, `createuser`). When `None`, binaries are resolved from
236    /// `$PATH` and common install locations.
237    pub postgresql_bin_dir: Option<PathBuf>,
238}
239
240impl TmpPostgrustFactory {
241    /// Build a Postgresql configuration for temporary databases as a String.
242    fn build_config(factory_config: &TmpPostgrustFactoryConfig, socket_dir: &Path) -> String {
243        let mut config = String::new();
244        // Minimize chance of running out of shared memory
245        config.push_str("shared_buffers = '12MB'\n");
246        // Disable TCP connections.
247        config.push_str("listen_addresses = ''\n");
248        // Listen on UNIX socket.
249        writeln!(
250            &mut config,
251            "unix_socket_directories = \'{}\'",
252            socket_dir.to_str().unwrap()
253        )
254        .expect("Failed to append unix socket to config");
255        // Disable fsync this will speed up unit tests in exchange for
256        // not guaranteeing that files will be written if postgresql
257        // crashes.
258        writeln!(&mut config, "fsync = \'{}\'", !factory_config.disable_fsync)
259            .expect("Failed to append fsync option to config");
260
261        config
262    }
263
264    /// Try to create a new factory by creating temporary directories and the necessary config.
265    #[instrument]
266    pub fn try_new(
267        factory_config: &TmpPostgrustFactoryConfig,
268    ) -> TmpPostgrustResult<TmpPostgrustFactory> {
269        let binaries = PostgresBinaries::resolve(factory_config.postgresql_bin_dir.as_deref())?;
270
271        let socket_dir = Builder::new()
272            .prefix("tmp-postgrust-socket")
273            .tempdir()
274            .map_err(TmpPostgrustError::CreateSocketDirFailed)?;
275        let cache_dir = Builder::new()
276            .prefix("tmp-postgrust-cache")
277            .tempdir()
278            .map_err(TmpPostgrustError::CreateCacheDirFailed)?;
279
280        synchronous::chown_to_non_root(cache_dir.path())?;
281        synchronous::chown_to_non_root(socket_dir.path())?;
282        synchronous::exec_init_db(&binaries.initdb, cache_dir.path())?;
283
284        let config = TmpPostgrustFactory::build_config(factory_config, socket_dir.path());
285
286        let factory = TmpPostgrustFactory {
287            socket_dir: Arc::new(socket_dir),
288            cache_dir: Arc::new(cache_dir),
289            config,
290            next_port: AtomicU32::new(5432),
291            binaries,
292        };
293        let process = factory.start_postgresql(&factory.cache_dir)?;
294        synchronous::exec_create_user(
295            &factory.binaries.createuser,
296            process.socket_dir.path(),
297            process.port,
298            &process.user_name,
299        )?;
300        synchronous::exec_create_db(
301            &factory.binaries.createdb,
302            process.socket_dir.path(),
303            process.port,
304            &process.user_name,
305            &process.db_name,
306        )?;
307
308        Ok(factory)
309    }
310
311    /// Try to create a new factory by creating temporary directories and the necessary config.
312    #[cfg(feature = "tokio-process")]
313    #[instrument]
314    pub async fn try_new_async(
315        factory_config: &TmpPostgrustFactoryConfig,
316    ) -> TmpPostgrustResult<TmpPostgrustFactory> {
317        let binaries = PostgresBinaries::resolve(factory_config.postgresql_bin_dir.as_deref())?;
318
319        let socket_dir = Builder::new()
320            .prefix("tmp-postgrust-socket")
321            .tempdir()
322            .map_err(TmpPostgrustError::CreateSocketDirFailed)?;
323        let cache_dir = Builder::new()
324            .prefix("tmp-postgrust-cache")
325            .tempdir()
326            .map_err(TmpPostgrustError::CreateCacheDirFailed)?;
327
328        asynchronous::chown_to_non_root(cache_dir.path()).await?;
329        asynchronous::chown_to_non_root(socket_dir.path()).await?;
330        asynchronous::exec_init_db(&binaries.initdb, cache_dir.path()).await?;
331
332        let config = TmpPostgrustFactory::build_config(factory_config, socket_dir.path());
333
334        let factory = TmpPostgrustFactory {
335            socket_dir: Arc::new(socket_dir),
336            cache_dir: Arc::new(cache_dir),
337            config,
338            next_port: AtomicU32::new(5432),
339            binaries,
340        };
341        let process = factory.start_postgresql_async(&factory.cache_dir).await?;
342        asynchronous::exec_create_user(
343            &factory.binaries.createuser,
344            process.socket_dir.path(),
345            process.port,
346            &process.user_name,
347        )
348        .await?;
349        asynchronous::exec_create_db(
350            &factory.binaries.createdb,
351            process.socket_dir.path(),
352            process.port,
353            &process.user_name,
354            &process.db_name,
355        )
356        .await?;
357
358        Ok(factory)
359    }
360
361    /// Run migrations against the cache directory, will cause all subsequent instances
362    /// to be run against a version of the database where the migrations have been applied.
363    ///
364    /// # Errors
365    ///
366    /// Will error if Postgresql is unable to start or if the migrate function returns
367    /// an error.
368    pub fn run_migrations(
369        &self,
370        migrate: impl Fn(&str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>,
371    ) -> TmpPostgrustResult<()> {
372        let process = self.start_postgresql(&self.cache_dir)?;
373
374        migrate(&process.connection_string()).map_err(TmpPostgrustError::MigrationsFailed)?;
375
376        Ok(())
377    }
378
379    /// Run migrations against the cache directory, will cause all subsequent instances
380    /// to be run against a version of the database where the migrations have been applied.
381    ///
382    /// # Errors
383    ///
384    /// Will error if Postgresql is unable to start or if the migrate function returns
385    /// an error.
386    #[cfg(feature = "tokio-process")]
387    pub async fn run_migrations_async<F>(&self, migrate: F) -> TmpPostgrustResult<()>
388    where
389        F: for<'r> Fn(
390            &'r str,
391        ) -> futures_util::future::BoxFuture<
392            'r,
393            Result<(), Box<dyn std::error::Error + Send + Sync>>,
394        >,
395    {
396        let process = self.start_postgresql_async(&self.cache_dir).await?;
397
398        migrate(&process.connection_string())
399            .await
400            .map_err(TmpPostgrustError::MigrationsFailed)?;
401
402        Ok(())
403    }
404
405    /// Start a new postgresql instance and return a process guard that will ensure it is cleaned
406    /// up when dropped.
407    #[instrument(skip(self))]
408    pub fn new_instance(&self) -> TmpPostgrustResult<synchronous::ProcessGuard> {
409        let data_directory = Builder::new()
410            .prefix("tmp-postgrust-db")
411            .tempdir()
412            .map_err(TmpPostgrustError::CreateDataDirFailed)?;
413        let data_directory_path = data_directory.path();
414
415        set_permissions(
416            &data_directory,
417            metadata(self.cache_dir.path()).unwrap().permissions(),
418        )
419        .unwrap();
420        synchronous::exec_copy_dir(self.cache_dir.path(), data_directory_path)?;
421
422        if !data_directory_path.join("PG_VERSION").exists() {
423            return Err(TmpPostgrustError::EmptyDataDirectory);
424        }
425
426        self.start_postgresql(&Arc::new(data_directory))
427    }
428
429    #[instrument(skip(self))]
430    fn start_postgresql(
431        &self,
432        dir: &Arc<TempDir>,
433    ) -> TmpPostgrustResult<synchronous::ProcessGuard> {
434        File::create(dir.path().join("postgresql.conf"))
435            .map_err(TmpPostgrustError::CreateConfigFailed)?
436            .write_all(self.config.as_bytes())
437            .map_err(TmpPostgrustError::CreateConfigFailed)?;
438
439        let port = self
440            .next_port
441            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
442
443        synchronous::chown_to_non_root(dir.path())?;
444        let mut postgres_process_handle =
445            synchronous::start_postgres_subprocess(&self.binaries.postgres, dir.path(), port)?;
446        let stdout = postgres_process_handle.stdout.take().unwrap();
447        let stderr = postgres_process_handle.stderr.take().unwrap();
448
449        let stdout_reader = BufReader::new(stdout).lines();
450        let mut stderr_reader = BufReader::new(stderr).lines();
451
452        while let Some(Ok(line)) = stderr_reader.next() {
453            debug!("Postgresql: {}", line);
454            if line.contains("database system is ready to accept connections") {
455                info!("temporary database system is read to accept connections");
456                break;
457            }
458        }
459
460        Ok(synchronous::ProcessGuard {
461            stdout_reader: Some(stdout_reader),
462            stderr_reader: Some(stderr_reader),
463            port,
464            db_name: TMP_POSTGRUST_DB_NAME.to_string(),
465            user_name: TMP_POSTGRUST_USER_NAME.to_string(),
466            postgres_process: postgres_process_handle,
467            _data_directory: Arc::clone(dir),
468            _cache_directory: Arc::clone(&self.cache_dir),
469            socket_dir: Arc::clone(&self.socket_dir),
470        })
471    }
472
473    /// Start a new postgresql instance and return a process guard that will ensure it is cleaned
474    /// up when dropped.
475    #[cfg(feature = "tokio-process")]
476    #[instrument(skip(self))]
477    pub async fn new_instance_async(&self) -> TmpPostgrustResult<asynchronous::ProcessGuard> {
478        use tokio::fs::{metadata, set_permissions};
479
480        let data_directory = Builder::new()
481            .prefix("tmp-postgrust-db")
482            .tempdir()
483            .map_err(TmpPostgrustError::CreateDataDirFailed)?;
484        let data_directory_path = data_directory.path();
485
486        set_permissions(
487            &data_directory,
488            metadata(self.cache_dir.path()).await.unwrap().permissions(),
489        )
490        .await
491        .unwrap();
492        asynchronous::exec_copy_dir(self.cache_dir.path(), data_directory_path).await?;
493
494        if !data_directory_path.join("PG_VERSION").exists() {
495            return Err(TmpPostgrustError::EmptyDataDirectory);
496        }
497
498        self.start_postgresql_async(&Arc::new(data_directory)).await
499    }
500
501    #[cfg(feature = "tokio-process")]
502    #[instrument(skip(self))]
503    async fn start_postgresql_async(
504        &self,
505        dir: &Arc<TempDir>,
506    ) -> TmpPostgrustResult<asynchronous::ProcessGuard> {
507        use tokio::io::AsyncBufReadExt;
508
509        File::create(dir.path().join("postgresql.conf"))
510            .map_err(TmpPostgrustError::CreateConfigFailed)?
511            .write_all(self.config.as_bytes())
512            .map_err(TmpPostgrustError::CreateConfigFailed)?;
513
514        let port = self
515            .next_port
516            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
517
518        asynchronous::chown_to_non_root(dir.path()).await?;
519
520        let permit = crate::asynchronous::MAX_CONCURRENT_PROCESSES
521            .get_or_init(|| tokio::sync::Semaphore::new(num_cpus::get()))
522            .acquire()
523            .await
524            .map_err(TmpPostgrustError::AsyncProcessPermitAcquireError)?;
525
526        let mut postgres_process_handle =
527            asynchronous::start_postgres_subprocess(&self.binaries.postgres, dir.path(), port)?;
528        let stdout = postgres_process_handle.stdout.take().unwrap();
529        let stderr = postgres_process_handle.stderr.take().unwrap();
530
531        let stdout_reader = tokio::io::BufReader::new(stdout).lines();
532        let mut stderr_reader = tokio::io::BufReader::new(stderr).lines();
533
534        while let Some(line) = stderr_reader.next_line().await.unwrap() {
535            debug!("Postgresql: {}", line);
536            if line.contains("database system is ready to accept connections") {
537                info!("temporary database system is read to accept connections");
538                break;
539            }
540        }
541
542        Ok(asynchronous::ProcessGuard {
543            stdout_reader: Some(stdout_reader),
544            stderr_reader: Some(stderr_reader),
545            port,
546            db_name: TMP_POSTGRUST_DB_NAME.to_string(),
547            user_name: TMP_POSTGRUST_USER_NAME.to_string(),
548            _data_directory: Arc::clone(dir),
549            _cache_directory: Arc::clone(&self.cache_dir),
550            socket_dir: Arc::clone(&self.socket_dir),
551            postgres_process: postgres_process_handle,
552            _process_permit: permit,
553        })
554    }
555}
556
557#[cfg(test)]
558mod tests {
559    use super::*;
560
561    use test_log::test;
562    use tokio_postgres::NoTls;
563    use tracing::error;
564
565    #[test(tokio::test)]
566    async fn it_works() {
567        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
568            disable_fsync: false,
569            ..Default::default()
570        })
571        .expect("failed to create factory");
572
573        let postgresql_proc = factory
574            .new_instance()
575            .expect("failed to create a new instance");
576
577        let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls)
578            .await
579            .expect("failed to connect to postgresql");
580
581        tokio::spawn(async move {
582            if let Err(e) = conn.await {
583                error!("connection error: {}", e);
584            }
585        });
586
587        client.query("SELECT 1;", &[]).await.unwrap();
588    }
589
590    #[test(tokio::test)]
591    async fn it_works_fsync_disabled() {
592        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
593            disable_fsync: true,
594            ..Default::default()
595        })
596        .expect("failed to create factory");
597
598        let postgresql_proc = factory
599            .new_instance()
600            .expect("failed to create a new instance");
601
602        let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls)
603            .await
604            .expect("failed to connect to postgresql");
605
606        tokio::spawn(async move {
607            if let Err(e) = conn.await {
608                error!("connection error: {}", e);
609            }
610        });
611
612        client.query("SELECT 1;", &[]).await.unwrap();
613    }
614
615    #[cfg(feature = "tokio-process")]
616    #[test(tokio::test)]
617    async fn it_works_async() {
618        let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig {
619            disable_fsync: false,
620            ..Default::default()
621        })
622        .await
623        .expect("failed to create factory");
624
625        let postgresql_proc = factory
626            .new_instance_async()
627            .await
628            .expect("failed to create a new instance");
629
630        let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls)
631            .await
632            .expect("failed to connect to postgresql");
633
634        tokio::spawn(async move {
635            if let Err(e) = conn.await {
636                error!("connection error: {}", e);
637            }
638        });
639
640        client.query("SELECT 1;", &[]).await.unwrap();
641    }
642
643    #[test(tokio::test)]
644    async fn two_simulatenous_processes() {
645        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
646            disable_fsync: false,
647            ..Default::default()
648        })
649        .expect("failed to create factory");
650
651        let proc1 = factory
652            .new_instance()
653            .expect("failed to create a new instance");
654
655        let proc2 = factory
656            .new_instance()
657            .expect("failed to create a new instance");
658
659        let (client1, conn1) = tokio_postgres::connect(&proc1.connection_string(), NoTls)
660            .await
661            .expect("failed to connect to postgresql");
662
663        tokio::spawn(async move {
664            if let Err(e) = conn1.await {
665                error!("connection error: {}", e);
666            }
667        });
668
669        let (client2, conn2) = tokio_postgres::connect(&proc2.connection_string(), NoTls)
670            .await
671            .expect("failed to connect to postgresql");
672
673        tokio::spawn(async move {
674            if let Err(e) = conn2.await {
675                error!("connection error: {}", e);
676            }
677        });
678
679        client1.query("SELECT 1;", &[]).await.unwrap();
680        client2.query("SELECT 1;", &[]).await.unwrap();
681    }
682
683    #[cfg(feature = "tokio-process")]
684    #[test(tokio::test)]
685    async fn two_simulatenous_processes_async() {
686        let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig {
687            disable_fsync: false,
688            ..Default::default()
689        })
690        .await
691        .expect("failed to create factory");
692
693        let proc1 = factory
694            .new_instance_async()
695            .await
696            .expect("failed to create a new instance");
697
698        let proc2 = factory
699            .new_instance_async()
700            .await
701            .expect("failed to create a new instance");
702
703        let (client1, conn1) = tokio_postgres::connect(&proc1.connection_string(), NoTls)
704            .await
705            .expect("failed to connect to postgresql");
706
707        tokio::spawn(async move {
708            if let Err(e) = conn1.await {
709                error!("connection error: {}", e);
710            }
711        });
712
713        let (client2, conn2) = tokio_postgres::connect(&proc2.connection_string(), NoTls)
714            .await
715            .expect("failed to connect to postgresql");
716
717        tokio::spawn(async move {
718            if let Err(e) = conn2.await {
719                error!("connection error: {}", e);
720            }
721        });
722
723        client1.query("SELECT 1;", &[]).await.unwrap();
724        client2.query("SELECT 1;", &[]).await.unwrap();
725    }
726
727    #[cfg(feature = "tokio-process")]
728    #[test(tokio::test)]
729    async fn default_process_factory_1() {
730        let proc = new_default_process_async().await.unwrap();
731
732        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
733            .await
734            .expect("failed to connect to postgresql");
735
736        tokio::spawn(async move {
737            if let Err(e) = conn.await {
738                error!("connection error: {}", e);
739            }
740        });
741
742        // Chance to catch concurrent tests or database that have already been used.
743        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
744    }
745
746    #[cfg(feature = "tokio-process")]
747    #[test(tokio::test)]
748    async fn default_process_factory_2() {
749        let proc = new_default_process_async().await.unwrap();
750
751        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
752            .await
753            .expect("failed to connect to postgresql");
754
755        tokio::spawn(async move {
756            if let Err(e) = conn.await {
757                error!("connection error: {}", e);
758            }
759        });
760
761        // Chance to catch concurrent tests or database that have already been used.
762        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
763    }
764
765    #[cfg(feature = "tokio-process")]
766    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
767    async fn default_process_factory_multithread_1() {
768        let proc = new_default_process_async().await.unwrap();
769
770        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
771            .await
772            .expect("failed to connect to postgresql");
773
774        tokio::spawn(async move {
775            if let Err(e) = conn.await {
776                error!("connection error: {}", e);
777            }
778        });
779
780        // Chance to catch concurrent tests or database that have already been used.
781        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
782    }
783
784    #[cfg(feature = "tokio-process")]
785    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
786    async fn default_process_factory_multithread_2() {
787        let proc = new_default_process_async().await.unwrap();
788
789        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
790            .await
791            .expect("failed to connect to postgresql");
792
793        tokio::spawn(async move {
794            if let Err(e) = conn.await {
795                error!("connection error: {}", e);
796            }
797        });
798
799        // Chance to catch concurrent tests or database that have already been used.
800        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
801    }
802
803    #[test(tokio::test)]
804    async fn explicit_bin_dir_works() {
805        let postgres_path = crate::search::find_postgresql_command(None, "postgres").expect(
806            "cannot derive bin_dir: postgres not found via PATH or known install locations",
807        );
808        let bin_dir = postgres_path
809            .parent()
810            .expect("postgres path has no parent dir")
811            .to_owned();
812
813        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
814            disable_fsync: true,
815            postgresql_bin_dir: Some(bin_dir),
816        })
817        .expect("failed to create factory with explicit bin dir");
818
819        let proc = factory
820            .new_instance()
821            .expect("failed to create a new instance");
822
823        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
824            .await
825            .expect("failed to connect to postgresql");
826
827        tokio::spawn(async move {
828            if let Err(e) = conn.await {
829                error!("connection error: {}", e);
830            }
831        });
832
833        client.query("SELECT 1;", &[]).await.unwrap();
834    }
835
836    #[test]
837    fn bogus_bin_dir_returns_error() {
838        let result = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
839            disable_fsync: true,
840            postgresql_bin_dir: Some(std::path::PathBuf::from("/nonexistent/bin/dir")),
841        });
842        assert!(
843            matches!(
844                result,
845                Err(TmpPostgrustError::PostgresCommandNotFound { .. })
846            ),
847            "expected PostgresCommandNotFound, got: {:?}",
848            result
849        );
850    }
851}
tmp_postgrust/lib.rs

tmp_postgrust/
lib.rs
