mod cache_integration;
mod cleanup;
mod connection;
mod delegation;
mod guard;
mod handle;
mod installation;
mod lifecycle;
pub(crate) mod panic_utils;
mod runtime;
mod runtime_mode;
mod shutdown;
#[cfg(unix)]
mod shutdown_hook;
#[cfg(all(
unix,
any(doc, test, feature = "cluster-unit-tests", feature = "dev-worker")
))]
pub use self::shutdown_hook::{process_is_running, read_postmaster_pid};
mod startup;
mod temporary_database;
mod worker_invoker;
mod worker_operation;
pub use self::connection::{ConnectionMetadata, TestClusterConnection};
pub use self::guard::ClusterGuard;
pub use self::handle::ClusterHandle;
pub use self::lifecycle::DatabaseName;
pub use self::temporary_database::TemporaryDatabase;
#[cfg(any(doc, test, feature = "cluster-unit-tests", feature = "dev-worker"))]
pub use self::worker_invoker::WorkerInvoker;
#[doc(hidden)]
pub use self::worker_operation::WorkerOperation;
use self::runtime::build_runtime;
use self::runtime_mode::ClusterRuntime;
pub(crate) use self::startup::setup_postgres_only;
#[cfg(feature = "async-api")]
use self::startup::start_postgres_async;
use self::startup::{cache_config_from_bootstrap, start_postgres};
use crate::bootstrap_for_tests;
use crate::env::ScopedEnv;
use crate::error::BootstrapResult;
use crate::observability::LOG_TARGET;
use std::ops::Deref;
use tracing::info_span;
#[derive(Debug)]
pub struct TestCluster {
pub(crate) handle: ClusterHandle,
pub(crate) guard: ClusterGuard,
}
#[cfg(feature = "async-api")]
enum StopAsyncPath {
WorkerManaged,
InProcess(Box<postgresql_embedded::PostgreSQL>),
Noop,
}
impl TestCluster {
pub fn new() -> BootstrapResult<Self> {
let (handle, guard) = Self::new_split()?;
Ok(Self { handle, guard })
}
pub fn new_split() -> BootstrapResult<(ClusterHandle, ClusterGuard)> {
let span = info_span!(target: LOG_TARGET, "test_cluster");
let (runtime, env_vars, env_guard, outcome) = {
let _entered = span.enter();
let initial_bootstrap = bootstrap_for_tests()?;
let cache_config = cache_config_from_bootstrap(&initial_bootstrap);
let runtime = build_runtime()?;
let env_vars = initial_bootstrap.environment.to_env();
let env_guard = ScopedEnv::apply(&env_vars);
let outcome = start_postgres(&runtime, initial_bootstrap, &env_vars, &cache_config)?;
(runtime, env_vars, env_guard, outcome)
};
let handle = ClusterHandle::new(outcome.bootstrap.clone());
let guard = ClusterGuard {
runtime: ClusterRuntime::Sync(runtime),
postgres: outcome.postgres,
bootstrap: outcome.bootstrap,
is_managed_via_worker: outcome.is_managed_via_worker,
env_vars,
worker_guard: None,
_env_guard: env_guard,
_cluster_span: span,
};
Ok((handle, guard))
}
#[cfg(feature = "async-api")]
pub async fn start_async() -> BootstrapResult<Self> {
let (handle, guard) = Self::start_async_split().await?;
Ok(Self { handle, guard })
}
#[cfg(feature = "async-api")]
pub async fn start_async_split() -> BootstrapResult<(ClusterHandle, ClusterGuard)> {
use tracing::Instrument;
let span = info_span!(target: LOG_TARGET, "test_cluster", async_mode = true);
let initial_bootstrap = bootstrap_for_tests()?;
let cache_config = cache_config_from_bootstrap(&initial_bootstrap);
let env_vars = initial_bootstrap.environment.to_env();
let env_guard = ScopedEnv::apply(&env_vars);
let outcome = Box::pin(start_postgres_async(
initial_bootstrap,
&env_vars,
&cache_config,
))
.instrument(span.clone())
.await?;
let handle = ClusterHandle::new(outcome.bootstrap.clone());
let guard = ClusterGuard {
runtime: ClusterRuntime::Async,
postgres: outcome.postgres,
bootstrap: outcome.bootstrap,
is_managed_via_worker: outcome.is_managed_via_worker,
env_vars,
worker_guard: None,
_env_guard: env_guard,
_cluster_span: span,
};
Ok((handle, guard))
}
#[doc(hidden)]
#[must_use]
pub fn with_worker_guard(self, worker_guard: Option<ScopedEnv>) -> Self {
Self {
handle: self.handle,
guard: self.guard.with_worker_guard(worker_guard),
}
}
#[cfg(feature = "async-api")]
fn stop_async_path(&mut self) -> StopAsyncPath {
if self.guard.is_managed_via_worker {
StopAsyncPath::WorkerManaged
} else if let Some(postgres) = self.guard.postgres.take() {
StopAsyncPath::InProcess(Box::new(postgres))
} else {
StopAsyncPath::Noop
}
}
#[cfg(feature = "async-api")]
pub async fn stop_async(mut self) -> BootstrapResult<()> {
let context = shutdown::stop_context(self.handle.settings());
shutdown::log_async_stop(&context, self.guard.is_managed_via_worker);
match self.stop_async_path() {
StopAsyncPath::WorkerManaged => {
shutdown::stop_worker_managed_async(
&self.guard.bootstrap,
&self.guard.env_vars,
&context,
)
.await
}
StopAsyncPath::InProcess(postgres) => {
let cleanup = shutdown::InProcessCleanup {
cleanup_mode: self.guard.bootstrap.cleanup_mode,
settings: &self.guard.bootstrap.settings,
context: &context,
};
shutdown::stop_in_process_async(
*postgres,
self.guard.bootstrap.shutdown_timeout,
cleanup,
)
.await
}
StopAsyncPath::Noop => Ok(()),
}
}
}
impl Deref for TestCluster {
type Target = ClusterHandle;
fn deref(&self) -> &Self::Target {
&self.handle
}
}
#[cfg(test)]
mod mod_tests;
#[cfg(all(test, feature = "cluster-unit-tests"))]
mod drop_logging_tests;
#[cfg(all(test, not(feature = "cluster-unit-tests")))]
#[path = "../../tests/test_cluster.rs"]
mod test_cluster_tests;