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            None,
307        )?;
308
309        Ok(factory)
310    }
311
312    /// Try to create a new factory by creating temporary directories and the necessary config.
313    #[cfg(feature = "tokio-process")]
314    #[instrument]
315    pub async fn try_new_async(
316        factory_config: &TmpPostgrustFactoryConfig,
317    ) -> TmpPostgrustResult<TmpPostgrustFactory> {
318        let binaries = PostgresBinaries::resolve(factory_config.postgresql_bin_dir.as_deref())?;
319
320        let socket_dir = Builder::new()
321            .prefix("tmp-postgrust-socket")
322            .tempdir()
323            .map_err(TmpPostgrustError::CreateSocketDirFailed)?;
324        let cache_dir = Builder::new()
325            .prefix("tmp-postgrust-cache")
326            .tempdir()
327            .map_err(TmpPostgrustError::CreateCacheDirFailed)?;
328
329        asynchronous::chown_to_non_root(cache_dir.path()).await?;
330        asynchronous::chown_to_non_root(socket_dir.path()).await?;
331        asynchronous::exec_init_db(&binaries.initdb, cache_dir.path()).await?;
332
333        let config = TmpPostgrustFactory::build_config(factory_config, socket_dir.path());
334
335        let factory = TmpPostgrustFactory {
336            socket_dir: Arc::new(socket_dir),
337            cache_dir: Arc::new(cache_dir),
338            config,
339            next_port: AtomicU32::new(5432),
340            binaries,
341        };
342        let process = factory.start_postgresql_async(&factory.cache_dir).await?;
343        asynchronous::exec_create_user(
344            &factory.binaries.createuser,
345            process.socket_dir.path(),
346            process.port,
347            &process.user_name,
348        )
349        .await?;
350        asynchronous::exec_create_db(
351            &factory.binaries.createdb,
352            process.socket_dir.path(),
353            process.port,
354            &process.user_name,
355            &process.db_name,
356            None,
357        )
358        .await?;
359
360        Ok(factory)
361    }
362
363    /// Run migrations against the cache directory, will cause all subsequent instances
364    /// to be run against a version of the database where the migrations have been applied.
365    ///
366    /// # Errors
367    ///
368    /// Will error if Postgresql is unable to start or if the migrate function returns
369    /// an error.
370    pub fn run_migrations(
371        &self,
372        migrate: impl Fn(&str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>,
373    ) -> TmpPostgrustResult<()> {
374        let process = self.start_postgresql(&self.cache_dir)?;
375
376        migrate(&process.connection_string()).map_err(TmpPostgrustError::MigrationsFailed)?;
377
378        Ok(())
379    }
380
381    /// Run migrations against the cache directory, will cause all subsequent instances
382    /// to be run against a version of the database where the migrations have been applied.
383    ///
384    /// # Errors
385    ///
386    /// Will error if Postgresql is unable to start or if the migrate function returns
387    /// an error.
388    #[cfg(feature = "tokio-process")]
389    pub async fn run_migrations_async<F>(&self, migrate: F) -> TmpPostgrustResult<()>
390    where
391        F: for<'r> Fn(
392            &'r str,
393        ) -> futures_util::future::BoxFuture<
394            'r,
395            Result<(), Box<dyn std::error::Error + Send + Sync>>,
396        >,
397    {
398        let process = self.start_postgresql_async(&self.cache_dir).await?;
399
400        migrate(&process.connection_string())
401            .await
402            .map_err(TmpPostgrustError::MigrationsFailed)?;
403
404        Ok(())
405    }
406
407    /// Start a new postgresql instance and return a process guard that will ensure it is cleaned
408    /// up when dropped.
409    #[instrument(skip(self))]
410    pub fn new_instance(&self) -> TmpPostgrustResult<synchronous::ProcessGuard> {
411        let data_directory = Builder::new()
412            .prefix("tmp-postgrust-db")
413            .tempdir()
414            .map_err(TmpPostgrustError::CreateDataDirFailed)?;
415        let data_directory_path = data_directory.path();
416
417        set_permissions(
418            &data_directory,
419            metadata(self.cache_dir.path()).unwrap().permissions(),
420        )
421        .unwrap();
422        synchronous::exec_copy_dir(self.cache_dir.path(), data_directory_path)?;
423
424        if !data_directory_path.join("PG_VERSION").exists() {
425            return Err(TmpPostgrustError::EmptyDataDirectory);
426        }
427
428        self.start_postgresql(&Arc::new(data_directory))
429    }
430
431    #[instrument(skip(self))]
432    fn start_postgresql(
433        &self,
434        dir: &Arc<TempDir>,
435    ) -> TmpPostgrustResult<synchronous::ProcessGuard> {
436        File::create(dir.path().join("postgresql.conf"))
437            .map_err(TmpPostgrustError::CreateConfigFailed)?
438            .write_all(self.config.as_bytes())
439            .map_err(TmpPostgrustError::CreateConfigFailed)?;
440
441        let port = self
442            .next_port
443            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
444
445        synchronous::chown_to_non_root(dir.path())?;
446        let mut postgres_process_handle =
447            synchronous::start_postgres_subprocess(&self.binaries.postgres, dir.path(), port)?;
448        let stdout = postgres_process_handle.stdout.take().unwrap();
449        let stderr = postgres_process_handle.stderr.take().unwrap();
450
451        let stdout_reader = BufReader::new(stdout).lines();
452        let mut stderr_reader = BufReader::new(stderr).lines();
453
454        while let Some(Ok(line)) = stderr_reader.next() {
455            debug!("Postgresql: {}", line);
456            if line.contains("database system is ready to accept connections") {
457                info!("temporary database system is read to accept connections");
458                break;
459            }
460        }
461
462        Ok(synchronous::ProcessGuard {
463            stdout_reader: Some(stdout_reader),
464            stderr_reader: Some(stderr_reader),
465            port,
466            db_name: TMP_POSTGRUST_DB_NAME.to_string(),
467            user_name: TMP_POSTGRUST_USER_NAME.to_string(),
468            postgres_process: postgres_process_handle,
469            _data_directory: Arc::clone(dir),
470            _cache_directory: Arc::clone(&self.cache_dir),
471            socket_dir: Arc::clone(&self.socket_dir),
472            createdb_bin: self.binaries.createdb.clone(),
473        })
474    }
475
476    /// Start a new postgresql instance and return a process guard that will ensure it is cleaned
477    /// up when dropped.
478    #[cfg(feature = "tokio-process")]
479    #[instrument(skip(self))]
480    pub async fn new_instance_async(&self) -> TmpPostgrustResult<asynchronous::ProcessGuard> {
481        use tokio::fs::{metadata, set_permissions};
482
483        let data_directory = Builder::new()
484            .prefix("tmp-postgrust-db")
485            .tempdir()
486            .map_err(TmpPostgrustError::CreateDataDirFailed)?;
487        let data_directory_path = data_directory.path();
488
489        set_permissions(
490            &data_directory,
491            metadata(self.cache_dir.path()).await.unwrap().permissions(),
492        )
493        .await
494        .unwrap();
495        asynchronous::exec_copy_dir(self.cache_dir.path(), data_directory_path).await?;
496
497        if !data_directory_path.join("PG_VERSION").exists() {
498            return Err(TmpPostgrustError::EmptyDataDirectory);
499        }
500
501        self.start_postgresql_async(&Arc::new(data_directory)).await
502    }
503
504    #[cfg(feature = "tokio-process")]
505    #[instrument(skip(self))]
506    async fn start_postgresql_async(
507        &self,
508        dir: &Arc<TempDir>,
509    ) -> TmpPostgrustResult<asynchronous::ProcessGuard> {
510        use tokio::io::AsyncBufReadExt;
511
512        File::create(dir.path().join("postgresql.conf"))
513            .map_err(TmpPostgrustError::CreateConfigFailed)?
514            .write_all(self.config.as_bytes())
515            .map_err(TmpPostgrustError::CreateConfigFailed)?;
516
517        let port = self
518            .next_port
519            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
520
521        asynchronous::chown_to_non_root(dir.path()).await?;
522
523        let permit = crate::asynchronous::MAX_CONCURRENT_PROCESSES
524            .get_or_init(|| tokio::sync::Semaphore::new(num_cpus::get()))
525            .acquire()
526            .await
527            .map_err(TmpPostgrustError::AsyncProcessPermitAcquireError)?;
528
529        let mut postgres_process_handle =
530            asynchronous::start_postgres_subprocess(&self.binaries.postgres, dir.path(), port)?;
531        let stdout = postgres_process_handle.stdout.take().unwrap();
532        let stderr = postgres_process_handle.stderr.take().unwrap();
533
534        let stdout_reader = tokio::io::BufReader::new(stdout).lines();
535        let mut stderr_reader = tokio::io::BufReader::new(stderr).lines();
536
537        while let Some(line) = stderr_reader.next_line().await.unwrap() {
538            debug!("Postgresql: {}", line);
539            if line.contains("database system is ready to accept connections") {
540                info!("temporary database system is read to accept connections");
541                break;
542            }
543        }
544
545        Ok(asynchronous::ProcessGuard {
546            stdout_reader: Some(stdout_reader),
547            stderr_reader: Some(stderr_reader),
548            port,
549            db_name: TMP_POSTGRUST_DB_NAME.to_string(),
550            user_name: TMP_POSTGRUST_USER_NAME.to_string(),
551            _data_directory: Arc::clone(dir),
552            _cache_directory: Arc::clone(&self.cache_dir),
553            socket_dir: Arc::clone(&self.socket_dir),
554            createdb_bin: self.binaries.createdb.clone(),
555            postgres_process: postgres_process_handle,
556            _process_permit: permit,
557        })
558    }
559}
560
561#[cfg(test)]
562mod tests {
563    use super::*;
564
565    use test_log::test;
566    use tokio_postgres::NoTls;
567    use tracing::error;
568
569    #[test(tokio::test)]
570    async fn it_works() {
571        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
572            disable_fsync: false,
573            ..Default::default()
574        })
575        .expect("failed to create factory");
576
577        let postgresql_proc = factory
578            .new_instance()
579            .expect("failed to create a new instance");
580
581        let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls)
582            .await
583            .expect("failed to connect to postgresql");
584
585        tokio::spawn(async move {
586            if let Err(e) = conn.await {
587                error!("connection error: {}", e);
588            }
589        });
590
591        client.query("SELECT 1;", &[]).await.unwrap();
592    }
593
594    #[test(tokio::test)]
595    async fn it_works_fsync_disabled() {
596        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
597            disable_fsync: true,
598            ..Default::default()
599        })
600        .expect("failed to create factory");
601
602        let postgresql_proc = factory
603            .new_instance()
604            .expect("failed to create a new instance");
605
606        let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls)
607            .await
608            .expect("failed to connect to postgresql");
609
610        tokio::spawn(async move {
611            if let Err(e) = conn.await {
612                error!("connection error: {}", e);
613            }
614        });
615
616        client.query("SELECT 1;", &[]).await.unwrap();
617    }
618
619    #[cfg(feature = "tokio-process")]
620    #[test(tokio::test)]
621    async fn it_works_async() {
622        let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig {
623            disable_fsync: false,
624            ..Default::default()
625        })
626        .await
627        .expect("failed to create factory");
628
629        let postgresql_proc = factory
630            .new_instance_async()
631            .await
632            .expect("failed to create a new instance");
633
634        let (client, conn) = tokio_postgres::connect(&postgresql_proc.connection_string(), NoTls)
635            .await
636            .expect("failed to connect to postgresql");
637
638        tokio::spawn(async move {
639            if let Err(e) = conn.await {
640                error!("connection error: {}", e);
641            }
642        });
643
644        client.query("SELECT 1;", &[]).await.unwrap();
645    }
646
647    #[test(tokio::test)]
648    async fn two_simulatenous_processes() {
649        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
650            disable_fsync: false,
651            ..Default::default()
652        })
653        .expect("failed to create factory");
654
655        let proc1 = factory
656            .new_instance()
657            .expect("failed to create a new instance");
658
659        let proc2 = factory
660            .new_instance()
661            .expect("failed to create a new instance");
662
663        let (client1, conn1) = tokio_postgres::connect(&proc1.connection_string(), NoTls)
664            .await
665            .expect("failed to connect to postgresql");
666
667        tokio::spawn(async move {
668            if let Err(e) = conn1.await {
669                error!("connection error: {}", e);
670            }
671        });
672
673        let (client2, conn2) = tokio_postgres::connect(&proc2.connection_string(), NoTls)
674            .await
675            .expect("failed to connect to postgresql");
676
677        tokio::spawn(async move {
678            if let Err(e) = conn2.await {
679                error!("connection error: {}", e);
680            }
681        });
682
683        client1.query("SELECT 1;", &[]).await.unwrap();
684        client2.query("SELECT 1;", &[]).await.unwrap();
685    }
686
687    #[cfg(feature = "tokio-process")]
688    #[test(tokio::test)]
689    async fn two_simulatenous_processes_async() {
690        let factory = TmpPostgrustFactory::try_new_async(&TmpPostgrustFactoryConfig {
691            disable_fsync: false,
692            ..Default::default()
693        })
694        .await
695        .expect("failed to create factory");
696
697        let proc1 = factory
698            .new_instance_async()
699            .await
700            .expect("failed to create a new instance");
701
702        let proc2 = factory
703            .new_instance_async()
704            .await
705            .expect("failed to create a new instance");
706
707        let (client1, conn1) = tokio_postgres::connect(&proc1.connection_string(), NoTls)
708            .await
709            .expect("failed to connect to postgresql");
710
711        tokio::spawn(async move {
712            if let Err(e) = conn1.await {
713                error!("connection error: {}", e);
714            }
715        });
716
717        let (client2, conn2) = tokio_postgres::connect(&proc2.connection_string(), NoTls)
718            .await
719            .expect("failed to connect to postgresql");
720
721        tokio::spawn(async move {
722            if let Err(e) = conn2.await {
723                error!("connection error: {}", e);
724            }
725        });
726
727        client1.query("SELECT 1;", &[]).await.unwrap();
728        client2.query("SELECT 1;", &[]).await.unwrap();
729    }
730
731    #[cfg(feature = "tokio-process")]
732    #[test(tokio::test)]
733    async fn default_process_factory_1() {
734        let proc = new_default_process_async().await.unwrap();
735
736        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
737            .await
738            .expect("failed to connect to postgresql");
739
740        tokio::spawn(async move {
741            if let Err(e) = conn.await {
742                error!("connection error: {}", e);
743            }
744        });
745
746        // Chance to catch concurrent tests or database that have already been used.
747        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
748    }
749
750    #[cfg(feature = "tokio-process")]
751    #[test(tokio::test)]
752    async fn default_process_factory_2() {
753        let proc = new_default_process_async().await.unwrap();
754
755        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
756            .await
757            .expect("failed to connect to postgresql");
758
759        tokio::spawn(async move {
760            if let Err(e) = conn.await {
761                error!("connection error: {}", e);
762            }
763        });
764
765        // Chance to catch concurrent tests or database that have already been used.
766        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
767    }
768
769    #[cfg(feature = "tokio-process")]
770    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
771    async fn default_process_factory_multithread_1() {
772        let proc = new_default_process_async().await.unwrap();
773
774        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
775            .await
776            .expect("failed to connect to postgresql");
777
778        tokio::spawn(async move {
779            if let Err(e) = conn.await {
780                error!("connection error: {}", e);
781            }
782        });
783
784        // Chance to catch concurrent tests or database that have already been used.
785        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
786    }
787
788    #[cfg(feature = "tokio-process")]
789    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
790    async fn default_process_factory_multithread_2() {
791        let proc = new_default_process_async().await.unwrap();
792
793        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
794            .await
795            .expect("failed to connect to postgresql");
796
797        tokio::spawn(async move {
798            if let Err(e) = conn.await {
799                error!("connection error: {}", e);
800            }
801        });
802
803        // Chance to catch concurrent tests or database that have already been used.
804        client.execute("CREATE TABLE lock ();", &[]).await.unwrap();
805    }
806
807    #[test(tokio::test)]
808    async fn explicit_bin_dir_works() {
809        let postgres_path = crate::search::find_postgresql_command(None, "postgres").expect(
810            "cannot derive bin_dir: postgres not found via PATH or known install locations",
811        );
812        let bin_dir = postgres_path
813            .parent()
814            .expect("postgres path has no parent dir")
815            .to_owned();
816
817        let factory = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
818            disable_fsync: true,
819            postgresql_bin_dir: Some(bin_dir),
820        })
821        .expect("failed to create factory with explicit bin dir");
822
823        let proc = factory
824            .new_instance()
825            .expect("failed to create a new instance");
826
827        let (client, conn) = tokio_postgres::connect(&proc.connection_string(), NoTls)
828            .await
829            .expect("failed to connect to postgresql");
830
831        tokio::spawn(async move {
832            if let Err(e) = conn.await {
833                error!("connection error: {}", e);
834            }
835        });
836
837        client.query("SELECT 1;", &[]).await.unwrap();
838    }
839
840    #[test]
841    fn bogus_bin_dir_returns_error() {
842        let result = TmpPostgrustFactory::try_new(&TmpPostgrustFactoryConfig {
843            disable_fsync: true,
844            postgresql_bin_dir: Some(std::path::PathBuf::from("/nonexistent/bin/dir")),
845        });
846        assert!(
847            matches!(
848                result,
849                Err(TmpPostgrustError::PostgresCommandNotFound { .. })
850            ),
851            "expected PostgresCommandNotFound, got: {:?}",
852            result
853        );
854    }
855}
tmp_postgrust/lib.rs

tmp_postgrust/
lib.rs
