Lines
99.26 %
Functions
50 %
Branches
100 %
//! Shared fixtures for the fleet-migration suites.
//!
//! Each suite is its own test binary, and that is deliberate: `server::db`'s
//! admin pool is a process-static `OnceCell` bound to the first runtime that
//! touches it, so scenarios sharing a process must also share one `#[tokio::test]`
//! — which hides them behind a single name in the coverage output. Separate
//! binaries give each scenario its own process, its own pool, and its own
//! reportable name.
//! Because the binaries run concurrently against ONE global directory, every
//! assertion here is per-tenant-id. Never assert on fleet-wide counts or on
//! `FleetReport::is_healthy`: another suite's deliberately-broken tenant is
//! visible to this one's run.
//! It lives in the lib rather than `tests/fleet_support/mod.rs` because that
//! module would be compiled into every suite, and each suite uses only part of
//! it — producing dead-code warnings in the binaries that use the rest.
use sqlx::postgres::PgPoolOptions;
use sqlx::{Executor, Row};
use uuid::Uuid;
pub fn admin_id() -> Uuid {
server::bootstrap::ADMIN_USER_ID
.parse()
.expect("admin id is a uuid")
}
fn database_url() -> String {
std::env::var("DATABASE_URL").expect("DATABASE_URL must be set for fleet tests")
pub fn dsn_for(database_name: &str) -> String {
let url = database_url();
let (prefix, rest) = url.split_once("://").expect("scheme");
let authority = rest.split('/').next().expect("authority");
format!("{prefix}://{authority}/{database_name}")
pub async fn admin_connect() -> sqlx::PgPool {
PgPoolOptions::new()
.max_connections(2)
.connect(&database_url())
.await
.expect("connect to admin DB")
async fn drop_database(admin: &sqlx::PgPool, database_name: &str) {
let _ = admin
.execute(sqlx::query(sqlx::AssertSqlSafe(format!(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity \
WHERE datname = '{database_name}' AND pid <> pg_backend_pid()"
))))
.await;
"DROP DATABASE IF EXISTS \"{database_name}\""
pub async fn create_database(admin: &sqlx::PgPool, database_name: &str) {
admin
"CREATE DATABASE \"{database_name}\""
.expect("create tenant database");
/// Registers a tenant in the directory, optionally creating its database.
///
/// A tenant whose database is never created is the "unreachable" case: the
/// directory says where it lives and nothing is there.
pub async fn register_tenant(admin: &sqlx::PgPool, label: &str, create: bool) -> (Uuid, String) {
let id = Uuid::new_v4();
let database_name = format!("nomi_fleet_{}", id.simple());
if create {
create_database(admin, &database_name).await;
.execute(
sqlx::query(
"INSERT INTO users (id, user_name, email, user_password, database_name) \
VALUES ($1, $2, $3, 'x', $4)",
)
.bind(id)
.bind(label)
.bind(format!("{label}-{id}@example.test"))
.bind(&database_name),
.expect("register tenant");
(id, database_name)
pub async fn forget_tenant(admin: &sqlx::PgPool, id: Uuid, database_name: &str) {
drop_database(admin, database_name).await;
let _ = sqlx::query("DELETE FROM users WHERE id = $1")
.execute(admin)
/// The quarantine reason recorded for a user, if any.
/// A row that does not exist reads the same as one carrying no quarantine. The
/// seeded admin only exists after `boot()` has run, and these suites migrate
/// without seeding, so requiring the row would make them depend on whatever a
/// previous run happened to leave behind.
pub async fn quarantine_of(admin: &sqlx::PgPool, id: Uuid) -> Option<String> {
sqlx::query("SELECT quarantine_reason FROM users WHERE id = $1")
.fetch_optional(admin)
.expect("read quarantine")
.and_then(|row| row.get::<Option<String>, _>(0))
/// How many migrations a tenant has applied.
pub async fn applied_migrations(database_name: &str) -> i64 {
let Ok(pool) = PgPoolOptions::new()
.max_connections(1)
.connect(&dsn_for(database_name))
else {
return -1;
};
let applied = sqlx::query_scalar("SELECT count(*) FROM _sqlx_migrations WHERE success")
.fetch_one(&pool)
.unwrap_or(0);
pool.close().await;
applied
/// The whole embedded set, not merely "some". Counting "at least one" would pass
/// a tenant stuck on 0001 with every later migration skipped — exactly the
/// failure the fleet migrator exists to prevent.
pub fn expected_migrations() -> i64 {
i64::try_from(server::db::MIGRATOR.iter().count()).expect("migration count")
pub async fn is_fully_migrated(database_name: &str) -> bool {
applied_migrations(database_name).await == expected_migrations()
/// Did this run account for *our* tenant? Per-id, because other suites run
/// concurrently against the same directory.
pub fn accounted_for(report: &server::fleet::FleetReport, id: Uuid) -> bool {
!report.quarantined.iter().any(|(q, _)| *q == id) && !report.unresolved.contains(&id)
pub fn quarantine_reason_in(report: &server::fleet::FleetReport, id: Uuid) -> Option<String> {
report
.quarantined
.iter()
.find(|(q, _)| *q == id)
.map(|(_, reason)| reason.clone())
/// Brings a tenant up to only the FIRST `count` migrations, leaving it on a
/// genuinely older schema.
/// Built by copying that many migration files into a scratch directory and
/// pointing a fresh `Migrator` at it: sqlx's embedded migrator has no
/// run-to-version API, and slicing by array index would be wrong here anyway —
/// the set has a deliberate version gap (1, 2, 3, 5, 6), so index N and version
/// N are different things.
pub async fn migrate_to_first(database_name: &str, count: usize) {
let source = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()
.expect("workspace root")
.join("migrations");
let mut files: Vec<_> = std::fs::read_dir(&source)
.expect("read migrations")
.filter_map(Result::ok)
.map(|e| e.file_name().to_string_lossy().into_owned())
.filter(|n| n.ends_with(".sql"))
.collect();
files.sort();
let scratch = std::env::temp_dir().join(format!("nomi-partial-{}", Uuid::new_v4().simple()));
std::fs::create_dir_all(&scratch).expect("scratch dir");
for name in files.iter().take(count) {
std::fs::copy(source.join(name), scratch.join(name)).expect("copy migration");
let partial = sqlx::migrate::Migrator::new(scratch.as_path())
.expect("build partial migrator");
let pool = PgPoolOptions::new()
.expect("connect to tenant");
partial.run(&pool).await.expect("apply partial migrations");
std::fs::remove_dir_all(&scratch).ok();