use std::io::Write;
use std::path::Path;
use crate::core::inventory;
use crate::sql::Pool;
use super::error::MigrateError;
use super::file::{self, DataOp, Migration, Operation};
use super::runner;
use super::snapshot::SchemaSnapshot;
pub async fn run(
pool: &Pool,
dir: &Path,
args: impl IntoIterator<Item = String>,
) -> Result<(), MigrateError> {
let mut stdout = std::io::stdout();
run_with_writer(pool, dir, args, &mut stdout).await
}
pub fn run_pool_free<W: Write>(
args: &[String],
writer: &mut W,
) -> Option<Result<(), MigrateError>> {
let cmd = args.first().map_or("", String::as_str);
Some(match cmd {
"" | "--help" | "-h" | "help" => print_help(writer).map_err(Into::into),
"version" | "--version" => version_cmd(writer),
"docs" => docs_cmd(writer),
"db:info" => db_info_cmd(writer),
"startapp" => startapp(&args[1..], writer),
c if MAKE_VERBS.contains(&c) => make_cmd(c, &args[1..], writer, compiled_feature),
"showurls" => showurls_cmd(&args[1..], writer),
"showmodels" => showmodels_cmd(&args[1..], writer),
_ => return None,
})
}
pub async fn run_with_writer<W: Write + Send>(
pool: &Pool,
dir: &Path,
args: impl IntoIterator<Item = String>,
writer: &mut W,
) -> Result<(), MigrateError> {
let args: Vec<String> = args.into_iter().collect();
let cmd = args.first().map_or("", String::as_str);
match cmd {
"" | "--help" | "-h" | "help" => {
print_help(writer)?;
Ok(())
}
"makemigrations" => makemigrations(dir, &args[1..], writer),
"migrate" => migrate(pool, dir, &args[1..], writer).await,
"downgrade" => downgrade(pool, dir, &args[1..], writer).await,
"showmigrations" | "status" => showmigrations(pool, dir, writer).await,
"sqlmigrate" => sqlmigrate_cmd(pool, dir, &args[1..], writer),
"forget-pending" => forget_pending_cmd(pool, dir, &args[1..], writer).await,
"startapp" => startapp(&args[1..], writer),
"add-data-op" => add_data_op_cmd(dir, &args[1..], writer),
c if MAKE_VERBS.contains(&c) => make_cmd(c, &args[1..], writer, compiled_feature),
"about" => about_cmd(pool, writer).await,
"check" => check_cmd(pool, dir, &args[1..], writer).await,
"docs" => docs_cmd(writer),
"version" | "--version" => version_cmd(writer),
"db:dump" => db_dump_cmd(&args[1..]),
"db:restore" => db_restore_cmd(&args[1..], writer),
"db:info" => db_info_cmd(writer),
"dumpdata" => dumpdata_cmd(pool, &args[1..], DataScope::All, writer).await,
"loaddata" => loaddata_cmd(pool, &args[1..], DataScope::All, writer).await,
"showurls" => showurls_cmd(&args[1..], writer),
"showmodels" => showmodels_cmd(&args[1..], writer),
#[cfg(feature = "admin")]
"create-admin" => crate::admin::create_admin_cmd(pool, &args[1..], writer).await,
#[cfg(feature = "admin")]
"reset-password" => crate::admin::reset_password_cmd(pool, &args[1..], writer).await,
#[cfg(feature = "admin")]
"change-password" => crate::admin::change_password_cmd(pool, &args[1..], writer).await,
#[cfg(feature = "admin")]
"set-admin-active" => crate::admin::set_admin_active_cmd(pool, &args[1..], writer).await,
#[cfg(all(feature = "admin", feature = "totp"))]
"reset-admin-totp" => crate::admin::reset_admin_totp_cmd(pool, &args[1..], writer).await,
"flush" => flush_cmd(pool, &args[1..], FlushScope::All, writer).await,
"prune" => prune_cmd(pool, &args[1..], writer).await,
"audit-cleanup" => audit_cleanup_cmd(pool, &args[1..], writer).await,
#[cfg(feature = "cache")]
verb @ ("clear-cache" | "clearsessions") => {
clear_cache_cmd(pool, verb, &args[1..], writer).await
}
#[cfg(feature = "cache")]
"createcachetable" | "create-cache-table" => {
createcachetable_cmd(pool, &args[1..], writer).await
}
"sendtestemail" => sendtestemail_cmd(&args[1..], writer).await,
"inspectdb" => super::inspectdb::inspectdb_cmd(pool, &args[1..], writer).await,
other => Err(MigrateError::Validation(format!(
"unknown subcommand: `{other}` (run with --help for usage)"
))),
}
}
fn print_help<W: Write>(w: &mut W) -> std::io::Result<()> {
writeln!(w, "rustango::manage — migration and admin command runner\n")?;
writeln!(w, "USAGE:")?;
writeln!(w, " manage <COMMAND> [args]\n")?;
writeln!(w, "COMMANDS:")?;
writeln!(w, " makemigrations [name]")?;
writeln!(
w,
" Diff the inventory registry against the latest snapshot"
)?;
writeln!(
w,
" and write the next migration file. `name` overrides the"
)?;
writeln!(w, " auto-derived suffix.\n")?;
writeln!(w, " makemigrations --empty <name>")?;
writeln!(
w,
" Write an empty migration scaffold (`forward: []`) for"
)?;
writeln!(
w,
" hand-authored data migrations. Edit the JSON to add"
)?;
writeln!(w, " `data` ops with sql + reverse_sql.\n")?;
writeln!(w, " migrate")?;
writeln!(w, " Apply every pending migration in lex order.\n")?;
writeln!(w, " migrate <target>")?;
writeln!(
w,
" Forward or back to <target>. `zero` unapplies every"
)?;
writeln!(w, " applied migration.\n")?;
writeln!(w, " migrate --dry-run")?;
writeln!(
w,
" Print the SQL each pending migration would run; never"
)?;
writeln!(
w,
" writes. Reads the ledger so the preview is accurate.\n"
)?;
writeln!(w, " migrate --squash")?;
writeln!(
w,
" Delete every pending (un-applied) migration JSON and"
)?;
writeln!(
w,
" regenerate a single fresh diff via makemigrations. Dev-"
)?;
writeln!(
w,
" iteration escape hatch when an evolving model produces a"
)?;
writeln!(
w,
" pending migration the validator rejects (e.g. AddColumn"
)?;
writeln!(
w,
" NOT NULL with no default). Refuses to touch applied rows.\n"
)?;
writeln!(w, " downgrade [N]")?;
writeln!(w, " Step back N applied migrations (default 1).\n")?;
writeln!(w, " showmigrations | status")?;
writeln!(w, " List migrations with [X]/[ ] applied marker.\n")?;
writeln!(w, " sqlmigrate <name>")?;
writeln!(
w,
" Print the SQL the named migration would emit when applied.\n"
)?;
writeln!(w, " forget-pending <name>")?;
writeln!(
w,
" Delete a migration JSON that has NOT been applied yet,"
)?;
writeln!(
w,
" so the next `makemigrations` regenerates the diff."
)?;
writeln!(
w,
" Refuses if the migration is recorded in the ledger.\n"
)?;
writeln!(
w,
" add-data-op --sql <SQL> [--reverse-sql <SQL>] [--name <name>] [--to <migration> | --scope <s>]"
)?;
writeln!(
w,
" Add a data transformation op (up + optional down)."
)?;
writeln!(w, " --sql Forward SQL to run (required).")?;
writeln!(
w,
" --reverse-sql Rollback SQL. Omit for irreversible ops."
)?;
writeln!(
w,
" --name Name suffix for the new migration file."
)?;
writeln!(
w,
" --to Append to an existing migration instead of creating one.\n"
)?;
writeln!(w, " about")?;
writeln!(w, " Print framework version, registered models/apps,")?;
writeln!(w, " and detected backend configuration.\n")?;
writeln!(w, " check [--deploy]")?;
writeln!(
w,
" Run system audits — pending migrations, missing models, common"
)?;
writeln!(
w,
" misconfigurations. With --deploy: production hardening checks."
)?;
writeln!(w, " Exits non-zero on any error-level finding.\n")?;
writeln!(w, " flush [--yes] [--app <label>] [--model <name>]")?;
writeln!(
w,
" Wipe all rows from registered model tables. Schema + migrations"
)?;
writeln!(
w,
" ledger stay intact. Without --yes, prints the planned"
)?;
writeln!(w, " action and exits (no DB write).\n")?;
writeln!(w, " prune [--model <name>] [--except <name>] [--pretend]")?;
writeln!(w, " Bulk-delete stale rows from every Prunable model.")?;
writeln!(
w,
" --pretend counts matches without deleting; --model / --except"
)?;
writeln!(w, " narrow the scope (repeatable).\n")?;
#[cfg(feature = "admin")]
{
writeln!(
w,
" create-admin <username> [--password-stdin | --generate] [--superuser]"
)?;
writeln!(
w,
" Create an admin user; prompts for the password by default.\n"
)?;
writeln!(
w,
" reset-password <username> | change-password <username>"
)?;
writeln!(
w,
" Set an admin user's password; change-password checks the old one.\n"
)?;
writeln!(w, " set-admin-active <username> --on|--off")?;
writeln!(
w,
" Turn an admin user off (ends their sessions) or back on.\n"
)?;
}
#[cfg(all(feature = "admin", feature = "totp"))]
{
writeln!(w, " reset-admin-totp <username>")?;
writeln!(
w,
" Remove an admin user's two-factor device (lost phone).\n"
)?;
}
writeln!(w, " audit-cleanup --days <N> | --keep-last <N> [--yes]")?;
writeln!(
w,
" Trim the audit log: entries older than N days, or all but the"
)?;
writeln!(
w,
" N newest per row. N <= 0 deletes everything and needs --yes.\n"
)?;
writeln!(w, " clear-cache [--table <name>]")?;
writeln!(w, " (alias: clearsessions)")?;
writeln!(
w,
" Purge expired rows from a DatabaseCache-backed table."
)?;
writeln!(w, " Default table: rustango_cache.\n")?;
writeln!(w, " createcachetable [--table <name>]")?;
writeln!(w, " (alias: create-cache-table)")?;
writeln!(
w,
" Idempotent CREATE TABLE IF NOT EXISTS for DatabaseCache."
)?;
writeln!(w, " Default table: rustango_cache. Safe to re-run.\n")?;
writeln!(
w,
" sendtestemail --to <addr> [--from <addr>] [--subject <text>]"
)?;
writeln!(
w,
" Send a test message through the [mail] backend (console / memory /"
)?;
writeln!(
w,
" null / smtp). Use to verify SMTP credentials before deploy.\n"
)?;
writeln!(w, " showmodels [--format plain|json] [--app <label>]")?;
writeln!(
w,
" Print every model registered via #[derive(Model)] + inventory."
)?;
writeln!(
w,
" Confirms registration / debugs missing-model surprises.\n"
)?;
writeln!(w, " showurls [--format plain|json]")?;
writeln!(
w,
" Print every named URL pattern registered via `register_url!`."
)?;
writeln!(
w,
" Useful for debugging routing + auditing `{{% url %}}` references.\n"
)?;
writeln!(w, " loaddata <fixture.json> [--fail-fast]")?;
writeln!(
w,
" Insert every row in the JSON fixture. Default skips failing"
)?;
writeln!(
w,
" rows with a warning; --fail-fast aborts on the first failure.\n"
)?;
writeln!(
w,
" dumpdata [--model <name>] [--exclude <name>] [--indent <N>]"
)?;
writeln!(w, " Export every registered model's rows as a JSON")?;
writeln!(
w,
" fixture (`[{{\"model\": \"app.Model\", \"pk\": N, \"fields\": {{...}}}}]`)."
)?;
writeln!(
w,
" --model limits to a single model; pass multiple times for a set.\n"
)?;
writeln!(w, " docs")?;
writeln!(
w,
" Open docs.rs/rustango in the default browser. Set \
RUSTANGO_NO_BROWSER to just print the URL.\n"
)?;
writeln!(w, " version | --version")?;
writeln!(w, " Print the rustango framework version.\n")?;
writeln!(
w,
" (To bootstrap a new project from scratch, install + run"
)?;
writeln!(
w,
" `cargo install cargo-rustango` then `cargo rustango new <name>`.)\n"
)?;
writeln!(
w,
" make:viewset <Name> [--model <Model>] [--tenant | --no-tenant]"
)?;
writeln!(
w,
" --tenant emits a `ViewSet::for_model(...).tenant_router(...)`"
)?;
writeln!(
w,
" shape that resolves a per-request connection via `Tenant`"
)?;
writeln!(
w,
" instead of baking a pool at mount time (required for tenancy)."
)?;
writeln!(
w,
" Auto-detected from Cargo.toml when the rustango dep enables"
)?;
writeln!(
w,
" `tenancy` — pass `--no-tenant` to override the auto-detection."
)?;
writeln!(w, " make:api_routes <app> [--tenant]")?;
writeln!(
w,
" Scaffold src/<app>/api_routes.rs — the per-app composer that"
)?;
writeln!(
w,
" merges every viewset's router into a single Router<()>."
)?;
writeln!(w, " make:serializer <Name> [--model <Model>]")?;
writeln!(w, " make:form <Name>")?;
writeln!(w, " make:job <Name>")?;
writeln!(w, " make:scheduled <Name>")?;
writeln!(w, " make:worker <Name>")?;
writeln!(w, " make:notification <Name>")?;
writeln!(w, " make:middleware <Name>")?;
writeln!(w, " make:test <Name>")?;
writeln!(
w,
" Scaffold a single source file with the chosen shape."
)?;
writeln!(
w,
" Writes to src/<snake_name>.rs (skips if exists).\n"
)?;
writeln!(
w,
" db:dump [--out <file>] [--data-only|--schema-only] [--no-owner]"
)?;
writeln!(
w,
" Run pg_dump against $DATABASE_URL. Default: prints SQL to"
)?;
writeln!(
w,
" stdout (omit --out to pipe). --data-only / --schema-only"
)?;
writeln!(
w,
" mirror pg_dump's flags. --no-owner skips OWNER lines.\n"
)?;
writeln!(w, " db:restore <file> [--clean --yes]")?;
writeln!(
w,
" Run psql against $DATABASE_URL with `\\i <file>`, in one transaction. With"
)?;
writeln!(
w,
" --clean, prepend a `DROP SCHEMA public CASCADE; CREATE SCHEMA public;`"
)?;
writeln!(
w,
" so the restore lands on a clean database; --yes confirms the drop.\n"
)?;
writeln!(w, " db:info")?;
writeln!(
w,
" Print the resolved DB URL (password redacted), detected"
)?;
writeln!(
w,
" backend, and which `postgres`/`mysql` Cargo features are"
)?;
writeln!(w, " compiled in. Read-only — does not connect.\n")?;
writeln!(w, " inspectdb [--schema <name>] [--table <name>]")?;
writeln!(
w,
" Connect to DATABASE_URL and emit `#[derive(Model)]`"
)?;
writeln!(
w,
" source for every base table in `--schema` (default `public`)."
)?;
writeln!(w, " Pipe to a file the user reviews + edits. Adopts")?;
writeln!(w, " rustango against an existing database")?;
writeln!(w, " without rewriting it.\n")?;
writeln!(w, " startapp <name> [--with-manage-bin]")?;
writeln!(w, " Scaffold an app module under src/<name>/")?;
writeln!(
w,
" (models.rs + views.rs + urls.rs + mod.rs). Idempotent;"
)?;
writeln!(
w,
" existing files are left untouched. With --with-manage-bin,"
)?;
writeln!(w, " also writes src/bin/manage.rs.")?;
Ok(())
}
fn makemigrations<W: Write>(dir: &Path, args: &[String], w: &mut W) -> Result<(), MigrateError> {
let mut empty = false;
let mut merge = false;
let mut drops = super::make::DropAndAdd::Refuse;
let mut name: Option<String> = None;
let mut app: Option<String> = None;
let mut scope_override: Option<crate::core::ModelScope> = None;
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--empty" => empty = true,
"--merge" => merge = true,
"--drop-and-add" => drops = super::make::DropAndAdd::Allow,
"--app" => {
app = Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--app requires an app name".into())
})?);
}
"--scope" => {
let raw = iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--scope requires \"registry\" or \"tenant\"".into())
})?;
scope_override =
Some(crate::core::ModelScope::from_str(&raw).ok_or_else(|| {
MigrateError::Validation(format!(
"--scope must be \"registry\" or \"tenant\", got {raw:?}"
))
})?);
}
"--help" | "-h" => {
writeln!(
w,
"makemigrations [name] diff registry, write next file in <dir>\n\
makemigrations --app <app> [name] diff one app, write to <project_root>/<app>/migrations/\n\
makemigrations --scope <s> [name] <s> = registry|tenant; one file with that MigrationScope\n\
makemigrations --empty <name> empty scaffold for data ops (--scope picks the chain)\n\
makemigrations --merge reconcile a divergent chain (two leaves\n\
on the same parent) by writing an empty\n\
NNNN_merge.json\n\
makemigrations --drop-and-add write a dropped column beside a new one of\n\
the same type as drop + add (deletes its data);\n\
refused by default as a likely rename\n\
\n\
In tenancy projects (any registered model with scope = \"registry\"),\n\
a flagless makemigrations splits the diff into TWO files — one for\n\
registry-scoped models, one for tenant-scoped — so framework tables\n\
don't bleed across scopes when migrate-tenants fans out."
)?;
return Ok(());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
if name.is_some() {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
name = Some(other.to_owned());
}
}
}
if merge {
if empty || name.is_some() || app.is_some() || scope_override.is_some() {
return Err(MigrateError::Validation(
"makemigrations --merge does not combine with --empty / --app / --scope / [name]"
.into(),
));
}
return run_merge(dir, w);
}
if empty {
let Some(n) = name else {
return Err(MigrateError::Validation(
"makemigrations --empty requires a name".into(),
));
};
let scope = match scope_override {
Some(crate::core::ModelScope::Registry) => super::MigrationScope::Registry,
Some(crate::core::ModelScope::Tenant) | None => super::MigrationScope::Tenant,
};
let mig = make_empty_in_scope(dir, &n, scope)?;
writeln!(
w,
"wrote {} (empty scaffold — fill in `forward` with data ops)",
file_path(dir, &mig.name).display()
)?;
return Ok(());
}
if let Some(app_name) = app {
let project_root = dir.parent().unwrap_or(dir);
match super::make::make_migrations_for_app_with(project_root, &app_name, name.as_deref(), drops)? {
Some(mig) => {
let app_dir = project_root.join(&app_name).join("migrations");
writeln!(w, "wrote {}", file_path(&app_dir, &mig.name).display())?;
for op in &mig.forward {
writeln!(w, " + {}", describe_op(op))?;
}
}
None => writeln!(
w,
"app `{app_name}`: no changes — models match latest snapshot (or no models with this app_label)"
)?,
}
return Ok(());
}
if let Some(scope) = scope_override {
return write_scoped_migration(dir, scope, name.as_deref(), drops, w);
}
if has_registry_models() {
let run = super::make::make_migrations_tenancy(dir, name.as_deref(), drops)?;
let project_root = dir.parent().unwrap_or(dir);
for (scope, m) in &run.system {
writeln!(
w,
"wrote {} (system/{} scope)",
file_path(&project_root.join("system").join("migrations"), &m.name).display(),
scope.as_str(),
)?;
for op in &m.forward {
writeln!(w, " + {}", describe_op(op))?;
}
}
for (scope, mig) in &run.app {
match mig {
Some(m) => {
writeln!(
w,
"wrote {} ({} scope)",
file_path(dir, &m.name).display(),
scope.as_str(),
)?;
for op in &m.forward {
writeln!(w, " + {}", describe_op(op))?;
}
}
None => writeln!(w, "no changes for {} scope", scope.as_str(),)?,
}
}
return Ok(());
}
match super::make::make_migrations_with(dir, name.as_deref(), drops)? {
Some(mig) => {
writeln!(w, "wrote {}", file_path(dir, &mig.name).display())?;
for op in &mig.forward {
writeln!(w, " + {}", describe_op(op))?;
}
}
None => writeln!(w, "no changes — registry matches latest snapshot")?,
}
Ok(())
}
fn has_registry_models() -> bool {
inventory::iter::<crate::core::ModelEntry>
.into_iter()
.any(|e| e.schema.scope == crate::core::ModelScope::Registry && e.schema.managed)
}
fn write_scoped_migration<W: Write>(
dir: &Path,
scope: crate::core::ModelScope,
name: Option<&str>,
drops: super::make::DropAndAdd,
w: &mut W,
) -> Result<(), MigrateError> {
match super::make::make_migrations_for_scope_with(dir, scope, name, drops)? {
Some(mig) => {
writeln!(
w,
"wrote {} ({} scope)",
file_path(dir, &mig.name).display(),
scope.as_str(),
)?;
for op in &mig.forward {
writeln!(w, " + {}", describe_op(op))?;
}
}
None => writeln!(
w,
"no changes — {} models match latest snapshot",
scope.as_str(),
)?,
}
Ok(())
}
fn run_merge<W: Write>(dir: &Path, w: &mut W) -> Result<(), MigrateError> {
match crate::migrate::make::make_merge_migration(dir)? {
Some(mig) => {
writeln!(w, "wrote {}", file_path(dir, &mig.name).display())?;
writeln!(
w,
" merge node — empty `forward`, anchors the chain after divergent leaves"
)?;
}
None => writeln!(
w,
"no merge needed — chain has at most one leaf (the history is already linear)"
)?,
}
Ok(())
}
async fn migrate<W: Write>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
let mut dry_run = false;
let mut squash = false;
let mut positional: Option<&str> = None;
for arg in args {
match arg.as_str() {
"--dry-run" => dry_run = true,
"--squash" => squash = true,
"--help" | "-h" => {
writeln!(
w,
"migrate apply pending migrations\n\
migrate <target> forward or back to <target> (`zero` wipes)\n\
migrate --dry-run preview the SQL without writing\n\
migrate --squash delete every pending (un-applied) migration JSON\n\
and regenerate a single fresh diff. Dev-iteration\n\
escape hatch — refuses to touch applied rows."
)?;
return Ok(());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
if positional.is_some() {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
positional = Some(other);
}
}
}
if squash {
if dry_run || positional.is_some() {
return Err(MigrateError::Validation(
"`migrate --squash` does not combine with `--dry-run` or a positional target"
.into(),
));
}
return migrate_squash(pool, dir, w).await;
}
if dry_run {
if positional.is_some() {
return Err(MigrateError::Validation(
"`migrate <target> --dry-run` is not supported — use plain `--dry-run` to preview pending forward migrations".into(),
));
}
let dirs = [dir.to_path_buf()];
return migrate_dry_run_cmd(pool, &dirs, &SystemView::tenant(dir), w).await;
}
if let Some(target) = positional {
if target != "zero" {
runner::target_on_disk(&super::file::list_dir(dir)?, dir, target)?;
}
if target != "zero" && !target_is_applied(pool, target).await? {
let unapplied = std::sync::Mutex::new(Vec::new());
let slot = &unapplied;
let mut out = Vec::new();
let run = migrate_with_framework(
pool,
dir,
&mut out,
runner::Signals::Fire { tenant: None },
|held| async move {
let moved =
runner::migrate_to_locked(held, pool, dir, target, runner::LEDGER_TABLE)
.await?;
*slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = moved.unapplied;
Ok(moved.applied)
},
)
.await;
let unapplied = unapplied
.into_inner()
.unwrap_or_else(std::sync::PoisonError::into_inner);
for m in unapplied {
writeln!(w, " unapplied {}", m.name)?;
}
w.write_all(&out)?;
return run;
}
let touched = runner::migrate_to_pool(pool, dir, target).await?;
if touched.is_empty() {
writeln!(w, "already at {target}")?;
} else {
let applied = runner::applied_set_pool(pool).await?;
for m in &touched {
let verb = if applied.contains(&m.name) {
"applied"
} else {
"unapplied"
};
writeln!(w, " {verb} {}", m.name)?;
}
}
return Ok(());
}
migrate_with_framework(
pool,
dir,
w,
runner::Signals::Fire { tenant: None },
|held| runner::migrate_pool_locked(held, pool, dir, None),
)
.await
}
async fn target_is_applied(pool: &Pool, target: &str) -> Result<bool, MigrateError> {
let applied = runner::with_migrate_lock_held(pool, |held| {
runner::ledger_names(held, pool, runner::LEDGER_TABLE)
})
.await?;
Ok(applied.contains(target))
}
pub(crate) struct SystemView {
dir: std::path::PathBuf,
scope: super::MigrationScope,
generates: &'static [crate::core::ModelScope],
}
impl SystemView {
pub(crate) fn tenant(dir: &Path) -> Self {
Self {
dir: super::make::SystemChain::dir_for(dir),
scope: super::MigrationScope::Tenant,
generates: &[
crate::core::ModelScope::Registry,
crate::core::ModelScope::Tenant,
],
}
}
pub(crate) fn registry(dir: &Path) -> Self {
Self {
dir: super::make::SystemChain::dir_for(dir),
scope: super::MigrationScope::Registry,
generates: &[crate::core::ModelScope::Registry],
}
}
fn planned(&self, files: Option<Vec<Migration>>) -> Result<Vec<Migration>, MigrateError> {
let mut all = files.unwrap_or_default();
for &scope in self.generates {
if let Some(step) = super::make::plan_system_step(scope, None, &all)? {
all.push(step);
}
}
Ok(all)
}
fn files(&self) -> Result<Option<Vec<Migration>>, MigrateError> {
if !self.dir.is_dir() {
return Ok(None);
}
let all = file::list_dir(&self.dir)?;
Ok((!all.is_empty()).then_some(all))
}
}
const NOT_GENERATED: &str = "the framework's system migrations are not on disk yet; \
`migrate` generates and applies them";
pub(crate) async fn migrate_dry_run_cmd<W: Write>(
pool: &Pool,
dirs: &[std::path::PathBuf],
system: &SystemView,
w: &mut W,
) -> Result<(), MigrateError> {
let mut previews = Vec::new();
let system_files = system.files()?;
let not_generated = system_files.is_none();
let all = system.planned(system_files)?;
let ledger = runner::SYSTEM_LEDGER_TABLE;
let owned = super::chains::project_owned(dirs)?;
let steps = runner::preview_pending(
pool,
&system.dir,
&all,
Some(system.scope),
ledger,
true,
&owned,
)
.await?;
previews.extend(steps.into_iter().map(|p| (p, "system, ")));
for dir in dirs {
let project = runner::migrate_dry_run_pool(pool, dir).await?;
previews.extend(project.into_iter().map(|p| (p, "")));
}
let mut after = Vec::new();
let converge = super::chains::converge_preview(pool, &all, system.scope, dirs).await?;
if !converge.is_empty() {
after.push(("converge framework tables".to_owned(), converge));
}
let tenant = system.scope == super::MigrationScope::Tenant;
let bootstrap = shared_bootstrap()
.into_iter()
.chain(tenant.then(tenant_bootstrap).into_iter().flatten());
let created = super::chains::pending_creates(pool, &all, system.scope, dirs).await?;
for b in bootstrap {
if created.contains(b.table) || runner::try_table_exists_here(pool, b.table).await? {
continue;
}
let ddl = (b.ddl)(pool.dialect()).map_err(MigrateError::Validation)?;
let ddl = ddl
.into_iter()
.map(super::chains::PreviewLine::Sql)
.collect();
after.push((format!("bootstrap {}", b.table), ddl));
}
if not_generated {
writeln!(w, "-- {NOT_GENERATED}")?;
}
if previews.is_empty() && after.is_empty() {
writeln!(w, "nothing to migrate (already up to date)")?;
return Ok(());
}
writeln!(
w,
"-- DRY RUN: {} pending migration(s); no SQL will be executed",
previews.len()
)?;
for (p, chain) in &previews {
writeln!(w)?;
let atomic = if p.atomic { "atomic" } else { "non-atomic" };
writeln!(w, "-- {} ({chain}{atomic})", p.name)?;
for stmt in &p.statements {
writeln!(w, "{stmt};")?;
}
}
for (label, lines) in &after {
writeln!(w)?;
writeln!(w, "-- {label} (idempotent)")?;
for line in lines {
match line {
super::chains::PreviewLine::Sql(stmt) => writeln!(w, "{stmt};")?,
super::chains::PreviewLine::Note(note) => writeln!(w, "-- {note}")?,
}
}
}
Ok(())
}
pub(crate) async fn migrate_with_framework<W, F, Fut>(
pool: &Pool,
dir: &Path,
w: &mut W,
signals: super::Signals<'_>,
project: F,
) -> Result<(), MigrateError>
where
W: Write,
F: FnOnce(super::LockHeld) -> Fut,
Fut: std::future::Future<Output = Result<Vec<Migration>, MigrateError>>,
{
let chain = system_chain(dir)?;
let applied = super::chains::migrate_chains(
pool,
&chain,
super::MigrationScope::Tenant,
&[dir.to_path_buf()],
None,
signals,
project,
)
.await?;
for m in &applied.system {
writeln!(w, " applied {} (system)", m.name)?;
}
for m in &applied.project {
writeln!(w, " applied {}", m.name)?;
}
if applied.system.is_empty() && applied.project.is_empty() {
writeln!(w, "nothing to migrate (already up to date)")?;
}
ensure_bootstrap_tables(pool).await?;
ensure_tenant_bootstrap_tables(pool).await?;
#[cfg(feature = "sqlite")]
{
let fixed = super::sqlite_datetime::normalise_sqlite_datetimes(pool).await?;
if !fixed.is_clean() {
writeln!(
w,
" normalised {} SQLite datetime value(s) in {} — these were written by \
the pre-0.57.11 default and did not compare against a bound timestamp",
fixed.rows,
fixed.columns.join(", "),
)?;
}
}
Ok(())
}
pub(crate) async fn ensure_bootstrap_tables(pool: &Pool) -> Result<(), sqlx::Error> {
for b in shared_bootstrap() {
(b.ensure)(pool).await?;
}
Ok(())
}
pub(crate) async fn ensure_tenant_bootstrap_tables(pool: &Pool) -> Result<(), sqlx::Error> {
for b in tenant_bootstrap() {
(b.ensure)(pool).await?;
}
Ok(())
}
type EnsureFuture<'a> =
std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), sqlx::Error>> + Send + 'a>>;
struct Bootstrap {
table: &'static str,
ensure: for<'a> fn(&'a Pool) -> EnsureFuture<'a>,
ddl: fn(&dyn crate::sql::Dialect) -> Result<Vec<String>, String>,
}
fn shared_bootstrap() -> Vec<Bootstrap> {
use crate::core::Model as _;
fn audit(pool: &Pool) -> EnsureFuture<'_> {
Box::pin(crate::audit::ensure_table_pool(pool))
}
fn i18n(pool: &Pool) -> EnsureFuture<'_> {
Box::pin(crate::i18n::db::ensure_table_pool(pool))
}
#[cfg(all(feature = "admin", feature = "totp"))]
fn totp(pool: &Pool) -> EnsureFuture<'_> {
Box::pin(crate::admin::totp_store::ensure_table(pool))
}
#[cfg(all(feature = "tenancy", feature = "totp"))]
fn operator_totp(pool: &Pool) -> EnsureFuture<'_> {
Box::pin(crate::tenancy::operator_totp::ensure_table(pool))
}
vec![
Bootstrap {
table: crate::audit::AuditLog::SCHEMA.table,
ensure: audit,
ddl: |d| super::ensure::idempotent_ddl(&crate::audit::bootstrap_snapshot(), d),
},
#[cfg(all(feature = "admin", feature = "totp"))]
Bootstrap {
table: crate::admin::totp_store::AdminTotp::SCHEMA.table,
ensure: totp,
ddl: |d| {
super::ensure::idempotent_ddl(&crate::admin::totp_store::bootstrap_snapshot(), d)
},
},
#[cfg(all(feature = "tenancy", feature = "totp"))]
Bootstrap {
table: crate::tenancy::OperatorTotp::SCHEMA.table,
ensure: operator_totp,
ddl: |d| {
super::ensure::idempotent_ddl(
&crate::tenancy::operator_totp::bootstrap_snapshot(),
d,
)
},
},
Bootstrap {
table: crate::i18n::db::Translation::SCHEMA.table,
ensure: i18n,
ddl: |d| {
Ok(crate::i18n::db::ddl(d)
.split(';')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_owned)
.collect())
},
},
]
}
fn tenant_bootstrap() -> Vec<Bootstrap> {
#[cfg(feature = "passkey")]
fn passkey(pool: &Pool) -> EnsureFuture<'_> {
Box::pin(crate::passkey::ensure_table(pool))
}
vec![
#[cfg(feature = "passkey")]
Bootstrap {
table: {
use crate::core::Model as _;
crate::passkey::WebauthnCredential::SCHEMA.table
},
ensure: passkey,
ddl: |d| super::ensure::idempotent_ddl(&crate::passkey::bootstrap_snapshot(), d),
},
]
}
async fn migrate_squash<W: Write>(pool: &Pool, dir: &Path, w: &mut W) -> Result<(), MigrateError> {
runner::ensure_ledger_pool(pool).await?;
let all = file::list_dir(dir)?;
let applied = runner::applied_set_pool(pool).await?;
let pending: Vec<&Migration> = all.iter().filter(|m| !applied.contains(&m.name)).collect();
if pending.is_empty() {
writeln!(
w,
"no pending migrations to squash (every JSON is in the ledger)"
)?;
return Ok(());
}
if pending.len() == 1 {
return Err(MigrateError::Validation(format!(
"only one pending migration (`{}`) — use `forget-pending {}` instead of `--squash`",
pending[0].name, pending[0].name,
)));
}
for m in &pending {
if applied.contains(&m.name) {
return Err(MigrateError::Validation(format!(
"migrate --squash: refused — migration `{}` appears applied. \
Use `migrate <prev>` or `downgrade` to unapply it first.",
m.name,
)));
}
}
for m in &pending {
if let Some((i, kind)) = m.forward.iter().enumerate().find_map(|(i, op)| {
let kind = match op {
Operation::Data(_) => "data op",
Operation::Callback(_) => "callback",
Operation::Schema(super::SchemaChange::RenameTable { .. }) => "RenameTable",
Operation::Schema(super::SchemaChange::RenameColumn { .. }) => "RenameColumn",
Operation::Schema(_) => return None,
};
Some((i, kind))
}) {
return Err(MigrateError::Validation(format!(
"migrate --squash: refused — `{}` forward[{i}] is a {kind}, which a fresh diff \
would lose. Fold the pending files by hand instead.",
m.name,
)));
}
}
if super::make::models_ahead_of(&all, has_registry_models())? {
return Err(MigrateError::Validation(
"migrate --squash: refused — the models have changes no migration holds yet. \
Run `makemigrations` first, then squash."
.into(),
));
}
writeln!(
w,
"squashing {} pending migration(s) into a fresh diff:",
pending.len()
)?;
let removed: Vec<(String, super::MigrationScope)> =
pending.iter().map(|m| (m.name.clone(), m.scope)).collect();
for m in &pending {
let path = file_path(dir, &m.name);
std::fs::remove_file(&path).map_err(|e| {
MigrateError::Validation(format!("migrate --squash: rm {}: {e}", path.display()))
})?;
writeln!(w, " removed {}", path.display())?;
}
writeln!(w, "regenerating diff against current model registry...")?;
let before: std::collections::HashSet<String> =
file::list_dir(dir)?.into_iter().map(|m| m.name).collect();
makemigrations(dir, &["--drop-and-add".to_owned()], w)?;
stamp_replaces(dir, &before, &removed, w)
}
fn system_chain(dir: &Path) -> Result<super::make::SystemChain, MigrateError> {
super::make::SystemChain::for_migrations_dir(
dir,
&[
crate::core::ModelScope::Registry,
crate::core::ModelScope::Tenant,
],
)
}
fn stamp_replaces<W: Write>(
dir: &Path,
before: &std::collections::HashSet<String>,
removed: &[(String, super::MigrationScope)],
w: &mut W,
) -> Result<(), MigrateError> {
for mut mig in file::list_dir(dir)? {
if before.contains(&mig.name) {
continue; }
let claims: Vec<String> = removed
.iter()
.filter(|(_, scope)| *scope == mig.scope)
.map(|(name, _)| name.clone())
.collect();
if claims.is_empty() {
continue;
}
writeln!(w, " {} now replaces: {}", mig.name, claims.join(", "))?;
mig.replaces = claims;
file::write(&file_path(dir, &mig.name), &mig)?;
}
Ok(())
}
async fn downgrade<W: Write>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
let steps: usize = if let Some(arg) = args.first() {
arg.parse().map_err(|_| {
MigrateError::Validation(format!(
"invalid step count: {arg} (expected a non-negative integer)"
))
})?
} else {
1
};
let touched = runner::downgrade_pool(pool, dir, steps).await?;
if touched.is_empty() {
writeln!(w, "nothing to downgrade")?;
} else {
for m in &touched {
writeln!(w, " rolled back {}", m.name)?;
}
}
Ok(())
}
fn sqlmigrate_cmd<W: Write>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
let mut positional: Option<&str> = None;
for arg in args {
match arg.as_str() {
"--help" | "-h" => {
writeln!(
w,
"sqlmigrate <name> print the SQL the named migration would emit\n\
without applying it"
)?;
return Ok(());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
if positional.is_some() {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
positional = Some(other);
}
}
}
let name = positional
.ok_or_else(|| MigrateError::Validation("sqlmigrate requires a migration name".into()))?;
let system = super::make::SystemChain::dir_for(dir);
let in_system = !file::list_dir(dir)?.iter().any(|m| m.name == name)
&& system.is_dir()
&& file::list_dir(&system)?.iter().any(|m| m.name == name);
let src = if in_system { system.as_path() } else { dir };
let preview = runner::sqlmigrate_one(src, name, pool.dialect())?;
writeln!(
w,
"-- {} ({})",
preview.name,
if preview.atomic {
"atomic"
} else {
"non-atomic"
}
)?;
for stmt in &preview.statements {
writeln!(w, "{stmt};")?;
}
Ok(())
}
async fn showmigrations<W: Write>(pool: &Pool, dir: &Path, w: &mut W) -> Result<(), MigrateError> {
showmigrations_with(pool, dir, &SystemView::tenant(dir), None, w).await
}
pub(crate) type TenantLedgers = [(String, std::collections::HashSet<String>)];
pub(crate) struct TenantChains<'a> {
pub(crate) project: &'a TenantLedgers,
pub(crate) system: &'a TenantLedgers,
}
pub(crate) async fn showmigrations_with<W: Write>(
pool: &Pool,
dir: &Path,
system: &SystemView,
tenants: Option<&TenantChains<'_>>,
w: &mut W,
) -> Result<(), MigrateError> {
use super::MigrationScope;
let all = file::list_dir(dir)?;
let applied = runner::applied_or_empty(pool, runner::LEDGER_TABLE).await?;
if all.is_empty() {
writeln!(w, "(no migrations in {})", dir.display())?;
} else {
writeln!(w, "Migrations in {}:", dir.display())?;
}
chain_lines(w, &all, &applied, tenants.map(|t| t.project), |_| true)?;
let Some(files) = system.files()? else {
writeln!(w, "System migrations: {NOT_GENERATED}")?;
return Ok(());
};
let applied = runner::applied_or_empty(pool, runner::SYSTEM_LEDGER_TABLE).await?;
writeln!(w, "System migrations in {}:", system.dir.display())?;
let tenant_steps = tenants.is_some();
chain_lines(w, &files, &applied, tenants.map(|t| t.system), |m| {
m.scope == system.scope || (tenant_steps && m.scope == MigrationScope::Tenant)
})
}
fn chain_lines<W: Write>(
w: &mut W,
all: &[Migration],
applied: &std::collections::HashSet<String>,
tenants: Option<&TenantLedgers>,
shown: impl Fn(&Migration) -> bool,
) -> Result<(), MigrateError> {
use runner::MigrationState as S;
let per_tenant: Vec<Vec<S>> = tenants
.unwrap_or_default()
.iter()
.map(|(_, done)| {
runner::migration_states(all, done)
.map(|(_, s)| s)
.collect()
})
.collect();
for (i, (m, state)) in runner::migration_states(all, applied).enumerate() {
if !shown(m) {
continue;
}
if tenants.is_some() && m.scope == super::MigrationScope::Tenant {
let n = per_tenant.len();
if n == 0 {
writeln!(w, " [-] {} (no tenants)", m.name)?;
continue;
}
let done = per_tenant.iter().filter(|t| t[i] != S::Pending).count();
let mark = match done {
0 => "[ ]",
d if d == n => "[X]",
_ => "[~]",
};
writeln!(w, " {mark} {} ({done}/{n} tenants)", m.name)?;
} else {
writeln!(w, " {}", state_line(m, state))?;
}
}
Ok(())
}
fn state_line(m: &Migration, state: runner::MigrationState) -> String {
match state {
runner::MigrationState::Applied => format!("[X] {}", m.name),
runner::MigrationState::Replaced => format!("[X] {} (replaced by a squash)", m.name),
runner::MigrationState::Pending => format!("[ ] {}", m.name),
}
}
async fn forget_pending_cmd<W: Write>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
forget_pending_cmd_with(pool, dir, args, w, async { Ok(Vec::new()) }).await
}
pub(crate) async fn forget_pending_cmd_with<W: Write>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
tenant_ledgers: impl std::future::Future<
Output = Result<Vec<(String, std::collections::HashSet<String>)>, MigrateError>,
>,
) -> Result<(), MigrateError> {
let target = match args.first().map(String::as_str) {
Some("--help") | Some("-h") | None => {
writeln!(w, "forget-pending <name>")?;
writeln!(
w,
" Delete a migration JSON that has NOT been applied yet, so the"
)?;
writeln!(
w,
" next `makemigrations` regenerates the diff against current"
)?;
writeln!(
w,
" models. Refuses to delete an already-applied migration (would"
)?;
writeln!(
w,
" orphan the `{}` ledger row and break later runs).",
runner::LEDGER_TABLE
)?;
writeln!(w)?;
writeln!(
w,
" <name> can be the full migration name (e.g. `0003_auto`)"
)?;
writeln!(w, " or a unique substring; ambiguous matches error.")?;
return Ok(());
}
Some(s) if s.starts_with('-') => {
return Err(MigrateError::Validation(format!(
"forget-pending: expected positional <name> first, got flag `{s}`"
)));
}
Some(s) => s,
};
runner::ensure_ledger_pool(pool).await?;
let all = file::list_dir(dir)?;
let applied = runner::applied_set_pool(pool).await?;
let exact: Vec<&Migration> = all.iter().filter(|m| m.name == target).collect();
let candidates: Vec<&Migration> = if exact.len() == 1 {
exact
} else {
all.iter().filter(|m| m.name.contains(target)).collect()
};
let migration = match candidates.len() {
0 => {
return Err(MigrateError::Validation(format!(
"forget-pending: no migration matches `{target}` in {}",
dir.display()
)));
}
1 => candidates[0],
n => {
let names: Vec<&str> = candidates.iter().map(|m| m.name.as_str()).collect();
return Err(MigrateError::Validation(format!(
"forget-pending: `{target}` is ambiguous ({n} matches: {}); pass the full migration name",
names.join(", ")
)));
}
};
if applied.contains(&migration.name) {
return Err(MigrateError::Validation(format!(
"forget-pending: migration `{}` is already applied (recorded in `{}`). \
Use `migrate <prev>` or `downgrade` to unapply it first, then \
`forget-pending` to drop the JSON.",
migration.name,
runner::LEDGER_TABLE,
)));
}
if let Some((slug, _)) = tenant_ledgers
.await?
.into_iter()
.find(|(_, applied)| applied.contains(&migration.name))
{
return Err(MigrateError::Validation(format!(
"forget-pending: migration `{}` is already applied on tenant `{slug}`. \
Unapply it there first.",
migration.name,
)));
}
let dependants: Vec<&str> = all
.iter()
.filter(|m| {
m.prev.as_deref() == Some(migration.name.as_str()) || m.merges.contains(&migration.name)
})
.map(|m| m.name.as_str())
.collect();
if !dependants.is_empty() {
return Err(MigrateError::Validation(format!(
"forget-pending: `{}` is built on by {}; forget those first",
migration.name,
dependants.join(", "),
)));
}
let path = file_path(dir, &migration.name);
std::fs::remove_file(&path).map_err(|e| {
MigrateError::Validation(format!("forget-pending: rm {}: {e}", path.display()))
})?;
writeln!(w, "deleted {}", path.display())?;
writeln!(w, " next: `cargo run -- makemigrations` will regenerate")?;
writeln!(w, " the diff against the current model registry.")?;
Ok(())
}
pub fn make_empty(dir: &Path, name: &str) -> Result<Migration, MigrateError> {
make_empty_in_scope(dir, name, super::MigrationScope::default())
}
pub fn make_empty_in_scope(
dir: &Path,
name: &str,
scope: super::MigrationScope,
) -> Result<Migration, MigrateError> {
write_scaffold(dir, name, scope, Vec::new())
}
fn write_scaffold(
dir: &Path,
name: &str,
scope: super::MigrationScope,
forward: Vec<Operation>,
) -> Result<Migration, MigrateError> {
let prior = file::list_dir(dir)?;
let chain: Vec<&Migration> = prior.iter().filter(|m| m.scope == scope).collect();
let head = chain.last().copied();
let prev_snapshot = head.map_or_else(SchemaSnapshot::default, |m| {
super::make::effective_snapshot(&chain, m)
});
let prev_name = head.map(|m| m.name.clone());
let next_index = prior
.last()
.and_then(|m| file::extract_index(&m.name))
.map_or(1, |n| n + 1);
let full_name = format!("{next_index:04}_{name}");
let mig = Migration {
name: full_name.clone(),
created_at: chrono::Utc::now().to_rfc3339(),
prev: prev_name,
atomic: true,
scope,
replaces: Vec::new(),
merges: Vec::new(),
snapshot: prev_snapshot,
forward,
};
if !dir.exists() {
std::fs::create_dir_all(dir)?;
}
file::write(&file_path(dir, &mig.name), &mig)?;
Ok(mig)
}
fn file_path(dir: &Path, name: &str) -> std::path::PathBuf {
dir.join(format!("{name}.json"))
}
pub fn make_data_migration(
dir: &Path,
name: &str,
sql: &str,
reverse_sql: Option<&str>,
) -> Result<Migration, MigrateError> {
make_data_migration_in_scope(
dir,
name,
sql,
reverse_sql,
super::MigrationScope::default(),
)
}
pub fn make_data_migration_in_scope(
dir: &Path,
name: &str,
sql: &str,
reverse_sql: Option<&str>,
scope: super::MigrationScope,
) -> Result<Migration, MigrateError> {
let op = Operation::Data(DataOp {
sql: sql.to_owned(),
reverse_sql: reverse_sql.map(str::to_owned),
reversible: reverse_sql.is_some(),
});
write_scaffold(dir, name, scope, vec![op])
}
pub fn append_data_op(
dir: &Path,
migration_name: &str,
sql: &str,
reverse_sql: Option<&str>,
) -> Result<(), MigrateError> {
let path = file_path(dir, migration_name);
let mut mig = file::load(&path).map_err(|e| match e {
MigrateError::Io(io) if io.kind() == std::io::ErrorKind::NotFound => {
MigrateError::Validation(format!(
"migration `{migration_name}` not found at {}",
path.display()
))
}
other => other,
})?;
mig.forward.push(Operation::Data(DataOp {
sql: sql.to_owned(),
reverse_sql: reverse_sql.map(str::to_owned),
reversible: reverse_sql.is_some(),
}));
file::write(&path, &mig)?;
Ok(())
}
fn add_data_op_cmd<W: Write>(dir: &Path, args: &[String], w: &mut W) -> Result<(), MigrateError> {
let mut sql: Option<String> = None;
let mut reverse_sql: Option<String> = None;
let mut name: Option<String> = None;
let mut to: Option<String> = None;
let mut scope: Option<super::MigrationScope> = None;
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--scope" => {
let raw = iter.next().ok_or_else(|| {
MigrateError::Validation("--scope requires \"registry\" or \"tenant\"".into())
})?;
scope = match crate::core::ModelScope::from_str(raw) {
Some(crate::core::ModelScope::Registry) => {
Some(super::MigrationScope::Registry)
}
Some(crate::core::ModelScope::Tenant) => Some(super::MigrationScope::Tenant),
None => {
return Err(MigrateError::Validation(format!(
"--scope must be \"registry\" or \"tenant\", got {raw:?}"
)))
}
};
}
"--sql" => {
sql =
Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--sql requires a value".into())
})?);
}
"--reverse-sql" => {
reverse_sql = Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--reverse-sql requires a value".into())
})?);
}
"--name" => {
name =
Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--name requires a value".into())
})?);
}
"--to" => {
to = Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--to requires a migration name".into())
})?);
}
"--help" | "-h" => {
writeln!(
w,
"add-data-op --sql <SQL> [--reverse-sql <SQL>] [--name <name>] [--to <migration> | --scope <s>]"
)?;
return Ok(());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected argument: `{other}` — use --sql, --reverse-sql, --name, --to, --scope"
)));
}
}
}
let sql = sql.ok_or_else(|| MigrateError::Validation("--sql is required".into()))?;
if let Some(migration_name) = to {
if scope.is_some() {
return Err(MigrateError::Validation(
"--to appends to that file; it does not combine with --scope".into(),
));
}
append_data_op(dir, &migration_name, &sql, reverse_sql.as_deref())?;
writeln!(w, "appended data op to {migration_name}.json")?;
} else {
let name = name.unwrap_or_else(|| "data_op".to_owned());
let mig = make_data_migration_in_scope(
dir,
&name,
&sql,
reverse_sql.as_deref(),
scope.unwrap_or_default(),
)?;
let rev_note = if reverse_sql.is_some() {
" (reversible)"
} else {
" (irreversible)"
};
writeln!(w, "wrote {}{rev_note}", file_path(dir, &mig.name).display())?;
}
Ok(())
}
fn describe_op(op: &Operation) -> String {
match op {
Operation::Schema(c) => format!("{c:?}"),
Operation::Data(d) => {
let head: String = d.sql.chars().take(60).collect();
let ellipsis = if d.sql.chars().count() > 60 {
"…"
} else {
""
};
format!("data: {head}{ellipsis}")
}
Operation::Callback(c) => format!("callback: {}", c.name),
}
}
fn startapp<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let mut iter = args.iter();
let app_name = iter
.next()
.cloned()
.ok_or_else(|| MigrateError::Validation(usage()))?;
let mut with_manage_bin = false;
let mut into: Option<String> = None;
while let Some(arg) = iter.next() {
match arg.as_str() {
"--with-manage-bin" => with_manage_bin = true,
"--into" => {
into = Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--into requires a directory argument".into())
})?);
}
"--help" | "-h" => {
writeln!(w, "{}", usage())?;
return Ok(());
}
other => {
return Err(MigrateError::Validation(format!(
"startapp: unknown argument `{other}` (run --help for usage)"
)));
}
}
}
let opts = super::scaffold::StartAppOptions {
app_name: app_name.clone(),
manage_bin: with_manage_bin.then_some(super::scaffold::SINGLE_TENANT_MANAGE_BIN),
base_dir: into.map(std::path::PathBuf::from),
crate_root: None,
};
let cwd = std::env::current_dir()?;
let report = super::scaffold::startapp(&cwd, &opts)?;
write_startapp_report(w, &app_name, &report)
}
fn write_startapp_report<W: Write>(
w: &mut W,
app_name: &str,
report: &super::scaffold::StartAppReport,
) -> Result<(), MigrateError> {
if report.written.is_empty() && report.skipped.is_empty() {
writeln!(w, "startapp: nothing to do")?;
return Ok(());
}
writeln!(w, "startapp `{app_name}`")?;
for path in &report.written {
writeln!(w, " + wrote {path}")?;
}
for path in &report.skipped {
writeln!(w, " · {path} already exists — left untouched")?;
}
for path in &report.patched {
writeln!(w, " ~ patched {path} (auto-mounted new app)")?;
}
for hint in &report.manual_steps {
writeln!(w, " ! manual: {hint}")?;
}
if !report.written.is_empty() {
writeln!(w, "next:")?;
writeln!(w, " {}", super::scaffold::STARTAPP_NEXT)?;
}
Ok(())
}
fn usage() -> String {
"startapp <name> [--with-manage-bin]\n \
Scaffold an app module under src/<name>/ (mod.rs +\n \
models.rs + views.rs + urls.rs). Idempotent: existing files\n \
are left untouched. <name> must be a valid Rust identifier.\n\n \
--with-manage-bin\n \
Also write src/bin/manage.rs with the single-tenant dispatcher\n \
boilerplate. Skipped if the file already exists."
.to_owned()
}
fn about_db_url(raw: Option<String>) -> String {
raw.map_or_else(|| "(unset)".into(), |s| redact_url(&s))
}
async fn about_cmd<W: Write>(pool: &Pool, w: &mut W) -> Result<(), MigrateError> {
let registered_models = crate::core::inventory::iter::<crate::core::ModelEntry>
.into_iter()
.count();
let mut apps: std::collections::BTreeSet<&'static str> = std::collections::BTreeSet::new();
for entry in crate::core::inventory::iter::<crate::core::ModelEntry> {
if let Some(app) = entry.resolved_app_label() {
apps.insert(app);
}
}
writeln!(w, "rustango")?;
writeln!(w, " version: {}", env!("CARGO_PKG_VERSION"))?;
writeln!(w, " models: {registered_models} registered")?;
writeln!(
w,
" apps: {} ({})",
apps.len(),
if apps.is_empty() {
"none".to_owned()
} else {
apps.iter().copied().collect::<Vec<_>>().join(", ")
}
)?;
let env_label = std::env::var("RUSTANGO_ENV").unwrap_or_else(|_| "(unset)".into());
writeln!(w, " RUSTANGO_ENV: {env_label}")?;
let db_url = about_db_url(crate::env::database_url_from_env().ok());
writeln!(w, " DATABASE_URL: {db_url}")?;
write!(w, " db_connect: ")?;
let ok = crate::sql::raw_execute_pool(pool, "SELECT 1", Vec::new())
.await
.is_ok();
writeln!(w, "{}", if ok { "ok" } else { "FAILED" })?;
Ok(())
}
async fn collation_audit(pool: &Pool, audit: &mut DeployAuditFindings) {
let Some(sql) = pool.dialect().default_collation_sql() else {
return;
};
match crate::sql::raw_query_pool::<(String,)>(sql, Vec::new(), pool).await {
Ok(rows) => {
if let Some(w) = rows.first().and_then(|(c,)| collation_warning(c)) {
audit.warnings.push(w);
}
}
Err(e) => audit
.warnings
.push(format!("could not read the database collation: {e}")),
}
}
async fn time_zone_audit(pool: &Pool, audit: &mut DeployAuditFindings) {
let Some(sql) = pool.dialect().named_time_zone_sql() else {
return;
};
match crate::sql::raw_query_pool::<(Option<String>,)>(sql, Vec::new(), pool).await {
Ok(rows) => {
let converted = rows.first().and_then(|(c,)| c.as_deref());
audit.warnings.extend(time_zone_warning(converted));
}
Err(e) => audit
.warnings
.push(format!("could not check named time zones: {e}")),
}
}
fn time_zone_warning(converted: Option<&str>) -> Option<String> {
converted.is_none().then(|| {
"the database has no time zone tables: `trunc_with_tz` with a named zone returns \
NULL — load them with `mysql_tzinfo_to_sql`, or pass `±HH:MM` offsets"
.to_owned()
})
}
fn collation_warning(collation: &str) -> Option<String> {
const FIX: &str = "create the database with `COLLATE utf8mb4_0900_as_cs` \
(see the MySQL section of docs/getting-started.md)";
if collation.ends_with("_ci") {
Some(format!(
"database default collation `{collation}` ignores case: `=` and `unique` on \
text columns differ from PostgreSQL/SQLite — {FIX}"
))
} else if collation.ends_with("_bin") {
Some(format!(
"database default collation `{collation}` makes text columns unreadable as \
`String` — {FIX}"
))
} else {
None
}
}
#[cfg(feature = "admin-sso")]
async fn sso_link_audit(pool: &Pool, audit: &mut DeployAuditFindings) {
match crate::sso::check::admin_providers(pool).await {
Ok(slugs) => audit
.warnings
.extend(slugs.iter().map(|s| crate::sso::check::admin_warning(s))),
Err(e) => audit.warnings.push(format!(
"[sso] could not check the admin SSO providers: {e}"
)),
}
}
pub(crate) type TenantApplied = Vec<(String, Result<TenantNames, String>)>;
pub(crate) struct TenantNames {
pub(crate) project: std::collections::HashSet<String>,
pub(crate) system: std::collections::HashSet<String>,
}
pub(crate) struct TenancyCheck<D, L> {
pub(crate) deploy: D,
pub(crate) ledgers: L,
}
const TENANTS_NAMED: usize = 10;
fn check_dirs(dir: &Path, apps: bool) -> Vec<(String, std::path::PathBuf)> {
if !apps {
return if dir.exists() {
vec![(String::new(), dir.to_path_buf())]
} else {
Vec::new()
};
}
let root = file::project_root_of(dir);
file::project_migration_dirs(dir)
.into_iter()
.filter(|d| d.exists())
.map(|d| {
let app = d
.parent()
.filter(|p| *p != root && d != dir)
.and_then(|p| p.file_name())
.map(|n| format!("{}/", n.to_string_lossy()))
.unwrap_or_default();
(app, d)
})
.collect()
}
fn pending_names(
files: &[(String, Vec<super::Migration>)],
applied: &std::collections::HashSet<String>,
scope: Option<super::MigrationScope>,
) -> Vec<String> {
files
.iter()
.flat_map(|(app, migs)| {
runner::migration_states(migs, applied)
.filter(|(m, s)| {
*s == runner::MigrationState::Pending && scope.is_none_or(|sc| m.scope == sc)
})
.map(move |(m, _)| format!("{app}{}", m.name))
.collect::<Vec<_>>()
})
.collect()
}
enum Ledgers<L> {
Single,
Tenancy { tenants: Option<L> },
}
async fn pending_migration_audit<L>(
pool: &Pool,
dir: &Path,
files: &[(String, Vec<super::Migration>)],
(system, system_scope): (&[(String, Vec<super::Migration>)], super::MigrationScope),
ledgers: Ledgers<L>,
findings: (&mut Vec<String>, &mut Vec<String>, &mut Vec<String>),
) where
L: std::future::Future<Output = Result<TenantApplied, String>>,
{
let (info, warnings, errors) = findings;
match super::make::planned_migrations(dir, has_registry_models()) {
Ok(planned) if !planned.is_empty() => {
let ops: Vec<String> = planned
.iter()
.flat_map(|m| m.forward.iter().map(describe_op))
.collect();
warnings.push(format!(
"models have changes no migration holds — run `manage makemigrations`:\n {}",
ops.join("\n ")
));
}
Ok(_) => {}
Err(e) => errors.push(format!("makemigrations would fail: {e}")),
}
let (scope, tenants) = match ledgers {
Ledgers::Single => (None, None),
Ledgers::Tenancy { tenants } => (Some(super::MigrationScope::Registry), Some(tenants)),
};
match runner::applied_or_empty(pool, runner::LEDGER_TABLE).await {
Ok(applied) => {
let pending = pending_names(files, &applied, scope);
if !pending.is_empty() {
warnings.push(format!(
"{} migration(s) not applied — run `manage migrate`: {}",
pending.len(),
pending.join(", ")
));
}
}
Err(e) => errors.push(format!("cannot read the migration ledger: {e}")),
}
if !system.is_empty() {
match runner::applied_or_empty(pool, runner::SYSTEM_LEDGER_TABLE).await {
Ok(applied) => {
let pending = pending_names(system, &applied, Some(system_scope));
if !pending.is_empty() {
warnings.push(format!(
"{} system migration(s) not applied — run `manage migrate`: {}",
pending.len(),
pending.join(", ")
));
}
}
Err(e) => errors.push(format!("cannot read the system migration ledger: {e}")),
}
}
let tenant_files = files
.iter()
.chain(system)
.flat_map(|(_, m)| m)
.any(|m| m.scope == super::MigrationScope::Tenant);
let Some(tenants) = tenants.filter(|_| tenant_files) else {
return;
};
let Some(ledgers) = tenants else {
info.push("tenant ledgers not read (--no-tenants)".into());
return;
};
match ledgers.await {
Ok(tenants) => tenant_pending_audit(files, system, &tenants, warnings),
Err(e) => errors.push(format!("cannot list the tenants: {e}")),
}
}
fn tenant_pending_audit(
files: &[(String, Vec<super::Migration>)],
system: &[(String, Vec<super::Migration>)],
tenants: &TenantApplied,
warnings: &mut Vec<String>,
) {
let tenant = Some(super::MigrationScope::Tenant);
let mut behind = Vec::new();
let mut unreadable = Vec::new();
for (slug, applied) in tenants {
match applied {
Ok(applied) => {
let mut pending = pending_names(system, &applied.system, tenant);
pending.extend(pending_names(files, &applied.project, tenant));
if !pending.is_empty() {
behind.push(format!("{slug}: {}", pending.join(", ")));
}
}
Err(e) => unreadable.push(format!("{slug}: {e}")),
}
}
let list = |lines: &[String]| {
let mut shown: Vec<&str> = lines
.iter()
.take(TENANTS_NAMED)
.map(String::as_str)
.collect();
let more = lines.len().saturating_sub(TENANTS_NAMED);
let tail = format!("… and {more} more");
if more > 0 {
shown.push(&tail);
}
shown.join("\n ")
};
if !behind.is_empty() {
warnings.push(format!(
"{} tenant(s) behind on tenant migrations — run `manage migrate-tenants`:\n {}",
behind.len(),
list(&behind)
));
}
if !unreadable.is_empty() {
warnings.push(format!(
"could not read {} tenant ledger(s):\n {}",
unreadable.len(),
list(&unreadable)
));
}
}
async fn check_cmd<W: Write>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
type Never<T> = std::future::Ready<T>;
check_cmd_with(
pool,
dir,
args,
w,
None::<TenancyCheck<Never<DeployAuditFindings>, Never<Result<TenantApplied, String>>>>,
)
.await
}
pub(crate) async fn check_cmd_with<W: Write, D, L>(
pool: &Pool,
dir: &Path,
args: &[String],
w: &mut W,
tenancy: Option<TenancyCheck<D, L>>,
) -> Result<(), MigrateError>
where
D: std::future::Future<Output = DeployAuditFindings>,
L: std::future::Future<Output = Result<TenantApplied, String>>,
{
let flags: &[&str] = if tenancy.is_some() {
&["--deploy", "--no-tenants"]
} else {
&["--deploy"]
};
if let Some(other) = args.iter().find(|a| !flags.contains(&a.as_str())) {
let usage = flags.iter().map(|f| format!(" [{f}]")).collect::<String>();
return Err(MigrateError::Validation(format!(
"check: unknown argument `{other}` (usage: check{usage})"
)));
}
let read_tenants = !args.iter().any(|a| a == "--no-tenants");
let (deploy_findings, ledgers) = match tenancy {
Some(t) => (
Some(t.deploy),
Ledgers::Tenancy {
tenants: read_tenants.then_some(t.ledgers),
},
),
None => (None, Ledgers::Single),
};
let deploy = args.iter().any(|a| a == "--deploy");
let mut errors: Vec<String> = Vec::new();
let mut warnings: Vec<String> = Vec::new();
let mut info: Vec<String> = Vec::new();
writeln!(
w,
"running rustango system check{}...",
if deploy { " (deploy mode)" } else { "" }
)?;
let model_count = crate::core::inventory::iter::<crate::core::ModelEntry>
.into_iter()
.count();
if model_count == 0 {
errors.push("no models registered — every #[derive(Model)] struct must be `pub use`d through the binary's crate root".into());
} else {
info.push(format!("{model_count} models registered via inventory"));
}
if crate::sql::raw_execute_pool(pool, "SELECT 1", Vec::new())
.await
.is_err()
{
errors.push("cannot connect to database — verify DATABASE_URL is reachable".into());
} else {
info.push("database reachable".into());
}
let dirs = check_dirs(dir, matches!(ledgers, Ledgers::Tenancy { .. }));
if !dirs.is_empty() {
let mut files = Vec::with_capacity(dirs.len());
for (app, d) in dirs {
files.push((app, file::list_dir(&d)?));
}
let view = match &ledgers {
Ledgers::Tenancy { .. } => SystemView::registry(dir),
Ledgers::Single => SystemView::tenant(dir),
};
let system = view
.files()?
.map(|m| vec![("system/".to_owned(), m)])
.unwrap_or_default();
let count: usize = files.iter().map(|(_, m)| m.len()).sum();
if count == 0 && model_count > 0 {
warnings.push(
"models registered but no migrations on disk — run `manage makemigrations`".into(),
);
} else {
info.push(format!("{count} migration(s) on disk"));
let findings = (&mut info, &mut warnings, &mut errors);
let system = (system.as_slice(), view.scope);
pending_migration_audit(pool, dir, &files, system, ledgers, findings).await;
}
}
if deploy {
let mut audit = DeployAuditFindings::default();
run_deploy_audit(&deploy_audit_env(deploy_findings.is_some()), &mut audit);
#[cfg(feature = "config")]
run_settings_audit(&mut audit);
#[cfg(feature = "cache")]
login_store_audit(crate::account_lockout::shared(), &mut audit);
collation_audit(pool, &mut audit).await;
time_zone_audit(pool, &mut audit).await;
#[cfg(feature = "admin-sso")]
sso_link_audit(pool, &mut audit).await;
if let Some(extra) = deploy_findings {
let extra = extra.await;
audit.info.extend(extra.info);
audit.warnings.extend(extra.warnings);
audit.errors.extend(extra.errors);
}
let dialect = pool.dialect();
let active = dialect.name();
for entry in crate::core::inventory::iter::<crate::core::ModelEntry>.into_iter() {
if let Some(want) = entry.schema.required_db_vendor {
if want != active {
audit.warnings.push(format!(
"model `{}` declares `required_db_vendor = \"{want}\"` \
but the active database backend is `{active}` — \
queries that depend on backend-specific features may fail",
entry.schema.name,
));
}
}
for token in entry.schema.required_db_features {
if !dialect.supports(token) {
audit.warnings.push(format!(
"model `{}` declares `required_db_features` token `{token}` \
which the active `{active}` dialect does not advertise — \
queries depending on it will fail at runtime",
entry.schema.name,
));
}
}
}
info.extend(audit.info);
warnings.extend(audit.warnings);
errors.extend(audit.errors);
}
for msg in &info {
writeln!(w, " [info] {msg}")?;
}
for msg in &warnings {
writeln!(w, " [warning] {msg}")?;
}
for msg in &errors {
writeln!(w, " [error] {msg}")?;
}
if !errors.is_empty() {
return Err(MigrateError::Validation(format!(
"{} system check(s) failed",
errors.len()
)));
}
if warnings.is_empty() && errors.is_empty() {
writeln!(w, "all checks passed")?;
}
Ok(())
}
fn docs_cmd<W: Write>(w: &mut W) -> Result<(), MigrateError> {
let url = "https://docs.rs/rustango";
writeln!(w, "{url}")?;
if cfg!(test) || std::env::var_os("RUSTANGO_NO_BROWSER").is_some() {
return Ok(());
}
let opener = if cfg!(target_os = "macos") {
Some(("open", url))
} else if cfg!(target_os = "linux") {
Some(("xdg-open", url))
} else if cfg!(target_os = "windows") {
Some(("cmd", "/C start"))
} else {
None
};
if let Some((cmd, _)) = opener {
let _ = std::process::Command::new(cmd).arg(url).spawn();
}
Ok(())
}
fn version_cmd<W: Write>(w: &mut W) -> Result<(), MigrateError> {
writeln!(w, "rustango {}", env!("CARGO_PKG_VERSION"))?;
Ok(())
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum NameShape {
Type,
Identifier,
}
fn parse_name_and_model(args: &[String]) -> Result<(String, Option<String>, String), MigrateError> {
parse_name_and_model_as(args, NameShape::Type)
}
fn parse_name_and_model_as(
args: &[String],
shape: NameShape,
) -> Result<(String, Option<String>, String), MigrateError> {
let mut name: Option<String> = None;
let mut model: Option<String> = None;
let mut crate_root: Option<String> = None;
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--model" => {
model =
Some(iter.next().cloned().ok_or_else(|| {
MigrateError::Validation("--model requires a value".into())
})?);
}
"--crate" => {
let v = iter
.next()
.cloned()
.ok_or_else(|| MigrateError::Validation("--crate requires a value".into()))?;
let trimmed = v.trim();
if trimmed.is_empty() {
return Err(MigrateError::Validation(
"--crate <name> must not be empty".into(),
));
}
crate_root = Some(trimmed.to_owned());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag `{other}`")));
}
other => {
if name.is_some() {
return Err(MigrateError::Validation(format!(
"unexpected positional `{other}`"
)));
}
name = Some(other.to_owned());
}
}
}
let name = name.ok_or_else(|| {
MigrateError::Validation("name is required (e.g. `manage make:viewset PostViewSet`)".into())
})?;
match shape {
NameShape::Type if !is_valid_type_name(&name) => {
return Err(MigrateError::Validation(format!(
"`{name}` is not a valid Rust type name (PascalCase, alphanumeric + underscore)"
)));
}
NameShape::Identifier if !is_valid_identifier(&name) => {
return Err(MigrateError::Validation(format!(
"`{name}` is not a valid Rust identifier (letters, digits, underscore; \
must not start with a digit)"
)));
}
_ => {}
}
let module = pascal_to_snake(&name);
if name == "Self" || is_reserved_module_name(&module) {
return Err(MigrateError::Validation(format!(
"`{name}` would make the module `{module}`, which is a Rust keyword or \
shadows a built-in crate — pick another name"
)));
}
Ok((name, model, crate_root.unwrap_or_else(|| "rustango".into())))
}
pub(super) fn is_reserved_module_name(name: &str) -> bool {
const RESERVED: &[&str] = &[
"abstract",
"alloc",
"as",
"async",
"await",
"become",
"box",
"break",
"const",
"continue",
"core",
"crate",
"do",
"dyn",
"else",
"enum",
"extern",
"false",
"final",
"fn",
"for",
"gen",
"if",
"impl",
"in",
"let",
"loop",
"macro",
"match",
"mod",
"move",
"mut",
"override",
"priv",
"proc_macro",
"pub",
"ref",
"return",
"self",
"static",
"std",
"struct",
"super",
"trait",
"true",
"try",
"type",
"typeof",
"unsafe",
"unsized",
"use",
"virtual",
"where",
"while",
"yield",
];
RESERVED.contains(&name)
}
fn is_valid_type_name(name: &str) -> bool {
let bytes = name.as_bytes();
!bytes.is_empty()
&& bytes[0].is_ascii_uppercase()
&& bytes
.iter()
.all(|b| b.is_ascii_alphanumeric() || *b == b'_')
}
fn is_valid_identifier(name: &str) -> bool {
let bytes = name.as_bytes();
!bytes.is_empty()
&& (bytes[0].is_ascii_alphabetic() || bytes[0] == b'_')
&& bytes
.iter()
.all(|b| b.is_ascii_alphanumeric() || *b == b'_')
}
fn pascal_to_snake(s: &str) -> String {
let mut out = String::with_capacity(s.len() + 4);
for (i, c) in s.chars().enumerate() {
if c.is_ascii_uppercase() && i > 0 {
out.push('_');
}
out.push(c.to_ascii_lowercase());
}
out
}
fn write_source(path: &std::path::Path, body: &str) -> std::io::Result<()> {
match super::scaffold::renamed_dep(std::path::Path::new(".")) {
Some(to) => std::fs::write(path, super::scaffold::rename_dep(body, &to)),
None => std::fs::write(path, body),
}
}
fn write_generated<W: Write>(
w: &mut W,
file_name: &str,
contents: String,
) -> Result<(), MigrateError> {
let path = std::path::PathBuf::from("src").join(file_name);
if path.exists() {
return Err(MigrateError::Validation(format!(
"{} already exists — refusing to overwrite",
path.display()
)));
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
write_source(&path, &contents)?;
writeln!(w, "wrote {}", path.display())?;
let module = file_name.trim_end_matches(".rs");
if std::path::Path::new("src/lib.rs").exists() {
writeln!(w, " add `pub mod {module};` to src/lib.rs")?;
} else {
writeln!(
w,
" add `mod {module};` to src/main.rs (or `pub mod ...;` to src/lib.rs)"
)?;
}
Ok(())
}
const MAKE_VERBS: &[&str] = &[
"make:viewset",
"make:api_routes",
"make:serializer",
"make:form",
"make:job",
"make:scheduled",
"make:worker",
"make:notification",
"make:middleware",
"make:test",
];
fn make_cmd<W: Write>(
cmd: &str,
args: &[String],
w: &mut W,
has_feature: impl Fn(&str) -> bool,
) -> Result<(), MigrateError> {
if let Some(feature) = missing_make_feature(cmd, &has_feature) {
return Err(missing_feature_error(cmd, feature));
}
match cmd {
"make:viewset" => make_viewset_cmd(args, w, has_feature),
"make:api_routes" => make_api_routes_cmd(args, w),
"make:serializer" => make_serializer_cmd(args, w),
"make:form" => make_form_cmd(args, w),
"make:job" => make_job_cmd(args, w),
"make:scheduled" => make_scheduled_cmd(args, w),
"make:worker" => make_worker_cmd(args, w),
"make:notification" => make_notification_cmd(args, w),
"make:middleware" => make_middleware_cmd(args, w),
"make:test" => make_test_cmd(args, w),
other => Err(MigrateError::Validation(format!("unknown verb `{other}`"))),
}
}
fn missing_feature_error(cmd: &str, feature: &str) -> MigrateError {
MigrateError::Validation(format!(
"{cmd}: the code it writes needs rustango's `{feature}` feature, which this \
project does not enable; add it to the rustango dependency's features"
))
}
fn missing_make_feature(cmd: &str, has_feature: impl Fn(&str) -> bool) -> Option<&'static str> {
let feature = match cmd {
"make:test" => "admin",
"make:serializer" => "serializer",
"make:form" => "forms",
"make:job" | "make:worker" => "jobs",
"make:scheduled" => "scheduler",
"make:notification" => "notifications",
_ => return None,
};
(!has_feature(feature)).then_some(feature)
}
fn compiled_feature(feature: &str) -> bool {
match feature {
"admin" => cfg!(feature = "admin"),
"tenancy" => cfg!(feature = "tenancy"),
"serializer" => cfg!(feature = "serializer"),
"forms" => cfg!(feature = "forms"),
"jobs" => cfg!(feature = "jobs"),
"scheduler" => cfg!(feature = "scheduler"),
"notifications" => cfg!(feature = "notifications"),
_ => false,
}
}
struct ModelRef {
path: String,
name: String,
}
impl ModelRef {
fn bare(name: &str) -> Self {
Self {
path: format!("crate::models::{name}"),
name: name.to_owned(),
}
}
fn resolve(arg: &str, src: &std::path::Path) -> Self {
let rel = arg.strip_prefix("crate::").unwrap_or(arg);
if let Some((_, name)) = rel.rsplit_once("::") {
return Self {
path: format!("crate::{rel}"),
name: name.to_owned(),
};
}
let declares = |dir: &std::path::Path| {
[dir.join("models.rs"), dir.join("models").join("mod.rs")]
.iter()
.filter_map(|f| std::fs::read_to_string(f).ok())
.any(|text| declares_struct(&text, arg))
};
if declares(src) {
return Self::bare(arg);
}
let mut apps: Vec<std::path::PathBuf> = std::fs::read_dir(src)
.into_iter()
.flatten()
.flatten()
.map(|e| e.path())
.filter(|p| p.is_dir())
.collect();
apps.sort();
for app in apps {
let Some(app_name) = app.file_name().and_then(|n| n.to_str()) else {
continue;
};
if declares(&app) {
return Self {
path: format!("crate::{app_name}::models::{arg}"),
name: arg.to_owned(),
};
}
}
Self::bare(arg)
}
}
fn declares_struct(text: &str, name: &str) -> bool {
text.lines().any(|line| {
line.trim_start()
.strip_prefix("pub struct ")
.and_then(|rest| rest.strip_prefix(name))
.is_some_and(|rest| !rest.starts_with(|c: char| c.is_alphanumeric() || c == '_'))
})
}
fn make_viewset_cmd<W: Write>(
args: &[String],
w: &mut W,
has_feature: impl Fn(&str) -> bool,
) -> Result<(), MigrateError> {
let mut explicit_tenant = false;
let mut explicit_no_tenant = false;
let mut filtered: Vec<String> = Vec::with_capacity(args.len());
for a in args {
if a == "--tenant" || a == "--tenant-aware" {
explicit_tenant = true;
} else if a == "--no-tenant" {
explicit_no_tenant = true;
} else {
filtered.push(a.clone());
}
}
let (tenant_aware, echoed_auto_detect) =
resolve_viewset_tenant_mode(explicit_tenant, explicit_no_tenant, project_uses_tenancy());
if echoed_auto_detect {
writeln!(
w,
"make:viewset: auto-detected tenancy mode from Cargo.toml (pass `--no-tenant` to override)"
)?;
}
let (verb, feature) = if tenant_aware {
("make:viewset --tenant", "tenancy")
} else {
("make:viewset", "admin")
};
if !has_feature(feature) {
return Err(missing_feature_error(verb, feature));
}
let (name, model, crate_root) = parse_name_and_model(&filtered)?;
let snake = pascal_to_snake(&name);
let model = ModelRef::resolve(
model.as_deref().unwrap_or("Post"),
std::path::Path::new("src"),
);
let body = if tenant_aware {
viewset_template_tenant(&name, &model, &snake, &crate_root)
} else {
viewset_template_pool(&name, &model, &snake, &crate_root, &project_crate_name())
};
write_generated(w, &format!("{snake}.rs"), body)
}
fn viewset_template_pool(
name: &str,
model: &ModelRef,
snake: &str,
crate_root: &str,
app_crate: &str,
) -> String {
let ModelRef {
path: model_path,
name: model,
} = model;
format!(
r#"//! Auto-scaffolded by `manage make:viewset {name}`.
use {crate_root}::ViewSet;
use {model_path};
#[derive(ViewSet)]
#[viewset(
model = {model},
// Required: the only fields the API renders and a body may write.
fields = "id, ",
filter_fields = "",
search_fields = "",
page_size = 20,
// Writes need a guard: add `permissions(create = "...", ...)` behind
// an auth layer, then drop `read_only`.
read_only,
)]
pub struct {name};
// Mount in src/main.rs: `urls::api()` has no pool, `nest_with` passes
// the server's. Nest a prefix once; merge more viewsets in one closure.
//
// .nest_with("/api", |pool| {app_crate}::{snake}::{name}::router("/{snake}", pool))
"#
)
}
fn viewset_template_tenant(name: &str, model: &ModelRef, snake: &str, crate_root: &str) -> String {
let ModelRef {
path: model_path,
name: model,
} = model;
format!(
r#"//! Auto-scaffolded by `manage make:viewset {name} --tenant`.
//!
//! Tenant-aware viewset: each request resolves the connection via
//! `{crate_root}::extractors::Tenant`, so the same `router()` serves
//! every tenant under their own subdomain / schema / database.
//!
//! Since v0.30 (#80), `tenant_router` carries the full static-router
//! builder chain — filter_fields / search_fields / ordering /
//! page_size / permissions_for_model all work in tenant mode too.
use std::sync::Arc;
use axum::Router;
use {crate_root}::core::Model as _;
use {crate_root}::tenancy::{{ModelBackend, RouterAuthExt as _}};
use {crate_root}::viewset::ViewSet;
use {model_path};
pub fn router() -> Router<()> {{
// The only fields the API renders and a body may write.
ViewSet::for_model({model}::SCHEMA, &["id"])
// .filter_fields(&["status", "owner_id"]) // ?status=draft&owner_id=42
// .search_fields(&["name", "description"]) // ?search=foo (ILIKE)
// .ordering(&[("created_at", true)]) // default ORDER BY
// .ordering_fields(&["name", "created_at"]) // ?ordering=-name allowlist
// .page_size(20)
.permissions_for_model::<{model}>() // CRUD codenames
// .read_only() // GET only
.tenant_router("/api/{snake}")
// The permission check needs a tenant user. `ModelBackend` is HTTP
// Basic only; a browser session from `/login` is not read here (401).
// Token clients: add `ApiKeyBackend` or a `JwtBackend` to the list.
.require_auth(vec![Arc::new(ModelBackend)])
}}
// Mount in your urls.rs:
//
// .merge(crate::{snake}::router())
"#
)
}
fn resolve_viewset_tenant_mode(
explicit_tenant: bool,
explicit_no_tenant: bool,
project_tenancy: bool,
) -> (bool, bool) {
if explicit_no_tenant {
return (false, false);
}
if explicit_tenant {
return (true, false);
}
(project_tenancy, project_tenancy)
}
fn project_uses_tenancy() -> bool {
let Ok(s) = std::fs::read_to_string("Cargo.toml") else {
return false;
};
let lower = s.to_ascii_lowercase();
let has_inline = lower.contains("rustango")
&& lower
.lines()
.any(|line| line.contains("rustango") && line.contains("\"tenancy\""));
let has_table_block =
lower.contains("[dependencies.rustango]") && lower.contains("\"tenancy\"");
has_inline || has_table_block
}
fn project_crate_name() -> String {
std::fs::read_to_string("Cargo.toml")
.ok()
.as_deref()
.and_then(package_name_from_cargo_toml)
.unwrap_or_else(|| "your_app".to_string())
}
fn package_name_from_cargo_toml(src: &str) -> Option<String> {
let mut in_package = false;
for line in src.lines() {
let trimmed = line.trim();
if trimmed.starts_with('[') {
in_package = trimmed == "[package]";
continue;
}
if !in_package {
continue;
}
if let Some(value) = trimmed
.strip_prefix("name")
.map(str::trim_start)
.and_then(|rest| rest.strip_prefix('='))
{
let name = value.trim().trim_matches('"');
if !name.is_empty() {
return Some(name.replace('-', "_"));
}
}
}
None
}
fn make_api_routes_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let mut tenant_aware = false;
let mut crate_root: Option<String> = None;
let mut filtered: Vec<String> = Vec::with_capacity(args.len());
let mut iter = args.iter();
while let Some(a) = iter.next() {
if a == "--tenant" || a == "--tenant-aware" {
tenant_aware = true;
} else if a == "--crate" {
let v = iter
.next()
.cloned()
.ok_or_else(|| MigrateError::Validation("--crate requires a value".into()))?;
let trimmed = v.trim();
if trimmed.is_empty() {
return Err(MigrateError::Validation(
"--crate <name> must not be empty".into(),
));
}
crate_root = Some(trimmed.to_owned());
} else if a == "--help" || a == "-h" {
writeln!(w, "make:api_routes <app> [--tenant] [--crate <name>]")?;
writeln!(
w,
" Scaffold src/<app>/api_routes.rs — the per-app router composer."
)?;
writeln!(
w,
" Use --tenant for tenancy projects (no PgPool argument; each"
)?;
writeln!(
w,
" viewset resolves its own per-request connection via the Tenant"
)?;
writeln!(w, " extractor).")?;
writeln!(
w,
" --crate <name> overrides the emitted `use …::sql::Pool;`"
)?;
writeln!(w, " crate root (default: `rustango`).")?;
return Ok(());
} else if a.starts_with('-') {
return Err(MigrateError::Validation(format!(
"make:api_routes: unrecognized flag `{a}`"
)));
} else {
filtered.push(a.clone());
}
}
let crate_root = crate_root.unwrap_or_else(|| "rustango".into());
let app = filtered.first().ok_or_else(|| {
MigrateError::Validation(
"app name is required (e.g. `manage make:api_routes regions`)".into(),
)
})?;
if filtered.len() > 1 {
return Err(MigrateError::Validation(format!(
"make:api_routes: expected one app name, got {} ({:?})",
filtered.len(),
filtered
)));
}
if !is_valid_app_name(app) {
return Err(MigrateError::Validation(format!(
"make:api_routes: app name `{app}` must match `[a-z_][a-z0-9_]*`"
)));
}
let app_dir = std::path::PathBuf::from("src").join(app);
if !app_dir.exists() {
return Err(MigrateError::Validation(format!(
"make:api_routes: src/{app}/ does not exist. Run `manage startapp {app}` first."
)));
}
let path = app_dir.join("api_routes.rs");
if path.exists() {
return Err(MigrateError::Validation(format!(
"{} already exists — refusing to overwrite",
path.display()
)));
}
let body = if tenant_aware {
api_routes_template_tenant(app)
} else {
api_routes_template_pool(app, &crate_root)
};
write_source(&path, &body)?;
writeln!(w, "wrote {}", path.display())?;
writeln!(w, " add `pub mod api_routes;` to src/{app}/mod.rs,")?;
if tenant_aware {
writeln!(w, " then in src/urls.rs, inside `api()`:")?;
writeln!(w, " .merge(crate::{app}::api_routes::api())")?;
} else {
let app_crate = project_crate_name();
writeln!(w, " then in src/main.rs, after `.api(urls::api())`:")?;
writeln!(
w,
" .nest_with(\"/api\", |pool| {app_crate}::{app}::api_routes::api(pool))"
)?;
}
Ok(())
}
fn is_valid_app_name(s: &str) -> bool {
let mut chars = s.chars();
match chars.next() {
Some(c) if c.is_ascii_lowercase() || c == '_' => {}
_ => return false,
}
chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
}
fn api_routes_template_tenant(app: &str) -> String {
format!(
r#"//! Auto-scaffolded by `manage make:api_routes {app} --tenant`.
//!
//! API routing for the `{app}` app: composes tenant viewsets into a
//! single `Router<()>`.
//!
//! Adding a resource:
//! 1. Run `manage make:viewset <Name> --model <Model> --tenant`.
//! 2. Add one `.merge(...)` line below.
use axum::Router;
pub fn api() -> Router<()> {{
Router::new()
// .merge(crate::<snake>::router())
}}
"#
)
}
fn api_routes_template_pool(app: &str, crate_root: &str) -> String {
format!(
r#"//! Auto-scaffolded by `manage make:api_routes {app}`.
//!
//! API routing for the `{app}` app. Composes per-model viewsets
//! into a single `Router<()>`. Each viewset captures the supplied
//! pool at mount time.
//!
//! The pool is `sql::Pool`, not a driver-typed one: it dispatches on
//! the `DATABASE_URL` scheme, so this file compiles and runs on all
//! three backends. A `PgPool` here would pin the whole app to
//! Postgres — `ViewSet::router_pool` takes exactly this type.
//!
//! Adding a resource:
//! 1. Run `manage make:viewset <Name> --model <Model>`.
//! 2. Add one `.merge(...)` line below.
use axum::Router;
use {crate_root}::sql::Pool;
pub fn api(pool: Pool) -> Router<()> {{
let _pool = pool;
Router::new()
// `main.rs` nests this under `/api`.
// .merge(crate::<snake>::<Name>::router("/<snake>", _pool.clone()))
}}
"#
)
}
fn make_serializer_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, model, crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let model = ModelRef::resolve(
model.as_deref().unwrap_or("Post"),
std::path::Path::new("src"),
);
write_generated(
w,
&format!("{snake}.rs"),
serializer_template(&name, &model, &crate_root),
)
}
fn serializer_template(name: &str, model: &ModelRef, crate_root: &str) -> String {
let ModelRef {
path: model_path,
name: model,
} = model;
format!(
r#"//! Auto-scaffolded by `manage make:serializer {name}`.
use {crate_root}::sql::Auto;
use {crate_root}::Serializer;
use {model_path};
#[derive(Serializer, serde::Deserialize, Default)]
#[serializer(model = {model})]
pub struct {name} {{
// A serializer field must match its model field's type **exactly**.
// The scaffolders emit `pub id: Auto<i64>` for a primary key, so
// `pub id: i64` here does not compile — it fails inside the derive
// with `expected &i64, found &Auto<i64>`, which does not obviously
// point back to this line.
//
// `read_only` keeps it out of the writable set while still
// returning it, which is what you want: drop the field entirely and
// the API answers a create with no identifier, so the client cannot
// address what it just made.
#[serializer(read_only)]
pub id: Auto<i64>,
// pub title: String,
// #[serializer(source = "body")] // publish under a different name
// pub content: String,
}}
"#
)
}
fn make_form_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let body = format!(
r#"//! Auto-scaffolded by `manage make:form {name}`.
use {crate_root}::Form as DeriveForm;
#[derive(DeriveForm)]
pub struct {name} {{
#[form(min_length = 1, max_length = 200)]
pub title: String,
pub body: Option<String>,
}}
"#
);
write_generated(w, &format!("{snake}.rs"), body)
}
fn make_job_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let body = format!(
r#"//! Auto-scaffolded by `manage make:job {name}`.
//!
//! A background job: enqueued from a handler, executed later by a
//! worker. For work that runs on a timer instead, see
//! `manage make:scheduled`.
use {crate_root}::jobs::{{Job, JobError}};
use serde::{{Deserialize, Serialize}};
/// The payload. `run` receives **only this** — no pool, no tenant, no
/// request context: workers are spawned tasks and inherit no
/// task-local state. Carry everything the job needs in these fields.
#[derive(Serialize, Deserialize)]
pub struct {name} {{
pub id: i64,
}}
#[async_trait::async_trait] // add `async-trait` to your Cargo.toml
impl Job for {name} {{
/// Globally unique: this is the routing key a worker matches a
/// queued row against. Two job types must not share one.
const NAME: &'static str = "{snake}";
/// A ceiling on **total attempts**, not on retries — 5 is one run
/// plus four retries. Backoff between them is 1s, 2s, 4s, 8s.
const MAX_ATTEMPTS: u32 = 5;
async fn run(&self) -> Result<(), JobError> {{
// Err(JobError::Retryable(..)) to back off and try again;
// Err(JobError::Fatal(..)) to dead-letter immediately.
let _ = self.id;
Ok(())
}}
}}
// Wire up wherever you build the queue — the web process, a worker
// binary from `make:worker`, or both:
//
// queue.register::<{name}>().await; // BEFORE start(); every
// // process that calls start()
// // must register every type it
// // might pick up, or the row is
// // locked and abandoned
// queue.start().await;
//
// and to enqueue one, from a handler:
//
// queue.dispatch(&{name} {{ id: 42 }}).await?;
"#
);
write_generated(w, &format!("{snake}.rs"), body)
}
fn make_scheduled_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let app_crate = project_crate_name();
let body = format!(
r#"//! Auto-scaffolded by `manage make:scheduled {name}`.
//!
//! A task that runs on a timer, not in response to a request. For work
//! enqueued by a handler and executed by a worker, see
//! `manage make:job`.
use std::sync::Arc;
use {crate_root}::sql::Pool;
pub struct {name} {{
pub pool: Pool,
}}
impl {name} {{
pub async fn run(self: Arc<Self>) {{
// The scheduler isolates panics per task, but a task that
// returns early on error simply skips that tick — log, do not
// swallow silently.
let _ = &self.pool;
}}
}}
// Start it from a worker binary (`manage make:worker`), not main.rs:
// main.rs has no pool, and code before `Cli::run()` runs for every CLI
// verb. In the worker, after `let pool = ...;`:
//
// let scheduler = {crate_root}::scheduler::Scheduler::new();
// let task = Arc::new({app_crate}::{snake}::{name} {{ pool: pool.clone() }});
// scheduler.every("{snake}", std::time::Duration::from_secs(60), move || {{
// let task = Arc::clone(&task);
// async move {{ task.run().await }}
// }});
// let _schedule = scheduler.start(); // ticks until dropped
//
// The first run happens after one full interval, not immediately.
"#
);
write_generated(w, &format!("{snake}.rs"), body)
}
fn make_worker_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let body = worker_template(&name, &snake, &crate_root, &project_crate_name());
write_generated_bin(w, &snake, body)
}
fn worker_template(name: &str, snake: &str, crate_root: &str, app_crate: &str) -> String {
format!(
r#"//! Auto-scaffolded by `manage make:worker {name}`.
//!
//! A standalone worker process: it drains the job queue and serves no
//! HTTP. Run it alongside the web process, or as its own container.
//!
//! Put this at `src/bin/{snake}.rs` and run it with `cargo run --bin {snake}`.
use std::sync::Arc;
use std::time::Duration;
use {crate_root}::jobs::{{DatabaseJobQueue, JobQueue}};
use {crate_root}::sql::Pool;
#[{crate_root}::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {{
// A worker never runs main.rs, so it loads `.env` itself.
let _ = dotenvy::dotenv();
// DATABASE_URL, or DB_USER + DB_NAME; the error names both.
let url = {crate_root}::env::database_url_from_env().map_err(|e| {{
format!("{{e}}. Set it in your shell, or copy '.env.example' to '.env'.")
}})?;
let pool = Pool::connect(&url).await?;
// The queue is tri-dialect despite the `Database` name — the table
// DDL and the row-pickup strategy are chosen from the pool's dialect.
DatabaseJobQueue::ensure_table_pool(&pool).await?;
// A worker killed mid-job (OOM, SIGKILL) never reaches its shutdown,
// so its rows stay locked. This unlocks them at start and every minute.
let queue = Arc::new(
DatabaseJobQueue::with_workers_pool(pool.clone(), 4)
.reclaim_stuck_after(Duration::from_secs(300)),
);
// Register EVERY job type this queue might see, not just the ones
// this process dispatches. A worker that picks up a row whose name
// is unregistered here logs and returns *without unlocking it* — the
// row is then stranded until the reclaim sweep, and
// it does not show up in `pending_count()`.
//
// Note the crate name: this file is its own binary crate, so `crate::`
// would mean *this* worker. Job types live in the library, and
// `manage make:job WelcomeEmail` writes `src/welcome_email.rs`.
//
// queue.register::<{app_crate}::welcome_email::WelcomeEmail>().await;
queue.start().await;
tracing::info!("{snake}: draining jobs");
// SIGINT *and* SIGTERM. `tokio::signal::ctrl_c()` alone is
// SIGINT-only, so under `docker stop` the drain below never runs.
{crate_root}::shutdown::shutdown_signal().await;
tracing::info!("{snake}: signal received, draining in-flight jobs");
queue.shutdown().await;
Ok(())
}}
"#
)
}
fn write_generated_bin<W: Write>(
w: &mut W,
bin_name: &str,
contents: String,
) -> Result<(), MigrateError> {
let path = std::path::PathBuf::from("src")
.join("bin")
.join(format!("{bin_name}.rs"));
if path.exists() {
return Err(MigrateError::Validation(format!(
"{} already exists — refusing to overwrite",
path.display()
)));
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
write_source(&path, &contents)?;
writeln!(w, "wrote {}", path.display())?;
writeln!(w, " run it with `cargo run --bin {bin_name}`")?;
writeln!(
w,
" NOTE: a second binary makes plain `cargo run` ambiguous. Add\n \
`default-run = \"<your-app>\"` under [package] in Cargo.toml to keep\n \
`cargo run -- migrate` working."
)?;
Ok(())
}
fn make_notification_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let body = format!(
r#"//! Auto-scaffolded by `manage make:notification {name}`.
//!
//! User-facing notification. For now this just builds an Email; once the
//! `{crate_root}::notifications` layer ships you'll add `via()` for multi-channel.
use {crate_root}::email::Email;
pub struct {name} {{
pub user_email: String,
pub subject: String,
}}
impl {name} {{
pub fn build_email(&self) -> Email {{
Email::new()
.to(&self.user_email)
.from("noreply@example.com")
.subject(&self.subject)
.body("Hello — this notification was generated by {name}.")
}}
}}
"#
);
write_generated(w, &format!("{snake}.rs"), body)
}
fn make_middleware_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, _crate_root) = parse_name_and_model(args)?;
let snake = pascal_to_snake(&name);
let body = format!(
r#"//! Auto-scaffolded by `manage make:middleware {name}`.
use axum::body::Body;
use axum::http::{{Request, Response}};
use axum::middleware::Next;
pub async fn {snake}(req: Request<Body>, next: Next) -> Response<Body> {{
// TODO: pre-handler logic
let response = next.run(req).await;
// TODO: post-handler logic
response
}}
// Apply with:
// router.layer(axum::middleware::from_fn({snake}))
"#
);
write_generated(w, &format!("{snake}.rs"), body)
}
fn make_test_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let (name, _, crate_root) = parse_name_and_model_as(args, NameShape::Identifier)?;
let snake = pascal_to_snake(&name);
let body = format!(
r#"//! Auto-scaffolded by `manage make:test {name}`.
//!
//! Integration test. Run with `cargo test --test {snake}`.
use {crate_root}::test_client::TestClient;
use axum::Router;
use axum::routing::get;
fn app() -> Router {{
// This test touches no database. As soon as it does — the moment
// `app()` opens a pool from `DATABASE_URL` — uncomment this. A test
// in `tests/` is its own crate and never runs `main`, so nothing has
// loaded `.env` for it and the variable reads as unset (#1299).
// let _ = dotenvy::dotenv();
Router::new().route("/hello", get(|| async {{ "hi" }}))
}}
#[tokio::test]
async fn {snake}() {{
let client = TestClient::new(app());
let r = client.get("/hello").send().await;
assert_eq!(r.status, 200);
assert_eq!(r.text(), "hi");
}}
"#
);
let path = std::path::PathBuf::from("tests").join(format!("{snake}.rs"));
if path.exists() {
return Err(MigrateError::Validation(format!(
"{} already exists — refusing to overwrite",
path.display()
)));
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
write_source(&path, &body)?;
writeln!(w, "wrote {}", path.display())?;
Ok(())
}
#[derive(Debug, PartialEq)]
struct DbDumpArgs {
out: Option<String>,
data_only: bool,
schema_only: bool,
no_owner: bool,
}
fn parse_db_dump_args(args: &[String]) -> Result<DbDumpArgs, MigrateError> {
let mut out: Option<String> = None;
let mut data_only = false;
let mut schema_only = false;
let mut no_owner = false;
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--out" | "-o" => {
out = Some(
iter.next()
.cloned()
.ok_or_else(|| MigrateError::Validation("--out requires a path".into()))?,
);
}
"--data-only" => data_only = true,
"--schema-only" => schema_only = true,
"--no-owner" => no_owner = true,
other => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
}
}
if data_only && schema_only {
return Err(MigrateError::Validation(
"--data-only and --schema-only are mutually exclusive".into(),
));
}
Ok(DbDumpArgs {
out,
data_only,
schema_only,
no_owner,
})
}
fn pg_dump_command(parsed: &DbDumpArgs, url: &str) -> Result<std::process::Command, MigrateError> {
let mut cmd = crate::dbshell::LibpqConn::new(url)
.map_err(MigrateError::Validation)?
.command("pg_dump");
cmd.args(build_pg_dump_argv(parsed));
Ok(cmd)
}
fn build_pg_dump_argv(parsed: &DbDumpArgs) -> Vec<String> {
let mut argv = Vec::new();
if parsed.data_only {
argv.push("--data-only".into());
}
if parsed.schema_only {
argv.push("--schema-only".into());
}
if parsed.no_owner {
argv.push("--no-owner".into());
}
if let Some(out) = &parsed.out {
argv.push("--file".into());
argv.push(out.clone());
}
argv
}
fn db_dump_cmd(args: &[String]) -> Result<(), MigrateError> {
let parsed = parse_db_dump_args(args)?;
let url = crate::env::database_url_from_env()
.map_err(|e| MigrateError::Validation(format!("db:dump needs a database: {e}")))?;
let mut cmd = pg_dump_command(&parsed, &url)?;
eprintln!("running: pg_dump {}", shown_args(&cmd));
let status = cmd.status().map_err(|e| {
MigrateError::Validation(format!("could not run pg_dump (is it on PATH?): {e}"))
})?;
if !status.success() {
return Err(MigrateError::Validation(format!(
"pg_dump exited with status {status}"
)));
}
Ok(())
}
#[derive(Debug, PartialEq)]
struct DbRestoreArgs {
file: String,
clean: bool,
yes: bool,
}
fn parse_db_restore_args(args: &[String]) -> Result<DbRestoreArgs, MigrateError> {
let mut file: Option<String> = None;
let mut clean = false;
let mut yes = false;
for arg in args {
match arg.as_str() {
"--clean" => clean = true,
"--yes" => yes = true,
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
if file.is_some() {
return Err(MigrateError::Validation(format!(
"unexpected argument: {other}"
)));
}
file = Some(other.to_owned());
}
}
}
let file = file.ok_or_else(|| {
MigrateError::Validation("db:restore <file> requires a dump file path".into())
})?;
Ok(DbRestoreArgs { file, clean, yes })
}
#[derive(Debug)]
struct RestorePlan {
file: String,
clean: bool,
}
impl RestorePlan {
fn check(
parsed: DbRestoreArgs,
confirm: impl FnOnce(&str) -> bool,
) -> Result<Self, MigrateError> {
let usable = std::fs::metadata(&parsed.file)
.map_err(|e| e.to_string())
.and_then(|m| {
if !parsed.clean {
Ok(())
} else if !m.is_file() {
Err("--clean needs a regular file".to_owned())
} else if m.len() == 0 {
Err("the file is empty".to_owned())
} else {
Ok(())
}
});
if let Err(e) = usable {
return Err(MigrateError::Validation(format!(
"db:restore: cannot read `{}`: {e}",
parsed.file
)));
}
if parsed.clean && !parsed.yes && !confirm(&parsed.file) {
return Err(MigrateError::Validation(
"db:restore --clean drops schema `public` first: pass --yes to confirm".into(),
));
}
Ok(Self {
file: parsed.file,
clean: parsed.clean,
})
}
}
fn confirm_clean_restore(file: &str) -> bool {
#[cfg(any(feature = "admin", feature = "tenancy"))]
{
let prompt =
format!("db:restore --clean drops schema `public`, then loads {file}. Type `yes`: ");
answered_yes(crate::manage_interactive::ask(&prompt))
}
#[cfg(not(any(feature = "admin", feature = "tenancy")))]
{
let _ = file;
false
}
}
#[cfg_attr(not(any(feature = "admin", feature = "tenancy")), allow(dead_code))]
fn answered_yes(answer: std::io::Result<Option<String>>) -> bool {
matches!(answer, Ok(Some(a)) if a == "yes")
}
fn psql_restore_command(
psql: &str,
parsed: &RestorePlan,
url: &str,
) -> Result<std::process::Command, MigrateError> {
let mut cmd = crate::dbshell::LibpqConn::new(url)
.map_err(MigrateError::Validation)?
.command(psql);
cmd.args(build_psql_argv(parsed));
Ok(cmd)
}
fn build_psql_argv(parsed: &RestorePlan) -> Vec<String> {
let mut argv = Vec::new();
argv.push("-v".into());
argv.push("ON_ERROR_STOP=1".into());
argv.push("--single-transaction".into());
if parsed.clean {
argv.push("-c".into());
argv.push("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;".into());
}
argv.push("-f".into());
argv.push(parsed.file.clone());
argv
}
fn db_restore_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let parsed = RestorePlan::check(parse_db_restore_args(args)?, confirm_clean_restore)?;
let url = crate::env::database_url_from_env()
.map_err(|e| MigrateError::Validation(format!("db:restore needs a database: {e}")))?;
run_psql_restore("psql", &parsed, &url, w)
}
fn run_psql_restore<W: Write>(
psql: &str,
parsed: &RestorePlan,
url: &str,
w: &mut W,
) -> Result<(), MigrateError> {
let mut cmd = psql_restore_command(psql, parsed, url)?;
writeln!(w, "running: psql {}", shown_args(&cmd))?;
let status = cmd.status().map_err(|e| {
MigrateError::Validation(format!("could not run psql (is it on PATH?): {e}"))
})?;
if !status.success() {
return Err(MigrateError::Validation(format!(
"psql exited with status {status}"
)));
}
Ok(())
}
fn db_info_cmd<W: Write>(w: &mut W) -> Result<(), MigrateError> {
writeln!(w, "rustango db:info")?;
writeln!(w, " framework version: {}", env!("CARGO_PKG_VERSION"))?;
let pg_enabled = cfg!(feature = "postgres");
let mysql_enabled = cfg!(feature = "mysql");
writeln!(
w,
" postgres feature: {}",
if pg_enabled { "enabled" } else { "disabled" }
)?;
writeln!(
w,
" mysql feature: {} (impl lands in v0.23.0-batch2)",
if mysql_enabled { "enabled" } else { "disabled" }
)?;
match crate::env::database_url_from_env() {
Ok(url) => {
let scheme = url.split("://").next().unwrap_or("(unknown)");
writeln!(w, " resolved URL: {}", redact_url(&url))?;
writeln!(w, " detected backend: {scheme}")?;
match scheme {
"postgres" | "postgresql" if !pg_enabled => {
writeln!(
w,
" ! warning: URL is postgres but the `postgres` feature is disabled — \
add `features = [\"postgres\"]` to rustango"
)?;
}
"mysql" if !mysql_enabled => {
writeln!(
w,
" ! warning: URL is mysql but the `mysql` feature is disabled — \
add `features = [\"mysql\"]` to rustango"
)?;
}
"mysql" if mysql_enabled => {
writeln!(
w,
" ! note: MySql connections will fail in v0.23.0-batch1 \
(MySqlDialect lands in batch2)"
)?;
}
_ => {}
}
}
Err(e) => {
writeln!(w, " resolved URL: (none — {e})")?;
writeln!(
w,
" hint: set DATABASE_URL or DB_USER+DB_NAME (+optional \
DB_HOST/DB_PORT/DB_PASSWORD/DB_DRIVER/DB_PARAMS)"
)?;
}
}
Ok(())
}
#[derive(Debug, Default, PartialEq)]
struct DumpdataArgs {
model_filters: Vec<String>,
excludes: Vec<String>,
indent: usize,
help: bool,
}
fn parse_dumpdata_args(args: &[String]) -> Result<DumpdataArgs, MigrateError> {
let mut out = DumpdataArgs {
indent: 2,
..Default::default()
};
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
out.help = true;
return Ok(out);
}
"--model" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--model expects a value".into()))?;
out.model_filters.push(v.clone());
}
"--exclude" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--exclude expects a value".into()))?;
out.excludes.push(v.clone());
}
"--indent" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--indent expects a value".into()))?;
out.indent = v.parse().map_err(|e| {
MigrateError::Validation(format!("--indent: not an integer ({e})"))
})?;
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
Ok(out)
}
pub(crate) async fn dumpdata_cmd<W: Write>(
pool: &Pool,
args: &[String],
scope: DataScope<'_>,
w: &mut W,
) -> Result<(), MigrateError> {
let parsed = parse_dumpdata_args(args)?;
if parsed.help {
writeln!(
w,
"dumpdata [--model app.Name] [--exclude app.Name] [--indent N]"
)?;
writeln!(w)?;
writeln!(
w,
" Export every registered model's rows as JSON in fixture"
)?;
writeln!(
w,
" shape: `[{{\"model\": \"app.Model\", \"pk\": N, \"fields\": {{...}}}}]`."
)?;
writeln!(w)?;
writeln!(
w,
" --model <name> Limit to a single model. Accepts either the full"
)?;
writeln!(
w,
" `app.Model` shape or the bare model name. Pass"
)?;
writeln!(
w,
" the flag multiple times to limit to a set."
)?;
writeln!(
w,
" --indent <N> JSON indent (default 2; 0 emits compact single-line)."
)?;
writeln!(
w,
" --exclude <name> Leave a model out (repeatable; same forms as --model)."
)?;
return Ok(());
}
let model_filters = &parsed.model_filters;
let indent = parsed.indent;
for name in model_filters.iter().chain(&parsed.excludes) {
let mut named = inventory::iter::<crate::core::ModelEntry>()
.filter(|e| model_matches(name, &dotted_model_name(e), e.schema))
.peekable();
let Some(owner) = named.peek().map(|e| e.schema.scope) else {
return Err(MigrateError::Validation(format!(
"dumpdata: unknown model `{name}` (use `app.Model` or `Model`)"
)));
};
if !named.any(|e| in_scope(e.schema, scope)) {
return Err(MigrateError::Validation(format!(
"dumpdata: {}",
out_of_scope(name, owner)
)));
}
}
let selected: Vec<&crate::core::ModelEntry> = inventory::iter::<crate::core::ModelEntry>()
.filter(|e| in_scope(e.schema, scope))
.filter(|e| {
let dotted = dotted_model_name(e);
let matches = |f: &String| model_matches(f, &dotted, e.schema);
(model_filters.is_empty() || model_filters.iter().any(matches))
&& !parsed.excludes.iter().any(matches)
})
.collect();
let schemas: Vec<_> = selected.iter().map(|e| e.schema).collect();
require_tenant_tables(pool, scope, &schemas, "dumpdata").await?;
let mut out = FixtureWriter::new(w, indent);
for entry in selected {
let schema = entry.schema;
let dotted_name = dotted_model_name(entry);
if !schema.managed && !runner::try_table_resolves(pool, schema.table).await? {
continue;
}
if let Some(f) = schema.scalar_fields().find(|f| !dumpable(f.ty)) {
return Err(MigrateError::Validation(format!(
"dumpdata: `{dotted_name}.{}` is a {:?} column, which dumpdata cannot \
export yet — leave it out with `--exclude {dotted_name}`",
f.name, f.ty
)));
}
let pk_field = schema.primary_key();
let fields: Vec<&'static crate::core::FieldSchema> = schema.scalar_fields().collect();
let paged = match pk_field {
Some(pk) => is_sole_pk(pool, schema.table, pk.column).await?,
None => false,
};
let mut after: Option<crate::core::SqlValue> = None;
loop {
let mut query = crate::core::SelectQuery::new(schema);
if let Some(pk) = pk_field {
query.order_by = vec![crate::core::OrderClause {
column: pk.column,
desc: false,
}
.into()];
}
if let Some(pk) = pk_field.filter(|_| paged) {
query.limit = Some(DUMP_PAGE);
if let Some(last) = after.take() {
query.where_clause = crate::core::WhereExpr::Predicate(crate::core::Filter {
column: pk.column,
op: crate::core::Op::Gt,
value: last,
});
}
}
let rows = crate::sql::select_rows_as_json_with(pool, &query, &fields, fixture_json)
.await
.map_err(|e| {
MigrateError::Validation(format!(
"dumpdata: select from `{}` failed: {e}",
schema.table
))
})?;
let full = paged && rows.len() == DUMP_PAGE as usize;
let pks: Vec<crate::core::SqlValue> = match pk_field {
Some(pk) => rows
.iter()
.filter_map(|r| r.get(pk.name))
.map(|v| json_to_sql_value(v, pk).map_err(MigrateError::Validation))
.collect::<Result<_, _>>()?,
None => Vec::new(),
};
let mut links: Vec<(&str, LinksBySource)> = Vec::new();
for chunk in pks.chunks(DUMP_PAGE as usize) {
for (name, by_src) in m2m_links(pool, schema, chunk).await? {
match links.iter_mut().find(|(n, _)| *n == name) {
Some((_, all)) => all.extend(by_src),
None => links.push((name, by_src)),
}
}
}
for mut row in rows {
let pk_value = match pk_field {
Some(pk) => row
.as_object_mut()
.and_then(|m| m.remove(pk.name))
.unwrap_or(serde_json::Value::Null),
None => serde_json::Value::Null,
};
if let Some(map) = row.as_object_mut() {
for (name, by_src) in &links {
let dsts = by_src.get(&pk_value.to_string()).cloned();
map.insert((*name).to_owned(), dsts.unwrap_or_default().into());
}
}
out.entry(&serde_json::json!({
"model": dotted_name,
"pk": pk_value,
"fields": row,
}))?;
}
if !full {
break;
}
after = pks.last().cloned();
}
}
out.finish()
}
const JSON_TAG: &str = "$json";
fn fixture_json(v: serde_json::Value) -> serde_json::Value {
if v.is_null() || json_tagged(&v).is_some() {
serde_json::json!({ JSON_TAG: v })
} else {
v
}
}
fn json_tagged(v: &serde_json::Value) -> Option<&serde_json::Value> {
v.as_object()
.filter(|m| m.len() == 1)
.and_then(|m| m.get(JSON_TAG))
}
fn loaddata_value(
v: &serde_json::Value,
field: &crate::core::FieldSchema,
) -> Result<crate::core::SqlValue, String> {
use crate::core::{FieldType, SqlValue};
if matches!(field.ty, FieldType::Json) {
if let Some(inner) = json_tagged(v) {
return Ok(SqlValue::Json(inner.clone()));
}
if v.is_null() && !field.nullable {
return Ok(SqlValue::Json(serde_json::Value::Null));
}
}
json_to_sql_value(v, field)
}
const DUMP_PAGE: i64 = 500;
#[derive(Debug, Clone, Copy)]
pub(crate) enum DataScope<'a> {
All,
#[cfg_attr(not(feature = "tenancy"), allow(dead_code))]
Registry,
#[cfg_attr(not(feature = "tenancy"), allow(dead_code))]
Tenant { schema: Option<&'a str> },
}
fn in_scope(s: &crate::core::ModelSchema, scope: DataScope<'_>) -> bool {
match scope {
DataScope::All => true,
DataScope::Registry => s.scope == crate::core::ModelScope::Registry,
DataScope::Tenant { .. } => s.scope == crate::core::ModelScope::Tenant,
}
}
async fn require_tenant_tables(
pool: &Pool,
scope: DataScope<'_>,
models: &[&'static crate::core::ModelSchema],
verb: &str,
) -> Result<(), MigrateError> {
let DataScope::Tenant {
schema: Some(schema),
} = scope
else {
return Ok(());
};
let tables = models
.iter()
.filter(|s| s.managed && !s.is_view)
.flat_map(|s| {
std::iter::once(s.table)
.chain(s.m2m.iter().filter(|r| r.auto_create).map(|r| r.through))
});
for table in tables {
if super::inspectdb::list_tables(pool, schema, Some(table))
.await?
.is_empty()
{
return Err(MigrateError::Validation(format!(
"{verb}: table `{table}` is not in tenant schema `{schema}`; migrate the tenant first"
)));
}
}
Ok(())
}
fn out_of_scope(name: &str, owner: crate::core::ModelScope) -> String {
let hint = match owner {
crate::core::ModelScope::Registry => "a registry model: run without `--tenant`",
_ => "a tenant model: pass `--tenant <slug>`",
};
format!("`{name}` is {hint}")
}
async fn is_sole_pk(pool: &Pool, table: &str, column: &str) -> Result<bool, MigrateError> {
let schema = super::ensure::creation_schema(pool)
.await?
.unwrap_or_default();
let pk = super::inspectdb::list_pk_columns(pool, &schema, table).await?;
Ok(matches!(pk.as_slice(), [c] if c == column))
}
struct FixtureWriter<'w, W: Write> {
w: std::io::BufWriter<&'w mut W>,
pretty: bool,
started: bool,
}
impl<'w, W: Write> FixtureWriter<'w, W> {
fn new(w: &'w mut W, indent: usize) -> Self {
Self {
w: std::io::BufWriter::new(w),
pretty: indent != 0,
started: false,
}
}
fn entry(&mut self, v: &serde_json::Value) -> Result<(), MigrateError> {
let json = |e| MigrateError::Validation(format!("dumpdata: serialize JSON: {e}"));
let sep = match (self.started, self.pretty) {
(false, false) => "[",
(false, true) => "[\n",
(true, false) => ",",
(true, true) => ",\n",
};
self.started = true;
self.w.write_all(sep.as_bytes())?;
if self.pretty {
let text = serde_json::to_string_pretty(v).map_err(json)?;
let lines: Vec<String> = text.lines().map(|l| format!(" {l}")).collect();
self.w.write_all(lines.join("\n").as_bytes())?;
} else {
serde_json::to_writer(&mut self.w, v).map_err(json)?;
}
Ok(())
}
fn finish(mut self) -> Result<(), MigrateError> {
let end = match (self.started, self.pretty) {
(false, _) => "[]",
(true, false) => "]",
(true, true) => "\n]",
};
writeln!(self.w, "{end}")?;
self.w.flush()?;
Ok(())
}
}
fn dotted_model_name(entry: &crate::core::ModelEntry) -> String {
match entry.resolved_app_label() {
Some(app) if !app.is_empty() => format!("{app}.{}", entry.schema.name),
_ => entry.schema.name.to_owned(),
}
}
fn model_matches(filter: &str, dotted: &str, schema: &crate::core::ModelSchema) -> bool {
filter == dotted || filter == schema.name
}
type LinksBySource = std::collections::HashMap<String, Vec<serde_json::Value>>;
type Junction = (
&'static crate::core::ModelSchema,
[&'static crate::core::FieldSchema; 2],
);
fn junction_of(
src: &'static crate::core::ModelSchema,
rel: &crate::core::M2MRelation,
) -> Result<Junction, MigrateError> {
let pk_ty = |schema: Option<&'static crate::core::ModelSchema>, table: &str| {
schema.and_then(crate::sql::m2m::end_type).ok_or_else(|| {
MigrateError::Validation(format!(
"M2M `{}`: `{table}` has no registered primary key",
rel.name
))
})
};
let dst = crate::core::ModelEntry::for_table(rel.to).map(|e| e.schema);
let junction = crate::sql::m2m::typed_junction(
rel.through,
&[
(rel.src_col, pk_ty(Some(src), src.table)?),
(rel.dst_col, pk_ty(dst, rel.to)?),
],
);
let end = |col: &str| {
junction.field_by_column(col).ok_or_else(|| {
MigrateError::Validation(format!("junction `{}` has no `{col}` column", rel.through))
})
};
Ok((junction, [end(rel.src_col)?, end(rel.dst_col)?]))
}
async fn m2m_links(
pool: &Pool,
schema: &'static crate::core::ModelSchema,
src_pks: &[crate::core::SqlValue],
) -> Result<Vec<(&'static str, LinksBySource)>, MigrateError> {
let mut out = Vec::new();
if src_pks.is_empty() {
return Ok(out);
}
for rel in schema.m2m.iter().filter(|r| r.auto_create) {
let (junction, [src, dst]) = junction_of(schema, rel)?;
let query = crate::core::SelectQuery::by_pk_in(junction, rel.src_col, src_pks.to_vec());
let rows = crate::sql::select_rows_as_json(pool, &query, &[src, dst])
.await
.map_err(|e| {
MigrateError::Validation(format!(
"dumpdata: select from `{}` failed: {e}",
rel.through
))
})?;
let mut by_src = LinksBySource::new();
for row in rows {
if let (Some(s), Some(d)) = (row.get(src.name), row.get(dst.name)) {
by_src.entry(s.to_string()).or_default().push(d.clone());
}
}
for dsts in by_src.values_mut() {
dsts.sort_by(|a, b| (a.as_i64(), a.as_str()).cmp(&(b.as_i64(), b.as_str())));
}
out.push((rel.name, by_src));
}
Ok(out)
}
type RowLinks = (usize, serde_json::Value, Vec<serde_json::Value>);
async fn load_links(
pool: &Pool,
schema: &'static crate::core::ModelSchema,
rel: &crate::core::M2MRelation,
rows: &[RowLinks],
fail_fast: bool,
) -> Result<Vec<String>, String> {
let (junction, [s, d]) = junction_of(schema, rel).map_err(|e| e.to_string())?;
let insert = |values| async move {
let query =
crate::core::BulkInsertQuery::new(junction, vec![rel.src_col, rel.dst_col], values)
.on_conflict(crate::core::ConflictClause::DoNothing);
crate::sql::bulk_insert_pool(pool, &query).await
};
let mut failed = Vec::new();
let mut converted = Vec::new();
for (line, src, dsts) in rows {
let row: Result<Vec<_>, String> = json_to_sql_value(src, s).and_then(|src| {
dsts.iter()
.map(|dst| Ok(vec![src.clone(), json_to_sql_value(dst, d)?]))
.collect()
});
match row {
Ok(values) => converted.push((line, values)),
Err(e) => failed.push(format!("entry #{line}: {e}")),
}
if fail_fast && !failed.is_empty() {
return Ok(failed);
}
}
let batch: Vec<_> = converted.iter().flat_map(|(_, v)| v.clone()).collect();
if batch.is_empty() || insert(batch).await.is_ok() {
return Ok(failed);
}
for (line, values) in converted {
if let Err(e) = insert(values).await {
failed.push(format!("entry #{line}: {e}"));
if fail_fast {
break;
}
}
}
Ok(failed)
}
#[derive(Debug, PartialEq)]
struct LoaddataArgs {
file: String,
fail_fast: bool,
help: bool,
}
fn parse_loaddata_args(args: &[String]) -> Result<LoaddataArgs, MigrateError> {
let mut file: Option<String> = None;
let mut fail_fast = false;
let mut help = false;
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
help = true;
break;
}
"--fail-fast" => fail_fast = true,
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
if file.is_some() {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
file = Some(other.to_owned());
}
}
}
if help {
return Ok(LoaddataArgs {
file: String::new(),
fail_fast: false,
help: true,
});
}
let file = file.ok_or_else(|| {
MigrateError::Validation("loaddata: missing <fixture.json> argument".into())
})?;
Ok(LoaddataArgs {
file,
fail_fast,
help: false,
})
}
pub(crate) async fn loaddata_cmd<W: Write>(
pool: &Pool,
args: &[String],
scope: DataScope<'_>,
w: &mut W,
) -> Result<(), MigrateError> {
let parsed = parse_loaddata_args(args)?;
if parsed.help {
writeln!(w, "loaddata <fixture.json> [--fail-fast]")?;
writeln!(w)?;
writeln!(
w,
" Insert every row in `fixture.json` via the registered model schemas."
)?;
writeln!(
w,
" Fixture shape: `[{{\"model\": \"app.Model\", \"pk\": N, \"fields\": {{...}}}}]`"
)?;
writeln!(w, " — the same shape `manage dumpdata` produces.")?;
writeln!(w)?;
writeln!(
w,
" --fail-fast Abort on the first error instead of skipping the row."
)?;
writeln!(
w,
" A failed or partial load is not rolled back: rows already inserted stay."
)?;
return Ok(());
}
let raw = std::fs::read_to_string(&parsed.file)
.map_err(|e| MigrateError::Validation(format!("loaddata: read `{}`: {e}", parsed.file)))?;
let entries: Vec<serde_json::Value> = serde_json::from_str(&raw).map_err(|e| {
MigrateError::Validation(format!("loaddata: parse `{}` as JSON: {e}", parsed.file))
})?;
let mut schemas: std::collections::HashMap<String, &'static crate::core::ModelSchema> =
std::collections::HashMap::new();
let mut elsewhere = std::collections::HashMap::new();
for entry in inventory::iter::<crate::core::ModelEntry>() {
let schema = entry.schema;
let app = entry.resolved_app_label().unwrap_or("");
let names = std::iter::once(schema.name.to_owned())
.chain((!app.is_empty()).then(|| format!("{app}.{}", schema.name)));
if in_scope(schema, scope) {
schemas.extend(names.map(|n| (n, schema)));
} else {
elsewhere.extend(names.map(|n| (n, schema.scope)));
}
}
let LoadPlan { order, deferred } = load_plan(&entries, &schemas);
let entries = in_fk_order(entries, &schemas, &order);
let mut named: Vec<&'static crate::core::ModelSchema> = Vec::new();
for s in entries
.iter()
.filter_map(|e| schemas.get(e.get("model").and_then(|v| v.as_str()).unwrap_or("")))
{
if !named.iter().any(|n| n.table == s.table) {
named.push(s);
}
}
require_tenant_tables(pool, scope, &named, "loaddata").await?;
let mut loaded = 0_usize;
let mut skipped = 0_usize;
let mut touched: Vec<&'static crate::core::ModelSchema> = Vec::new();
let mut abort: Option<String> = None;
let mut links: Vec<(
&'static crate::core::ModelSchema,
&'static crate::core::M2MRelation,
String,
Vec<RowLinks>,
)> = Vec::new();
let mut relinks: Vec<(String, crate::core::UpdateQuery)> = Vec::new();
for (idx, entry) in entries.into_iter().enumerate() {
let line = idx + 1;
let model_name = entry.get("model").and_then(|v| v.as_str()).unwrap_or("");
let schema = match schemas.get(model_name) {
Some(s) => *s,
None => {
let msg = match elsewhere.get(model_name) {
Some(owner) => format!(
"loaddata: entry #{line}: {}",
out_of_scope(model_name, *owner)
),
None => format!(
"loaddata: entry #{line}: unknown model `{model_name}` (registered: {} models)",
schemas.len() / 2, ),
};
if parsed.fail_fast {
abort = Some(msg);
break;
}
tracing::warn!("{msg}");
skipped += 1;
continue;
}
};
let mut fields_obj: serde_json::Map<String, serde_json::Value> = entry
.get("fields")
.and_then(|v| v.as_object())
.cloned()
.unwrap_or_default();
if let Some(pk_value) = entry.get("pk").cloned() {
if let Some(pk_field) = schema.primary_key() {
fields_obj.insert(pk_field.name.to_owned(), pk_value);
}
}
let mut columns: Vec<&'static str> = Vec::new();
let mut values: Vec<crate::core::SqlValue> = Vec::new();
let mut row_err: Option<String> = None;
for f in schema.scalar_fields() {
let Some(raw) = fields_obj
.get(f.name)
.filter(|v| !(f.primary_key && v.is_null()))
else {
continue;
};
match loaddata_value(raw, f) {
Ok(v) => {
columns.push(f.column);
values.push(v);
}
Err(e) => {
row_err = Some(format!("field `{}`: {e}", f.name));
break;
}
}
}
let mut row_links = Vec::new();
for rel in schema.m2m.iter().filter(|r| r.auto_create) {
match fields_obj.get(rel.name) {
None | Some(serde_json::Value::Null) => {}
Some(serde_json::Value::Array(dsts)) if !dsts.is_empty() => {
row_links.push((rel, dsts.clone()));
}
Some(serde_json::Value::Array(_)) => {}
Some(_) => row_err = Some(format!("field `{}`: expected a list of pks", rel.name)),
}
}
let src_pk = entry.get("pk").cloned().unwrap_or_default();
if !row_links.is_empty() && src_pk.is_null() {
row_err = Some("M2M links need the row's `pk`".to_owned());
}
if let Some(e) = row_err {
let msg = format!("loaddata: entry #{line} (`{model_name}`): {e}");
if parsed.fail_fast {
abort = Some(msg);
break;
}
tracing::warn!("{msg}");
skipped += 1;
continue;
}
let pk_field = schema.primary_key();
let relink = match pk_field {
Some(_) => take_deferred(schema, &columns, &mut values, &deferred),
None => Vec::new(),
};
let returning = match pk_field {
Some(pk) if !relink.is_empty() && !columns.contains(&pk.column) => vec![pk.column],
_ => vec![],
};
if let Some(pk) = pk_field.filter(|pk| {
pool.dialect().name() == "mysql" && !returning.is_empty() && !pk.is_serial()
}) {
let msg = format!(
"loaddata: entry #{line} (`{model_name}`): MySQL cannot return the \
database-generated `{}`, so this row's cycle link can't be set; give it a `pk`",
pk.name
);
if parsed.fail_fast {
abort = Some(msg);
break;
}
tracing::warn!("{msg}");
skipped += 1;
continue;
}
let query = crate::core::InsertQuery {
model: schema,
columns,
values,
returning,
on_conflict: None,
generated: vec![],
};
let inserted = match pk_field.filter(|_| !relink.is_empty()) {
None => crate::sql::insert_pool(pool, &query).await.map(|()| None),
Some(pk) => {
async {
let returned = if query.returning.is_empty() {
crate::sql::insert_pool(pool, &query).await?;
None
} else {
Some(crate::sql::insert_returning_pool(pool, &query).await?)
};
let value = match &returned {
Some(r) => crate::sql::inserted_pk(&query, r, pk)?,
None => {
let i = query.columns.iter().position(|c| *c == pk.column);
i.map(|i| query.values[i].clone())
.unwrap_or(crate::core::SqlValue::Null)
}
};
Ok(Some(crate::core::UpdateQuery {
model: schema,
set: relink,
where_clause: crate::core::WhereExpr::Predicate(crate::core::Filter {
column: pk.column,
op: crate::core::Op::Eq,
value,
}),
}))
}
.await
}
};
match inserted {
Ok(relink) => {
loaded += 1;
if let Some(update) = relink {
relinks.push((format!("entry #{line} (`{model_name}`)"), update));
}
if !touched.iter().any(|t| t.table == schema.table) {
touched.push(schema);
}
for (rel, dsts) in row_links {
let row = (line, src_pk.clone(), dsts);
match links
.iter_mut()
.find(|(s, r, _, _)| s.table == schema.table && r.name == rel.name)
{
Some((_, _, _, rows)) => rows.push(row),
None => links.push((schema, rel, model_name.to_owned(), vec![row])),
}
}
}
Err(e) => {
let msg = format!("loaddata: entry #{line} (`{model_name}`) insert: {e}");
if parsed.fail_fast {
abort = Some(msg);
break;
}
tracing::warn!("{msg}");
skipped += 1;
}
}
}
for (what, update) in relinks {
let res = match crate::sql::update_pool(pool, &update).await {
Ok(0) => Err("no row has the inserted pk".to_owned()),
r => r.map(drop).map_err(|e| e.to_string()),
};
if let Err(e) = res {
let msg = format!("loaddata: {what} link: {e}");
tracing::warn!("{msg}");
loaded -= 1;
skipped += 1;
if parsed.fail_fast && abort.is_none() {
abort = Some(msg);
}
}
}
if abort.is_none() {
'rels: for (schema, rel, model_name, rows) in links {
let (failed, lost) = match load_links(pool, schema, rel, &rows, parsed.fail_fast).await
{
Ok(failed) => {
let n = failed.len();
(failed, n)
}
Err(e) => (vec![e], rows.len()),
};
for e in failed {
let msg = format!("loaddata: `{model_name}` `{}`: {e}", rel.name);
if parsed.fail_fast {
abort = Some(msg);
break 'rels;
}
tracing::warn!("{msg}");
}
skipped += lost;
}
}
reset_sequences(pool, &touched).await?;
if let Some(msg) = abort {
return Err(MigrateError::Validation(msg));
}
writeln!(w, "loaddata: {loaded} loaded, {skipped} skipped")?;
if skipped > 0 {
return Err(MigrateError::Validation(format!(
"loaddata: {skipped} row(s) skipped — see the warnings above"
)));
}
Ok(())
}
fn dumpable(ty: crate::core::FieldType) -> bool {
use crate::core::FieldType as T;
!matches!(
ty,
T::Array(_) | T::Range(_) | T::HStore | T::Vector(_) | T::Geometry(_)
)
}
struct LoadPlan {
order: Vec<&'static crate::core::ModelSchema>,
deferred: DeferredLinks,
}
fn load_plan(
entries: &[serde_json::Value],
schemas: &std::collections::HashMap<String, &'static crate::core::ModelSchema>,
) -> LoadPlan {
let schema_of = |e: &serde_json::Value| {
schemas
.get(e.get("model").and_then(|v| v.as_str()).unwrap_or(""))
.copied()
};
let mut nodes: Vec<&'static crate::core::ModelSchema> = Vec::new();
for s in entries.iter().filter_map(schema_of) {
if !nodes.iter().any(|n| n.table == s.table) {
nodes.push(s);
}
}
let mut set_by_a_row: std::collections::HashSet<(&str, &str)> =
std::collections::HashSet::new();
for e in entries {
let (Some(s), Some(fields)) = (schema_of(e), e.get("fields")) else {
continue;
};
set_by_a_row.extend(
s.fields
.iter()
.filter(|f| fields.get(f.name).is_some_and(|v| !v.is_null()))
.map(|f| (s.table, f.column)),
);
}
let filled = |s: &crate::core::ModelSchema, col: &str| {
!s.fields.iter().any(|f| f.column == col) || set_by_a_row.contains(&(s.table, col))
};
let used =
|s: &crate::core::ModelSchema, col: Option<&'static str>| col.is_none_or(|c| filled(s, c));
let (mut children_first_order, cyclic) = children_first(&nodes, used);
let (rest, stuck) = children_first(&cyclic, |_, col| col.is_none());
children_first_order.extend(rest);
children_first_order.extend(stuck);
let order: Vec<_> = children_first_order.into_iter().rev().collect();
let at = |t: &str| order.iter().position(|s| s.table == t);
let in_cycle = |t: &str| cyclic.iter().any(|c| c.table == t);
let deferred = cyclic
.iter()
.flat_map(|s| {
fk_edges(s)
.filter(|(to, col)| {
*to != s.table && in_cycle(to) && used(s, *col) && at(to) > at(s.table)
})
.filter_map(|(_, col)| col.map(|c| (s.table, c)))
})
.collect();
LoadPlan { order, deferred }
}
type DeferredLinks = std::collections::HashSet<(&'static str, &'static str)>;
fn take_deferred(
schema: &'static crate::core::ModelSchema,
columns: &[&'static str],
values: &mut [crate::core::SqlValue],
deferred: &DeferredLinks,
) -> Vec<crate::core::Assignment> {
use crate::core::{Assignment, SqlValue};
columns
.iter()
.zip(values.iter_mut())
.filter(|(c, v)| deferred.contains(&(schema.table, **c)) && !matches!(v, SqlValue::Null))
.map(|(c, v)| Assignment::new(c, std::mem::replace(v, SqlValue::Null)))
.collect()
}
fn in_fk_order(
entries: Vec<serde_json::Value>,
schemas: &std::collections::HashMap<String, &'static crate::core::ModelSchema>,
order: &[&'static crate::core::ModelSchema],
) -> Vec<serde_json::Value> {
let mut keyed: Vec<(usize, serde_json::Value)> = entries
.into_iter()
.map(|e| {
let rank = e
.get("model")
.and_then(|v| v.as_str())
.and_then(|m| schemas.get(m))
.and_then(|s| order.iter().position(|o| o.table == s.table))
.map_or(0, |i| i + 1);
(rank, e)
})
.collect();
keyed.sort_by_key(|(rank, _)| *rank);
let mut out = Vec::with_capacity(keyed.len());
let mut group: Vec<serde_json::Value> = Vec::new();
let mut current = None;
for (rank, e) in keyed {
if current != Some(rank) {
out.extend(self_parents_first(std::mem::take(&mut group), schemas));
current = Some(rank);
}
group.push(e);
}
out.extend(self_parents_first(group, schemas));
out
}
fn self_parents_first(
entries: Vec<serde_json::Value>,
schemas: &std::collections::HashMap<String, &'static crate::core::ModelSchema>,
) -> Vec<serde_json::Value> {
struct Pending {
entry: serde_json::Value,
key: Option<(&'static str, String)>,
parents: Vec<(&'static str, String)>,
}
let mut pending: Vec<Pending> = entries
.into_iter()
.map(|entry| {
let schema = entry
.get("model")
.and_then(|v| v.as_str())
.and_then(|m| schemas.get(m));
let Some(schema) = schema else {
return Pending {
entry,
key: None,
parents: Vec::new(),
};
};
let key = entry
.get("pk")
.filter(|v| !v.is_null())
.map(|pk| (schema.table, pk.to_string()));
let parents = schema
.scalar_fields()
.filter(|f| {
matches!(
f.relation,
Some(crate::core::Relation::Fk { to, .. }
| crate::core::Relation::O2O { to, .. }) if to == schema.table
)
})
.filter_map(|f| entry.get("fields")?.get(f.name))
.filter(|v| !v.is_null())
.map(|v| (schema.table, v.to_string()))
.filter(|p| Some(p) != key.as_ref())
.collect();
Pending {
entry,
key,
parents,
}
})
.collect();
let mut out = Vec::with_capacity(pending.len());
while !pending.is_empty() {
let waiting: std::collections::HashSet<(&'static str, String)> =
pending.iter().filter_map(|p| p.key.clone()).collect();
let (ready, rest): (Vec<Pending>, Vec<Pending>) = pending
.into_iter()
.partition(|p| p.parents.iter().all(|k| !waiting.contains(k)));
if ready.is_empty() {
out.extend(rest.into_iter().map(|p| p.entry));
break;
}
out.extend(ready.into_iter().map(|p| p.entry));
pending = rest;
}
out
}
async fn reset_sequences(
pool: &Pool,
schemas: &[&'static crate::core::ModelSchema],
) -> Result<(), MigrateError> {
for schema in schemas {
let Some((sql, binds)) = reset_sequence_stmt(pool.dialect(), schema) else {
continue;
};
crate::sql::raw_execute_pool(pool, &sql, binds)
.await
.map_err(|e| {
MigrateError::Validation(format!(
"loaddata: resetting `{}`'s id sequence failed: {e}",
schema.table
))
})?;
}
Ok(())
}
pub(crate) fn reset_sequence_stmt(
dialect: &dyn crate::sql::Dialect,
schema: &crate::core::ModelSchema,
) -> Option<(String, Vec<crate::core::SqlValue>)> {
use crate::core::FieldType as T;
let pk = schema.primary_key()?;
if !pk.auto || !matches!(pk.ty, T::I16 | T::I32 | T::I64) {
return None;
}
let sql = dialect.reset_sequence_sql(schema.table, pk.column)?;
let binds = vec![
crate::core::SqlValue::String(dialect.quote_ident(schema.table)),
crate::core::SqlValue::String(pk.column.to_owned()),
];
Some((sql, binds))
}
pub(crate) fn json_to_sql_value(
v: &serde_json::Value,
field: &crate::core::FieldSchema,
) -> Result<crate::core::SqlValue, String> {
use crate::core::{FieldType, SqlValue};
if v.is_null() {
return Ok(SqlValue::Null);
}
match field.ty {
FieldType::I16 => v
.as_i64()
.and_then(|n| i16::try_from(n).ok())
.map(SqlValue::I16)
.ok_or_else(|| format!("expected i16, got {v}")),
FieldType::I32 => v
.as_i64()
.and_then(|n| i32::try_from(n).ok())
.map(SqlValue::I32)
.ok_or_else(|| format!("expected i32, got {v}")),
FieldType::I64 => v
.as_i64()
.or_else(|| v.as_str().and_then(|s| s.parse::<i64>().ok()))
.map(SqlValue::I64)
.ok_or_else(|| format!("expected i64, got {v}")),
FieldType::F32 => v
.as_f64()
.map(|n| SqlValue::F32(n as f32))
.ok_or_else(|| format!("expected f32, got {v}")),
FieldType::F64 => v
.as_f64()
.map(SqlValue::F64)
.ok_or_else(|| format!("expected f64, got {v}")),
FieldType::Bool => v
.as_bool()
.map(SqlValue::Bool)
.ok_or_else(|| format!("expected bool, got {v}")),
FieldType::String => v
.as_str()
.map(|s| SqlValue::String(s.to_owned()))
.ok_or_else(|| format!("expected string, got {v}")),
FieldType::DateTime => {
let s = v
.as_str()
.ok_or_else(|| format!("expected string for DateTime, got {v}"))?;
chrono::DateTime::parse_from_rfc3339(s)
.map(|d| SqlValue::DateTime(d.with_timezone(&chrono::Utc)))
.or_else(|_| {
chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S")
.map(|ndt| SqlValue::DateTime(ndt.and_utc()))
})
.map_err(|e| format!("DateTime parse: {e}"))
}
FieldType::Date => {
let s = v
.as_str()
.ok_or_else(|| format!("expected string for Date, got {v}"))?;
chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d")
.map(SqlValue::Date)
.map_err(|e| format!("Date parse: {e}"))
}
FieldType::Time => {
let s = v
.as_str()
.ok_or_else(|| format!("expected string for Time, got {v}"))?;
chrono::NaiveTime::parse_from_str(s, "%H:%M:%S%.f")
.or_else(|_| chrono::NaiveTime::parse_from_str(s, "%H:%M"))
.map(SqlValue::Time)
.map_err(|e| format!("Time parse: {e}"))
}
FieldType::Uuid => {
let s = v
.as_str()
.ok_or_else(|| format!("expected string for Uuid, got {v}"))?;
uuid::Uuid::parse_str(s)
.map(SqlValue::Uuid)
.map_err(|e| format!("Uuid parse: {e}"))
}
FieldType::Json => Ok(SqlValue::Json(v.clone())),
FieldType::Decimal => {
let s = v
.as_str()
.map(str::to_owned)
.or_else(|| v.as_f64().map(|n| n.to_string()))
.or_else(|| v.as_i64().map(|n| n.to_string()))
.ok_or_else(|| format!("expected number/string for Decimal, got {v}"))?;
s.parse::<rust_decimal::Decimal>()
.map(SqlValue::Decimal)
.map_err(|e| format!("Decimal parse: {e}"))
}
FieldType::Binary => {
let s = v
.as_str()
.ok_or_else(|| format!("expected lowercase-hex string for Binary, got {v}"))?;
if s.len() % 2 != 0 || !s.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err("Binary: not a valid lowercase-hex string".into());
}
let bytes: Vec<u8> = s
.as_bytes()
.chunks_exact(2)
.map(|c| {
let h = (c[0] as char).to_digit(16).unwrap_or(0) as u8;
let l = (c[1] as char).to_digit(16).unwrap_or(0) as u8;
(h << 4) | l
})
.collect();
Ok(SqlValue::Binary(bytes))
}
FieldType::Array(elem) => {
let arr = v
.as_array()
.ok_or_else(|| format!("expected JSON array for Array column, got {v}"))?;
let items = arr
.iter()
.map(|el| match elem {
crate::core::ArrayElem::Text => el
.as_str()
.map(|s| SqlValue::String(s.to_owned()))
.ok_or_else(|| format!("expected string array element, got {el}")),
crate::core::ArrayElem::Int => el
.as_i64()
.and_then(|n| i32::try_from(n).ok())
.map(SqlValue::I32)
.ok_or_else(|| format!("expected i32 array element, got {el}")),
crate::core::ArrayElem::BigInt => el
.as_i64()
.map(SqlValue::I64)
.ok_or_else(|| format!("expected i64 array element, got {el}")),
})
.collect::<Result<Vec<_>, _>>()?;
Ok(SqlValue::Array(items))
}
FieldType::Range(_) => v
.as_str()
.map(|s| SqlValue::RangeLiteral(s.to_owned()))
.ok_or_else(|| format!("expected range-literal string for Range column, got {v}")),
FieldType::HStore => {
let obj = v
.as_object()
.ok_or_else(|| format!("expected JSON object for HStore column, got {v}"))?;
let pairs = obj
.iter()
.map(|(k, val)| match val {
serde_json::Value::String(s) => Ok((k.clone(), Some(s.clone()))),
serde_json::Value::Null => Ok((k.clone(), None)),
other => Err(format!("hstore value must be string or null, got {other}")),
})
.collect::<Result<Vec<_>, _>>()?;
Ok(SqlValue::HStore(pairs))
}
FieldType::Vector(_) => {
let arr = v
.as_array()
.ok_or_else(|| format!("expected JSON array for Vector column, got {v}"))?;
let items = arr
.iter()
.map(|el| {
el.as_f64()
.map(|n| n as f32)
.ok_or_else(|| format!("expected numeric vector element, got {el}"))
})
.collect::<Result<Vec<_>, _>>()?;
Ok(SqlValue::Vector(items))
}
FieldType::Geometry(_) => {
let p: crate::sql::Point = serde_json::from_value(v.clone())
.map_err(|e| format!("expected JSON {{x, y, srid?}} for geometry column: {e}"))?;
Ok(p.into())
}
}
}
fn showurls_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let mut format = "plain";
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
writeln!(w, "showurls [--format <plain|json>]")?;
writeln!(w)?;
writeln!(
w,
" Print every named URL pattern registered via `register_url!`."
)?;
writeln!(w, " --format plain (default) | json")?;
return Ok(());
}
"--format" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--format expects a value".into()))?;
match v.as_str() {
"plain" | "json" => {
format = match v.as_str() {
"plain" => "plain",
"json" => "json",
_ => unreachable!(),
}
}
other => {
return Err(MigrateError::Validation(format!(
"--format: unknown value `{other}` (expected `plain` or `json`)"
)));
}
}
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
let mut routes: Vec<&'static crate::urls::NamedRoute> =
inventory::iter::<crate::urls::NamedRoute>().collect();
routes.sort_by_key(|r| r.name);
match format {
"json" => {
let items: Vec<serde_json::Value> = routes
.iter()
.map(|r| {
serde_json::json!({
"name": r.name,
"pattern": r.pattern,
})
})
.collect();
let rendered = serde_json::to_string_pretty(&items)
.map_err(|e| MigrateError::Validation(format!("showurls: serialize JSON: {e}")))?;
writeln!(w, "{rendered}")?;
}
_ => {
if routes.is_empty() {
writeln!(w, "(no named URLs registered)")?;
return Ok(());
}
let max_name = routes.iter().map(|r| r.name.len()).max().unwrap_or(0);
for r in &routes {
writeln!(w, " {:<width$} {}", r.name, r.pattern, width = max_name)?;
}
}
}
Ok(())
}
fn showmodels_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let mut format = "plain";
let mut app_filter: Option<String> = None;
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
writeln!(w, "showmodels [--format plain|json] [--app <label>]")?;
writeln!(w)?;
writeln!(
w,
" Print every model registered via #[derive(Model)] + inventory."
)?;
writeln!(w, " --format plain (default) | json")?;
writeln!(w, " --app <label> Limit to a single app.")?;
return Ok(());
}
"--format" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--format expects a value".into()))?;
match v.as_str() {
"plain" => format = "plain",
"json" => format = "json",
other => {
return Err(MigrateError::Validation(format!(
"--format: unknown value `{other}` (expected `plain` or `json`)"
)));
}
}
}
"--app" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--app expects a value".into()))?;
app_filter = Some(v.clone());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
#[derive(Debug)]
struct Row {
app: String,
name: &'static str,
table: &'static str,
fields: usize,
}
let mut rows: Vec<Row> = inventory::iter::<crate::core::ModelEntry>()
.map(|entry| Row {
app: entry.resolved_app_label().unwrap_or("").to_owned(),
name: entry.schema.name,
table: entry.schema.table,
fields: entry.schema.fields.len(),
})
.filter(|r| app_filter.as_deref().is_none_or(|f| r.app == f))
.collect();
rows.sort_by(|a, b| (a.app.as_str(), a.name).cmp(&(b.app.as_str(), b.name)));
match format {
"json" => {
let items: Vec<serde_json::Value> = rows
.iter()
.map(|r| {
serde_json::json!({
"app": r.app,
"name": r.name,
"table": r.table,
"fields": r.fields,
})
})
.collect();
let rendered = serde_json::to_string_pretty(&items).map_err(|e| {
MigrateError::Validation(format!("showmodels: serialize JSON: {e}"))
})?;
writeln!(w, "{rendered}")?;
}
_ => {
if rows.is_empty() {
let suffix = app_filter
.as_deref()
.map(|f| format!(" for app `{f}`"))
.unwrap_or_default();
writeln!(w, "(no models registered{suffix})")?;
return Ok(());
}
let max_app = rows.iter().map(|r| r.app.len()).max().unwrap_or(0).max(3); let max_name = rows.iter().map(|r| r.name.len()).max().unwrap_or(0).max(5);
let max_table = rows.iter().map(|r| r.table.len()).max().unwrap_or(0).max(5);
for r in &rows {
let app = if r.app.is_empty() {
"-"
} else {
r.app.as_str()
};
writeln!(
w,
" {:<aw$} {:<nw$} {:<tw$} {} fields",
app,
r.name,
r.table,
r.fields,
aw = max_app,
nw = max_name,
tw = max_table,
)?;
}
}
}
Ok(())
}
#[derive(Debug, Default, PartialEq)]
struct FlushArgs {
yes: bool,
apps: Vec<String>,
models: Vec<String>,
help: bool,
}
fn parse_flush_args(args: &[String]) -> Result<FlushArgs, MigrateError> {
let mut out = FlushArgs::default();
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
out.help = true;
return Ok(out);
}
"--yes" => out.yes = true,
"--app" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--app expects a value".into()))?;
out.apps.push(v.clone());
}
"--model" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--model expects a value".into()))?;
out.models.push(v.clone());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
Ok(out)
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum FlushScope<'a> {
All,
#[cfg_attr(not(feature = "tenancy"), allow(dead_code))]
Tenant { schema: Option<&'a str> },
}
fn pg_error_detail(e: &crate::sql::ExecError) -> Option<String> {
#[cfg(feature = "postgres")]
if let crate::sql::ExecError::Driver(sqlx::Error::Database(db)) = e {
return db
.try_downcast_ref::<sqlx::postgres::PgDatabaseError>()
.and_then(|pg| pg.detail())
.map(str::to_owned);
}
let _ = e;
None
}
fn flush_eligible(s: &crate::core::ModelSchema, scope: FlushScope<'_>) -> bool {
s.managed
&& !s.is_view
&& (matches!(scope, FlushScope::All) || s.scope == crate::core::ModelScope::Tenant)
}
fn flush_junctions(
targets: &[&'static crate::core::ModelSchema],
scope: FlushScope<'_>,
) -> Vec<&'static crate::core::ModelSchema> {
let is_target = |t: &str| targets.iter().any(|s| s.table == t);
let mut out: Vec<&'static crate::core::ModelSchema> = Vec::new();
for entry in inventory::iter::<crate::core::ModelEntry>() {
let s = entry.schema;
if !flush_eligible(s, scope) {
continue;
}
for m in s.m2m {
if m.auto_create
&& (is_target(s.table) || is_target(m.to))
&& !is_target(m.through)
&& !out.iter().any(|j| j.table == m.through)
{
out.push(crate::sql::m2m::junction(
m.through,
&[m.src_col, m.dst_col],
));
}
}
}
out
}
fn fk_edges(
s: &crate::core::ModelSchema,
) -> impl Iterator<Item = (&'static str, Option<&'static str>)> + '_ {
use crate::core::Relation;
let single = s.fields.iter().filter_map(|f| match f.relation {
Some(Relation::Fk { to, .. } | Relation::O2O { to, .. }) => {
Some((to, f.nullable.then_some(f.column)))
}
_ => None,
});
single.chain(s.composite_relations.iter().map(|c| (c.to, None)))
}
fn children_first(
nodes: &[&'static crate::core::ModelSchema],
binds: impl Fn(&crate::core::ModelSchema, Option<&'static str>) -> bool,
) -> (
Vec<&'static crate::core::ModelSchema>,
Vec<&'static crate::core::ModelSchema>,
) {
let links = |from: &crate::core::ModelSchema, to: &str| {
from.table != to && fk_edges(from).any(|(t, col)| t == to && binds(from, col))
};
let mut done = vec![false; nodes.len()];
let mut order = Vec::with_capacity(nodes.len());
while let Some(i) = (0..nodes.len()).find(|&i| {
!done[i] && !(0..nodes.len()).any(|j| !done[j] && links(nodes[j], nodes[i].table))
}) {
done[i] = true;
order.push(nodes[i]);
}
let stuck = (0..nodes.len())
.filter(|&i| !done[i])
.map(|i| nodes[i])
.collect();
(order, stuck)
}
struct FlushPlan {
unlink: Vec<(&'static crate::core::ModelSchema, Vec<&'static str>)>,
order: Vec<&'static crate::core::ModelSchema>,
stuck: Vec<&'static str>,
}
fn flush_plan(targets: &[&'static crate::core::ModelSchema]) -> FlushPlan {
let (mut order, cyclic) = children_first(targets, |_, _| true);
let (rest, stuck) = children_first(&cyclic, |_, col| col.is_none());
order.extend(rest);
order.extend(&stuck);
let in_cycle = |t: &str| cyclic.iter().any(|c| c.table == t);
let unlink = targets
.iter()
.map(|s| {
let cols: Vec<&'static str> = fk_edges(s)
.filter(|(to, _)| *to == s.table || (in_cycle(s.table) && in_cycle(to)))
.filter_map(|(_, col)| col)
.collect();
(*s, cols)
})
.filter(|(_, cols)| !cols.is_empty())
.collect();
FlushPlan {
unlink,
order,
stuck: stuck.iter().map(|s| s.table).collect(),
}
}
async fn delete_all_in_tx(
pool: &Pool,
junctions: &[&'static crate::core::ModelSchema],
plan: &FlushPlan,
) -> Result<(), (String, String)> {
use crate::core::{Assignment, DeleteQuery, SqlValue, UpdateQuery, WhereExpr};
let defer = pool.dialect().defer_foreign_keys_sql();
if defer.is_none() && !plan.stuck.is_empty() {
let tables = plan.stuck.join(", ");
let why = "they reference each other through NOT NULL foreign keys; no delete order exists";
return Err((tables, why.to_owned()));
}
let mut tx = crate::sql::write_transaction(pool)
.await
.map_err(|e| ("BEGIN".to_owned(), e.to_string()))?;
if let Some(sql) = defer {
crate::sql::raw_execute_tx(&mut tx, sql, Vec::new())
.await
.map_err(|e| ("BEGIN".to_owned(), e.to_string()))?;
}
for (schema, cols) in &plan.unlink {
let unlink = UpdateQuery {
model: schema,
set: cols
.iter()
.map(|c| Assignment::new(c, SqlValue::Null))
.collect(),
where_clause: WhereExpr::And(Vec::new()),
};
crate::sql::update_tx(&mut tx, &unlink)
.await
.map_err(|e| (schema.table.to_owned(), e.to_string()))?;
}
for schema in junctions.iter().chain(&plan.order) {
let all = DeleteQuery {
model: schema,
where_clause: WhereExpr::And(Vec::new()),
};
crate::sql::delete_tx(&mut tx, &all)
.await
.map_err(|e| (schema.table.to_owned(), e.to_string()))?;
}
tx.commit()
.await
.map_err(|e| ("COMMIT".to_owned(), e.to_string()))
}
pub(crate) async fn flush_cmd<W: Write>(
pool: &Pool,
args: &[String],
scope: FlushScope<'_>,
w: &mut W,
) -> Result<(), MigrateError> {
let parsed = parse_flush_args(args)?;
if parsed.help {
writeln!(w, "flush [--yes] [--app <label>] [--model <name>]")?;
writeln!(w)?;
writeln!(
w,
" Wipe all rows from registered model tables. Schema + migrations ledger"
)?;
writeln!(w, " stay intact. Unmanaged models and views are skipped.")?;
writeln!(w)?;
writeln!(
w,
" --yes Confirm. Without this, prints the planned action"
)?;
writeln!(w, " and exits without touching the DB.")?;
writeln!(w, " --app <label> Limit to one app (repeatable).")?;
writeln!(w, " --model <name> Limit to one model (repeatable).")?;
writeln!(w)?;
writeln!(
w,
" Postgres runs TRUNCATE … RESTART IDENTITY: ids restart. A table outside"
)?;
writeln!(
w,
" the filter that references a target makes the flush fail; nothing is cleared."
)?;
writeln!(
w,
" MySQL / SQLite delete the rows in one transaction and keep their id counters."
)?;
return Ok(());
}
let mut targets: Vec<&'static crate::core::ModelSchema> = Vec::new();
for entry in inventory::iter::<crate::core::ModelEntry>() {
let schema = entry.schema;
if !flush_eligible(schema, scope) {
continue;
}
let app = entry.resolved_app_label().unwrap_or("");
let dotted = if app.is_empty() {
schema.name.to_owned()
} else {
format!("{app}.{}", schema.name)
};
if !parsed.apps.is_empty() && !parsed.apps.iter().any(|a| a == app) {
continue;
}
if !parsed.models.is_empty()
&& !parsed
.models
.iter()
.any(|m| m == &dotted || m == schema.name)
{
continue;
}
targets.push(schema);
}
if targets.is_empty() {
writeln!(w, "flush: no tables match the filter (nothing to do)")?;
return Ok(());
}
let junctions = flush_junctions(&targets, scope);
if !parsed.yes {
writeln!(
w,
"flush: would clear {} table(s) (run with --yes to execute):",
junctions.len() + targets.len()
)?;
for t in junctions.iter().chain(&targets) {
writeln!(w, " - {}", t.table)?;
}
return Ok(());
}
let dialect = pool.dialect().name();
let mut cleared = 0_usize;
let mut failures: Vec<(String, String)> = Vec::new();
if dialect == "postgres" {
let d = pool.dialect();
let prefix = match scope {
FlushScope::Tenant { schema: Some(s) } => format!("{}.", d.quote_ident(s)),
_ => String::new(),
};
let quoted: Vec<String> = junctions
.iter()
.chain(&targets)
.map(|t| format!("{prefix}{}", d.quote_ident(t.table)))
.collect();
let sql = format!("TRUNCATE TABLE {} RESTART IDENTITY", quoted.join(", "));
match crate::sql::raw_execute_pool(pool, &sql, Vec::new()).await {
Ok(_) => cleared = quoted.len(),
Err(e) => failures.push((
"TRUNCATE".to_owned(),
match pg_error_detail(&e) {
Some(detail) => format!("{e} ({detail})"),
None => format!("{e} (tables: {})", quoted.join(", ")),
},
)),
}
} else {
match delete_all_in_tx(pool, &junctions, &flush_plan(&targets)).await {
Ok(()) => cleared = junctions.len() + targets.len(),
Err(failed) => failures.push(failed),
}
}
writeln!(w, "flush: cleared {cleared} table(s)")?;
if !failures.is_empty() {
writeln!(w, "flush: {} table(s) failed:", failures.len())?;
for (t, e) in &failures {
writeln!(w, " - {t}: {e}")?;
}
let what: Vec<String> = failures.iter().map(|(t, e)| format!("{t}: {e}")).collect();
return Err(MigrateError::Validation(format!(
"flush completed with {} failure(s): {}",
failures.len(),
what.join("; ")
)));
}
Ok(())
}
#[cfg(all(feature = "config", feature = "email"))]
#[derive(Debug, Default, PartialEq, Eq)]
struct SendTestEmailArgs {
to: Option<String>,
from: Option<String>,
subject: Option<String>,
help: bool,
}
#[cfg(all(feature = "config", feature = "email"))]
fn parse_sendtestemail_args(args: &[String]) -> Result<SendTestEmailArgs, MigrateError> {
let mut out = SendTestEmailArgs::default();
let mut iter = args.iter();
while let Some(a) = iter.next() {
match a.as_str() {
"--help" | "-h" => out.help = true,
"--to" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--to requires a value".to_owned()))?;
out.to = Some(v.clone());
}
"--from" => {
let v = iter.next().ok_or_else(|| {
MigrateError::Validation("--from requires a value".to_owned())
})?;
out.from = Some(v.clone());
}
"--subject" => {
let v = iter.next().ok_or_else(|| {
MigrateError::Validation("--subject requires a value".to_owned())
})?;
out.subject = Some(v.clone());
}
other if other.starts_with("--") => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
Ok(out)
}
#[derive(Debug, Default, PartialEq)]
struct PruneArgs {
only: Vec<String>,
except: Vec<String>,
pretend: bool,
help: bool,
}
fn parse_prune_args(args: &[String]) -> Result<PruneArgs, MigrateError> {
let mut out = PruneArgs::default();
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
out.help = true;
return Ok(out);
}
"--pretend" | "--dry-run" => out.pretend = true,
"--model" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--model expects a value".into()))?;
out.only.push(v.clone());
}
"--except" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--except expects a value".into()))?;
out.except.push(v.clone());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
Ok(out)
}
async fn audit_cleanup_cmd<W: Write>(
pool: &Pool,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
const USAGE: &str = "audit-cleanup --days <N> | --keep-last <N> [--yes]";
let (mut days, mut keep_last, mut yes) = (None, None, false);
let mut iter = args.iter();
let number = |flag: &str, v: Option<&String>| -> Result<i64, MigrateError> {
let raw = v.ok_or_else(|| MigrateError::Validation(format!("{flag} requires a value")))?;
raw.parse::<i64>().map_err(|_| {
MigrateError::Validation(format!("{flag} expects an integer, got `{raw}`"))
})
};
while let Some(flag) = iter.next() {
match flag.as_str() {
"--days" => days = Some(number("--days", iter.next())?),
"--keep-last" => keep_last = Some(number("--keep-last", iter.next())?),
"--yes" => yes = true,
"--help" | "-h" => {
writeln!(w, "{USAGE}")?;
return Ok(());
}
other => {
return Err(MigrateError::Validation(format!(
"audit-cleanup: unexpected argument `{other}` — usage: {USAGE}"
)))
}
}
}
let rule = crate::audit::Retention::from_flags(days, keep_last, yes)
.map_err(MigrateError::Validation)?;
let deleted = rule.prune(pool).await?;
writeln!(w, "audit-cleanup done: deleted={deleted}")?;
Ok(())
}
async fn prune_cmd<W: Write>(pool: &Pool, args: &[String], w: &mut W) -> Result<(), MigrateError> {
let parsed = parse_prune_args(args)?;
if parsed.help {
writeln!(w, "prune [--model <name>] [--except <name>] [--pretend]")?;
writeln!(w)?;
writeln!(
w,
" Bulk-delete stale rows from every model implementing the"
)?;
writeln!(
w,
" Prunable trait and registered via `register_prunable!`."
)?;
writeln!(w)?;
writeln!(w, " --model <name> Restrict to this table (repeatable).")?;
writeln!(
w,
" --except <name> Skip this table (repeatable). Beats --model."
)?;
writeln!(
w,
" --pretend Count matching rows; skip the DELETE."
)?;
let names = crate::prunable::registered_names();
if names.is_empty() {
writeln!(w)?;
writeln!(w, " (no prunable models currently registered)")?;
} else {
writeln!(w)?;
writeln!(w, " Registered: {}", names.join(", "))?;
}
return Ok(());
}
let opts = crate::prunable::PruneOptions {
only: parsed.only,
except: parsed.except,
};
if parsed.pretend {
let reports = crate::prunable::prune_pretend(pool, &opts)
.await
.map_err(|e| MigrateError::Validation(format!("prune --pretend: {e}")))?;
if reports.is_empty() {
writeln!(w, "prune --pretend: no matching prunable models")?;
} else {
writeln!(w, "prune --pretend: would delete:")?;
for r in &reports {
writeln!(w, " - {}: {} row(s)", r.table, r.rows)?;
}
}
} else {
let reports = crate::prunable::prune_all(pool, &opts)
.await
.map_err(|e| MigrateError::Validation(format!("prune: {e}")))?;
if reports.is_empty() {
writeln!(w, "prune: no matching prunable models")?;
} else {
let total: u64 = reports.iter().map(|r| r.rows).sum();
writeln!(w, "prune: deleted {total} row(s):")?;
for r in &reports {
writeln!(w, " - {}: {} row(s)", r.table, r.rows)?;
}
}
}
Ok(())
}
#[cfg(feature = "cache")]
#[derive(Debug, Default, PartialEq)]
struct ClearCacheArgs {
table: Option<String>,
help: bool,
}
#[cfg(feature = "cache")]
fn parse_clear_cache_args(args: &[String]) -> Result<ClearCacheArgs, MigrateError> {
let mut out = ClearCacheArgs::default();
let mut iter = args.iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
"--help" | "-h" => {
out.help = true;
return Ok(out);
}
"--table" => {
let v = iter
.next()
.ok_or_else(|| MigrateError::Validation("--table expects a value".into()))?;
out.table = Some(v.clone());
}
other if other.starts_with('-') => {
return Err(MigrateError::Validation(format!("unknown flag: {other}")));
}
other => {
return Err(MigrateError::Validation(format!(
"unexpected positional argument: {other}"
)));
}
}
}
Ok(out)
}
#[cfg(feature = "cache")]
async fn clear_cache_cmd<W: Write>(
pool: &Pool,
verb: &str,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
let parsed = parse_clear_cache_args(args)?;
if parsed.help {
writeln!(w, "clear-cache [--table <name>]")?;
writeln!(w, " (alias: clearsessions)")?;
writeln!(w)?;
writeln!(w, " Delete every expired row from a DatabaseCache table.")?;
writeln!(w, " Default table: rustango_cache. Use --table for the")?;
writeln!(w, " session-backend table or any app-specific cache.")?;
return Ok(());
}
let table = parsed.table.unwrap_or_else(|| "rustango_cache".to_owned());
let exists = super::runner::try_table_resolves(pool, &table)
.await
.map_err(|e| MigrateError::Validation(format!("{verb}: {e}")))?;
if !exists {
writeln!(
w,
"nothing to clear: no {table} table (run createcachetable if you use DatabaseCache)"
)?;
return Ok(());
}
let cache = crate::cache::DatabaseCache::new(pool.clone(), &table);
let rows = cache
.purge_expired()
.await
.map_err(|e| MigrateError::Validation(format!("{verb}: {e}")))?;
writeln!(w, "{verb}: purged {rows} expired row(s) from {table}")?;
Ok(())
}
#[cfg(feature = "cache")]
async fn createcachetable_cmd<W: Write>(
pool: &Pool,
args: &[String],
w: &mut W,
) -> Result<(), MigrateError> {
let parsed = parse_clear_cache_args(args)?;
if parsed.help {
writeln!(w, "createcachetable [--table <name>]")?;
writeln!(w, " (alias: create-cache-table)")?;
writeln!(w)?;
writeln!(
w,
" Idempotently CREATE the DatabaseCache table via per-dialect DDL."
)?;
writeln!(
w,
" Default table: rustango_cache. Safe to call at every boot."
)?;
return Ok(());
}
let table = parsed.table.unwrap_or_else(|| "rustango_cache".to_owned());
let cache = crate::cache::DatabaseCache::new(pool.clone(), &table);
cache
.ensure_table()
.await
.map_err(|e| MigrateError::Validation(format!("createcachetable: {e}")))?;
writeln!(w, "createcachetable: ensured table {table} exists")?;
Ok(())
}
#[cfg(all(feature = "config", feature = "email"))]
async fn sendtestemail_cmd<W: Write>(args: &[String], w: &mut W) -> Result<(), MigrateError> {
let parsed = parse_sendtestemail_args(args)?;
if parsed.help {
writeln!(
w,
"sendtestemail --to <addr> [--from <addr>] [--subject <text>]"
)?;
writeln!(w)?;
writeln!(
w,
" Send a test message through the [mail] backend (console / memory /"
)?;
writeln!(
w,
" null / smtp). Use to verify SMTP credentials or mail wiring."
)?;
writeln!(w)?;
writeln!(w, " --to Recipient address (REQUIRED).")?;
writeln!(
w,
" --from From address. Defaults to [mail].from_address."
)?;
writeln!(
w,
" --subject Subject line. Defaults to `rustango: test email`."
)?;
return Ok(());
}
let to = parsed.to.as_deref().ok_or_else(|| {
MigrateError::Validation(
"sendtestemail: --to <addr> is required (run with --help for usage)".to_owned(),
)
})?;
let settings = crate::config::Settings::load_from_env().map_err(|e| {
MigrateError::Validation(format!(
"sendtestemail: failed to load settings: {e} — check config/{{default,$tier}}.toml",
))
})?;
let from = parsed
.from
.clone()
.or_else(|| settings.mail.from_address.clone())
.ok_or_else(|| {
MigrateError::Validation(
"sendtestemail: no --from supplied and [mail].from_address is unset".to_owned(),
)
})?;
let subject = parsed
.subject
.clone()
.unwrap_or_else(|| "rustango: test email".to_owned());
let body = "If you're reading this in a real inbox, your mail backend is wired correctly.\n\n\
Sent by `manage sendtestemail`."
.to_owned();
let mailer = crate::email::from_settings(&settings.mail)
.map_err(|e| MigrateError::Validation(format!("sendtestemail: {e}")))?;
let backend = settings.mail.backend.as_deref().unwrap_or("console");
let email = crate::email::Email::new()
.to(to)
.from(&from)
.subject(subject)
.body(body);
match mailer.send(&email).await {
Ok(()) => {
writeln!(
w,
"sendtestemail: ok (backend = {backend}, to = {to}, from = {from})"
)?;
Ok(())
}
Err(e) => Err(MigrateError::Validation(format!(
"sendtestemail: mailer.send failed (backend = {backend}): {e}"
))),
}
}
#[cfg(not(feature = "config"))]
async fn sendtestemail_cmd<W: Write>(_args: &[String], _w: &mut W) -> Result<(), MigrateError> {
Err(MigrateError::Validation(
"sendtestemail: this build was compiled without the `config` feature — settings \
can't be loaded. Rebuild with `--features config`."
.to_owned(),
))
}
#[cfg(all(feature = "config", not(feature = "email")))]
async fn sendtestemail_cmd<W: Write>(_args: &[String], _w: &mut W) -> Result<(), MigrateError> {
Err(MigrateError::Validation(
"sendtestemail: this build was compiled without the `email` feature — \
the mail backend isn't linked. Rebuild with `--features email`."
.to_owned(),
))
}
fn shown_args(cmd: &std::process::Command) -> String {
cmd.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect::<Vec<_>>()
.join(" ")
}
use crate::sql::connect_diagnosis::redact as redact_url;
#[derive(Default, Clone)]
pub(crate) struct DeployAuditEnv {
pub rustango_env: Option<String>,
pub session_secret: Option<String>,
pub database_url: Option<String>,
pub database_style_conflict: Option<String>,
pub apex_domain: ApexDomain,
pub bind: Option<String>,
pub ungated_admin: bool,
}
#[derive(Default, Clone)]
pub(crate) enum ApexDomain {
#[default]
SingleTenant,
#[cfg_attr(not(feature = "tenancy"), allow(dead_code))]
Tenancy(Option<String>),
}
fn deploy_audit_env(tenancy: bool) -> DeployAuditEnv {
DeployAuditEnv {
rustango_env: std::env::var("RUSTANGO_ENV").ok(),
session_secret: std::env::var("RUSTANGO_SESSION_SECRET").ok(),
database_url: crate::env::database_url_from_env().ok(),
database_style_conflict: crate::env::database_style_conflict(),
#[cfg(feature = "tenancy")]
apex_domain: if tenancy {
ApexDomain::Tenancy(crate::tenancy::server::configured_apex_domain())
} else {
ApexDomain::SingleTenant
},
#[cfg(not(feature = "tenancy"))]
apex_domain: {
let _ = tenancy;
ApexDomain::SingleTenant
},
bind: std::env::var("RUSTANGO_BIND").ok(),
#[cfg(feature = "admin")]
ungated_admin: crate::admin::ungated_admin_built(),
#[cfg(not(feature = "admin"))]
ungated_admin: false,
}
}
#[doc(hidden)]
#[derive(Debug, Default)]
pub struct DeployAuditFindings {
pub info: Vec<String>,
pub warnings: Vec<String>,
pub errors: Vec<String>,
}
const APEX_UNSET: &str = "RUSTANGO_APEX_DOMAIN is unset / `localhost` — tenancy projects need \
the public-facing apex (e.g. `app.example.com`) so subdomain resolution + cookie scoping \
work in production.";
#[cfg(feature = "tenancy")]
const SIGNED_BY_SESSION_SECRET: &str =
"session cookies (operator and tenant ones included) and JWTs";
#[cfg(not(feature = "tenancy"))]
const SIGNED_BY_SESSION_SECRET: &str = "session cookies and JWTs";
pub(crate) fn run_deploy_audit(env: &DeployAuditEnv, out: &mut DeployAuditFindings) {
if env.ungated_admin {
out.warnings.push(
"[admin] the admin router has no authentication — anyone can read and write \
every model. Call `admin::Builder::with_session_auth(secret)`, or gate the \
route yourself."
.into(),
);
}
match env.rustango_env.as_deref() {
Some(tier) if crate::env::is_prod_tier(tier) => {
out.info
.push("RUSTANGO_ENV is set to a production value".into());
}
Some(other) => {
out.warnings.push(format!(
"RUSTANGO_ENV is `{other}` — set to `prod` (or `production`) in deployed env"
));
}
None => {
out.warnings.push(
"RUSTANGO_ENV is unset — set to `prod` so config loaders pick the right tier"
.into(),
);
}
}
match env.session_secret.as_deref() {
None => {
out.errors.push(format!(
"RUSTANGO_SESSION_SECRET is unset — on the prod tier the server will not \
start; off prod, {SIGNED_BY_SESSION_SECRET} use the dev key in ./var/, which \
a redeploy loses. Set via `openssl rand -base64 32`."
));
}
Some(s) if s.contains("change-me") || s.contains("placeholder") => {
out.errors.push(
"RUSTANGO_SESSION_SECRET still contains the scaffolder placeholder \
(`change-me-...`) — replace with a real secret via `openssl rand -base64 32`."
.into(),
);
}
#[cfg(any(feature = "admin", feature = "tenancy"))]
Some(s) => match crate::session::SessionSecret::from_b64(s) {
Ok(_) => out
.info
.push("RUSTANGO_SESSION_SECRET decodes to a valid ≥32-byte key".into()),
Err(e) => out.errors.push(format!(
"{e}. This is the same check the runtime applies, so the app would \
fall back to an ephemeral key and sign everyone out on restart."
)),
},
#[cfg(not(any(feature = "admin", feature = "tenancy")))]
Some(_) => out.info.push(
"RUSTANGO_SESSION_SECRET is set but not validated — this build has \
neither `admin` nor `tenancy`, so nothing in it signs cookies or JWTs."
.into(),
),
}
match env.database_url.as_deref() {
None => out.errors.push(
"no database configured: DATABASE_URL, or DB_USER + DB_NAME, is required in production"
.into(),
),
Some(url) if url.contains("localhost") || url.contains("127.0.0.1") => {
out.warnings.push(
"DATABASE_URL points at localhost / 127.0.0.1 — verify this is intended in \
production (typically a managed service hostname)"
.into(),
);
}
Some(_) => out.info.push("DATABASE_URL set".into()),
}
if let Some(conflict) = &env.database_style_conflict {
out.warnings.push(conflict.clone());
}
match &env.apex_domain {
ApexDomain::SingleTenant => {}
ApexDomain::Tenancy(None) => out.warnings.push(APEX_UNSET.into()),
ApexDomain::Tenancy(Some(d)) if d == "localhost" => out.warnings.push(APEX_UNSET.into()),
ApexDomain::Tenancy(Some(_)) => out
.info
.push("RUSTANGO_APEX_DOMAIN set to a non-localhost value".into()),
}
match env.bind.as_deref() {
Some(b) if b.starts_with("127.0.0.1") => {
out.warnings.push(format!(
"RUSTANGO_BIND={b} only listens on loopback — production usually wants \
`0.0.0.0:<port>` to accept external traffic"
));
}
Some(_) | None => {} }
}
#[cfg(feature = "cache")]
pub(crate) fn login_store_audit(
lockout: &crate::account_lockout::Lockout,
out: &mut DeployAuditFindings,
) {
if lockout.stores_nothing() {
out.warnings
.push(crate::account_lockout::NULL_STORE_WARNING.into());
} else if lockout.is_process_local() {
out.info
.push(crate::account_lockout::CHECK_DEPLOY_NOTE.into());
}
#[cfg(feature = "admin")]
out.info
.push(crate::login_throttle::PROCESS_LOCAL_NOTE.into());
}
#[cfg(feature = "config")]
fn run_settings_audit(out: &mut DeployAuditFindings) {
use crate::config::Settings;
let env_tier = Settings::current_env_tier();
out.info
.push(format!("config tier resolved to `{env_tier}`"));
settings_audit_loaded(&env_tier, Settings::load_from_env(), out);
}
#[cfg(feature = "config")]
fn settings_audit_loaded(
env_tier: &str,
loaded: Result<crate::config::Settings, crate::config::ConfigError>,
out: &mut DeployAuditFindings,
) {
let settings = match loaded {
Ok(s) => s,
Err(e) if e.is_missing_config() => {
if crate::env::is_prod_tier(env_tier) {
out.warnings.push(
"config/default.toml is missing in prod tier — no [security] settings \
(allowed_hosts, headers, SSL redirect) are applied"
.into(),
);
}
return;
}
Err(e) => {
out.warnings.push(format!(
"config: failed to load settings for audit: {e} — fix the file shape \
to enable the rest of the deploy audit"
));
return;
}
};
settings_audit_check(&env_tier, &settings, out);
}
#[cfg(feature = "config")]
#[doc(hidden)]
pub fn settings_audit_check(
env_tier: &str,
settings: &crate::config::Settings,
out: &mut DeployAuditFindings,
) {
let in_prod = crate::env::is_prod_tier(env_tier);
if !in_prod {
return;
}
match settings.security.headers_preset.as_deref() {
None | Some("strict" | "relaxed") => {}
Some(preset @ ("dev" | "none" | "empty")) => out.warnings.push(format!(
"[security] headers_preset = `{preset}` in prod tier — promote to `strict` \
so HSTS / X-Frame-Options / X-Content-Type-Options / Referrer-Policy are emitted"
)),
Some(preset) => out.warnings.push(format!(
"[security] headers_preset = `{preset}` is not `strict`, `relaxed`, `dev` or \
`none`; the layer falls back to `strict`. Fix the spelling."
)),
}
if settings.security.secure_cookies == Some(false) {
out.warnings.push(
"[security] secure_cookies = false in prod tier — session and CSRF cookies are \
sent over plain HTTP too. Remove it or set it to true."
.into(),
);
}
if matches!(settings.security.hsts_max_age_secs, Some(0)) {
out.warnings.push(
"[security] hsts_max_age_secs = 0 in prod tier — disables HSTS, leaving TLS-strip \
attacks viable on first request"
.into(),
);
}
if settings.security.allowed_hosts.is_empty() {
out.warnings.push(
"[security] allowed_hosts = [] in prod tier — `host_validation::AllowedHostsLayer` \
passes every Host header (no DisallowedHost protection). Populate with the canonical \
host + subdomain wildcards before deploying."
.into(),
);
} else if settings
.security
.allowed_hosts
.iter()
.any(|h| h.trim() == "*")
{
out.warnings.push(
"[security] allowed_hosts contains \"*\" in prod tier — every Host header passes, \
so host validation is off. List the real hosts instead."
.into(),
);
}
if settings
.security
.cors_allowed_origins
.iter()
.any(|o| o.trim() == "*")
{
out.warnings.push(
"[security] cors_allowed_origins contains \"*\" in prod tier — any site can read \
non-credentialed responses. List the real origins."
.into(),
);
}
if matches!(settings.server.request_timeout_secs, None | Some(0)) {
out.warnings.push(
"[server] request_timeout_secs unset or 0 in prod tier — no handler timeout, so \
stuck requests hold workers. Set about 30."
.into(),
);
}
if settings.server.max_body_bytes.is_none() {
out.info.push(
"[server] max_body_bytes unset in prod tier — no body-limit layer; only axum's \
2 MiB extractor default applies, and streamed bodies are unlimited."
.into(),
);
}
if settings.security.csrf_trusted_origins.is_empty() {
out.info.push(
"[security] csrf_trusted_origins = [] in prod tier — the CSRF Origin check accepts \
only the request's own Host. Add any SPA or other origin that POSTs cross-host, \
or its requests get 403."
.into(),
);
}
if matches!(settings.security.secure_ssl_redirect, Some(false) | None) {
out.info.push(
"[security] secure_ssl_redirect unset/false in prod tier — plain-HTTP requests \
aren't redirected. If your LB doesn't already enforce HTTPS, mount \
`ssl_redirect::SslRedirectLayer` and set this true (configure `secure_proxy_ssl_header` \
to avoid loops)."
.into(),
);
}
if !settings.security.secure_proxy_ssl_header.is_empty()
&& settings.security.secure_proxy_ssl_header.len() != 2
{
out.warnings.push(format!(
"[security] secure_proxy_ssl_header has {} entries — expected exactly 2 \
([header_name, expected_value]); ssl_redirect layer will drop the trusted-header \
pair and redirect-loop behind the LB",
settings.security.secure_proxy_ssl_header.len()
));
}
if let Some(kib) = settings.auth.argon2_memory_kib {
if kib < 19_456 {
out.warnings.push(format!(
"[auth] argon2_memory_kib = {kib} in prod tier — OWASP 2024 recommends ≥ 19456 \
for password hashing brute-force resistance"
));
}
}
if let Some(ttl) = settings.auth.jwt.access_ttl_secs {
if ttl > 3600 {
out.warnings.push(format!(
"[auth.jwt] access_ttl_secs = {ttl} in prod tier — access tokens > 1h widen \
the leaked-token blast radius. The refresh flow rotates them; keep this short."
));
}
}
if settings.audit.retention_days.is_none() {
out.info.push(
"[audit] retention_days unset — log grows forever; consider setting + scheduling \
`manage audit-cleanup --days <N>`"
.into(),
);
}
if matches!(settings.routes.legacy_preset, Some(true)) {
out.info.push(
"[routes] legacy_preset = true — using the pre-v0.29 `__`-prefixed URLs \
(`/__login`, `/__admin`, …). Switch to the friendly defaults when bookmarks allow."
.into(),
);
}
if matches!(settings.mail.backend.as_deref(), None | Some("console")) {
out.warnings.push(
"[mail] backend unset or `console` in prod tier — mail bodies, including reset \
and magic-link URLs, are printed to stdout. Set `smtp` (or `null`)."
.into(),
);
}
if let Some(bind) = settings.server.bind.as_deref() {
if bind.starts_with("127.0.0.1") || bind.starts_with("localhost") {
out.warnings.push(format!(
"[server] bind = `{bind}` in prod tier — only listens on loopback. \
Production usually wants `0.0.0.0:<port>` to accept external traffic."
));
}
}
if let Some(backend) = settings.database.resolved_backend() {
let tenancy_on = crate::config::Settings::detected_features()
.iter()
.any(|f| *f == "tenancy");
if tenancy_on && backend != "postgres" {
out.warnings.push(format!(
"[database] backend = `{backend}` with `tenancy` feature on — \
schema-mode multi-tenancy is Postgres-only by language semantics \
(it relies on `SET search_path`). Either switch to a Postgres URL \
in prod, or use `crate::tenancy::DatabasePools<DB>` for the \
one-database-per-tenant model that works on sqlite/mysql."
));
}
}
let admin = &settings.admin;
if matches!(admin.csrf_cookie_secure, Some(false)) {
out.warnings.push(
"[admin] csrf_cookie_secure = false in prod tier — this key has no effect yet \
(#1379): the CSRF cookie follows `[security] secure_cookies`. Remove it."
.into(),
);
}
if let Some(hex) = admin.primary_color.as_deref() {
let stripped = hex.trim_start_matches('#');
let valid_len = matches!(stripped.len(), 3 | 6 | 8);
let all_hex = stripped.chars().all(|c| c.is_ascii_hexdigit());
if !hex.starts_with('#') || !valid_len || !all_hex {
out.warnings.push(format!(
"[admin] primary_color = `{hex}` does not parse as a hex color (expected \
`#RRGGBB`, `#RGB`, or `#RRGGBBAA`). The key has no effect yet (#1379), so \
the admin accent is unchanged either way."
));
}
}
if let Some(mode) = admin.theme_mode.as_deref() {
if !matches!(mode, "auto" | "light" | "dark") {
out.warnings.push(format!(
"[admin] theme_mode = `{mode}` is not one of `auto` / `light` / `dark` — \
the chrome will ignore it and fall back to `auto`."
));
}
}
if matches!(admin.session_timeout_minutes, Some(0)) {
out.info.push(
"[admin] session_timeout_minutes = 0 in prod tier — this key has no effect yet \
(#1379): the admin has no idle timeout whatever it says."
.into(),
);
}
if let Some(prefix) = admin.url_prefix.as_deref() {
if prefix.ends_with('/') && prefix.len() > 1 {
out.warnings.push(format!(
"[admin] url_prefix = `{prefix}` ends with a trailing slash — Builder will \
strip it, but config files should write the canonical form (no trailing slash) \
so reviewers can grep across deployments."
));
}
}
if crate::config::Settings::detected_features()
.iter()
.any(|f| *f == "mcp")
{
if settings.mcp.allowed_origins.is_empty() {
out.info.push(
"[mcp] allowed_origins = [] in prod tier — the MCP endpoint is same-origin only. \
If browser-based agents call it cross-origin, populate the CORS allow-list; \
server-to-server agents (no browser) don't need it."
.into(),
);
}
if matches!(settings.mcp.enable_sse, Some(false)) {
out.info.push(
"[mcp] enable_sse = false in prod tier — no server→client notifications \
(list_changed / progress); agents must re-list to observe capability changes."
.into(),
);
}
if settings.mcp.rate_limit().is_none() {
out.info.push(
"[mcp] rate_limit_per_minute unset or 0 in prod tier — the agent token + JSON-RPC \
endpoints are unthrottled. Set a per-IP cap to blunt credential-stuffing / abuse."
.into(),
);
}
}
}
#[cfg(test)]
mod gen_tests {
use super::*;
#[cfg(not(feature = "tenancy"))]
#[test]
fn unmanaged_registry_model_is_not_tenancy() {
use crate::core::Model as _;
assert_eq!(
crate::i18n::db::Translation::SCHEMA.scope,
crate::core::ModelScope::Registry
);
assert!(!has_registry_models());
}
#[test]
fn pool_free_verbs_need_no_database() {
for verb in [
"",
"--help",
"-h",
"help",
"version",
"--version",
"docs",
"db:info",
] {
let args = vec![verb.to_owned()];
let mut out = Vec::new();
let res = run_pool_free(&args, &mut out);
assert!(
res.is_some(),
"`{verb}` must be dispatchable without a pool"
);
assert!(res.unwrap().is_ok(), "`{verb}` failed");
assert!(!out.is_empty(), "`{verb}` produced no output");
}
}
#[test]
fn generators_need_no_database() {
for verb in [
"make:viewset",
"make:serializer",
"make:form",
"make:job",
"make:scheduled",
"make:worker",
"make:notification",
"make:middleware",
"make:test",
"startapp",
] {
let args = vec![verb.to_owned()];
assert!(
run_pool_free(&args, &mut Vec::new()).is_some(),
"`{verb}` must be dispatchable without a pool"
);
}
}
#[test]
fn serializer_docs_snippet_matches_the_template() {
let docs = include_str!("../../../../docs/serializers.md");
let start = docs
.find("//! Auto-scaffolded by `manage make:serializer PostSerializer`.")
.expect("snippet");
let snippet = &docs[start..start + docs[start..].find("```").expect("fence")];
let rendered = serializer_template("PostSerializer", &ModelRef::bare("Post"), "rustango");
for line in snippet.lines().filter(|l| !l.trim_start().starts_with('/')) {
assert!(
rendered.contains(line),
"docs line not in template: `{line}`"
);
}
}
#[test]
fn database_verbs_are_not_short_circuited() {
for verb in [
"migrate",
"makemigrations",
"downgrade",
"showmigrations",
"about",
"check",
"dumpdata",
"loaddata",
] {
let args = vec![verb.to_owned()];
assert!(
run_pool_free(&args, &mut Vec::new()).is_none(),
"`{verb}` touches the database — it must fall through to `run`"
);
}
}
#[test]
fn pascal_to_snake_cases() {
assert_eq!(pascal_to_snake("Post"), "post");
assert_eq!(pascal_to_snake("PostViewSet"), "post_view_set");
assert_eq!(pascal_to_snake("API"), "a_p_i"); assert_eq!(pascal_to_snake("UserNotification"), "user_notification");
}
#[test]
fn make_test_accepts_the_documented_snake_case_name() {
let args = vec!["post_smoke".to_owned()];
let (name, _, _) = parse_name_and_model_as(&args, NameShape::Identifier)
.expect("the documented example must be accepted");
assert_eq!(name, "post_smoke");
assert_eq!(pascal_to_snake(&name), "post_smoke");
}
#[test]
fn make_test_still_accepts_pascal_case() {
let args = vec!["PostSmoke".to_owned()];
let (name, _, _) = parse_name_and_model_as(&args, NameShape::Identifier).unwrap();
assert_eq!(pascal_to_snake(&name), "post_smoke");
}
#[test]
fn type_generators_still_reject_snake_case() {
let args = vec!["post_view_set".to_owned()];
assert!(parse_name_and_model_as(&args, NameShape::Type).is_err());
assert!(parse_name_and_model(&args).is_err());
}
#[test]
fn identifier_rules() {
assert!(is_valid_identifier("post_smoke"));
assert!(is_valid_identifier("PostSmoke"));
assert!(is_valid_identifier("_leading"));
assert!(is_valid_identifier("a1"));
assert!(!is_valid_identifier("1leading")); assert!(!is_valid_identifier("has-dash"));
assert!(!is_valid_identifier(""));
}
#[test]
fn is_valid_type_name_accepts_pascal() {
assert!(is_valid_type_name("Post"));
assert!(is_valid_type_name("PostViewSet"));
assert!(is_valid_type_name("Foo_Bar"));
}
#[test]
fn is_valid_type_name_rejects_invalid() {
assert!(!is_valid_type_name(""));
assert!(!is_valid_type_name("post")); assert!(!is_valid_type_name("123Foo")); assert!(!is_valid_type_name("Post!")); }
#[test]
fn parse_name_and_model_basic() {
let (n, m, c) = parse_name_and_model(&["PostViewSet".into()]).unwrap();
assert_eq!(n, "PostViewSet");
assert_eq!(m, None);
assert_eq!(c, "rustango");
}
#[test]
fn parse_name_and_model_with_model_flag() {
let args: Vec<String> = vec!["PostViewSet".into(), "--model".into(), "Post".into()];
let (n, m, c) = parse_name_and_model(&args).unwrap();
assert_eq!(n, "PostViewSet");
assert_eq!(m, Some("Post".into()));
assert_eq!(c, "rustango");
}
#[test]
fn parse_name_and_model_with_crate_flag() {
let args: Vec<String> = vec!["PostSerializer".into(), "--crate".into(), "orm".into()];
let (n, m, c) = parse_name_and_model(&args).unwrap();
assert_eq!(n, "PostSerializer");
assert_eq!(m, None);
assert_eq!(c, "orm");
}
#[test]
fn parse_name_and_model_rejects_empty_crate() {
let args: Vec<String> = vec!["Post".into(), "--crate".into(), "".into()];
let err = parse_name_and_model(&args)
.err()
.expect("empty crate must fail");
assert!(format!("{err}").contains("must not be empty"));
}
#[test]
fn parse_name_and_model_rejects_missing_crate_value() {
let args: Vec<String> = vec!["Post".into(), "--crate".into()];
let err = parse_name_and_model(&args)
.err()
.expect("missing crate value must fail");
assert!(format!("{err}").contains("--crate requires a value"));
}
#[test]
fn parse_name_and_model_rejects_missing_name() {
let r = parse_name_and_model(&[]);
assert!(r.is_err());
}
#[test]
fn parse_name_and_model_rejects_lowercase_name() {
let r = parse_name_and_model(&["postviewset".into()]);
assert!(r.is_err());
}
#[test]
fn names_that_make_a_reserved_module_are_refused() {
for n in ["Type", "Match", "Std", "Core", "Crate", "Self", "Super"] {
assert!(parse_name_and_model(&[n.into()]).is_err(), "{n}");
}
for n in ["type", "self", "std"] {
assert!(
parse_name_and_model_as(&[n.into()], NameShape::Identifier).is_err(),
"{n}"
);
}
assert!(parse_name_and_model(&["Typed".into()]).is_ok());
}
#[test]
fn pool_templates_import_their_model() {
let post = ModelRef::bare("Post");
let vs = viewset_template_pool("PostViewSet", &post, "post_view_set", "rustango", "app");
let ser = serializer_template("PostSerializer", &post, "rustango");
for body in [&vs, &ser] {
assert!(body.contains("use crate::models::Post;"), "{body}");
}
let tenant = viewset_template_tenant("PostViewSet", &post, "post_view_set", "rustango");
assert!(
tenant.contains(".merge(crate::post_view_set::router())"),
"{tenant}"
);
}
#[test]
fn model_ref_finds_an_app_model() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("src");
std::fs::create_dir_all(src.join("blog")).unwrap();
std::fs::write(src.join("models.rs"), "pub struct Author;\n").unwrap();
std::fs::write(
src.join("blog/models.rs"),
"#[derive(Model)]\npub struct Post {\n pub id: i64,\n}\n",
)
.unwrap();
let path = |m: &str| ModelRef::resolve(m, &src).path;
assert_eq!(path("Post"), "crate::blog::models::Post");
assert_eq!(path("Author"), "crate::models::Author");
assert_eq!(path("Missing"), "crate::models::Missing");
assert_eq!(path("shop::models::Item"), "crate::shop::models::Item");
let explicit = ModelRef::resolve("crate::shop::models::Item", &src);
assert_eq!(
(explicit.path.as_str(), explicit.name.as_str()),
("crate::shop::models::Item", "Item")
);
let post = ModelRef::resolve("Post", &src);
let vs = viewset_template_pool("PostViewSet", &post, "post_view_set", "rustango", "my_app");
assert!(vs.contains("use crate::blog::models::Post;"), "{vs}");
assert!(vs.contains("model = Post,"), "{vs}");
assert!(
vs.contains(r#".nest_with("/api", |pool| my_app::post_view_set::PostViewSet::router("/post_view_set", pool))"#),
"{vs}"
);
}
#[test]
fn make_verbs_name_the_feature_their_output_needs() {
let none = |_: &str| false;
for (verb, feature) in [
("make:serializer", "serializer"),
("make:form", "forms"),
("make:job", "jobs"),
("make:worker", "jobs"),
("make:scheduled", "scheduler"),
("make:notification", "notifications"),
("make:test", "admin"),
] {
assert_eq!(missing_make_feature(verb, none), Some(feature), "{verb}");
assert_eq!(missing_make_feature(verb, |_| true), None, "{verb}");
}
for verb in ["make:middleware", "make:api_routes", "startapp"] {
assert_eq!(missing_make_feature(verb, none), None, "{verb}");
}
let err = make_cmd("make:test", &[], &mut Vec::new(), none).unwrap_err();
assert!(err.to_string().contains("`admin`"), "{err}");
}
#[test]
fn make_verbs_follow_a_renamed_dependency() {
let _guard = cwd_lock().lock().unwrap_or_else(|p| p.into_inner());
let dir = tempfile::tempdir().unwrap();
std::fs::write(
dir.path().join("Cargo.toml"),
"[package]\nname = \"short\"\n\n[dependencies]\n\
rgo = { package = \"rustango\", version = \"0.61\" }\n",
)
.unwrap();
let prev = std::env::current_dir().unwrap();
std::env::set_current_dir(dir.path()).unwrap();
let run = make_cmd(
"make:job",
&["SendMail".to_owned()],
&mut Vec::new(),
|_| true,
);
let body = std::fs::read_to_string("src/send_mail.rs");
let _ = std::env::set_current_dir(&prev);
run.unwrap();
let body = body.unwrap();
assert!(
body.contains("rgo::") && !body.contains("rustango::"),
"{body}"
);
}
#[test]
fn make_viewset_needs_the_feature_of_its_template() {
let run = |flag: &str, missing: &'static str| {
let args = [flag.to_owned(), "--crate".to_owned()];
make_cmd("make:viewset", &args, &mut Vec::new(), move |f| {
f != missing
})
.unwrap_err()
.to_string()
};
assert!(run("--tenant", "tenancy").contains("`tenancy`"));
assert!(run("--no-tenant", "admin").contains("`admin`"));
let err = run("--tenant", "admin");
assert!(err.contains("--crate requires a value"), "{err}");
}
#[test]
fn compiled_feature_matches_the_build() {
for (feature, on) in [
("admin", cfg!(feature = "admin")),
("tenancy", cfg!(feature = "tenancy")),
("serializer", cfg!(feature = "serializer")),
("forms", cfg!(feature = "forms")),
("jobs", cfg!(feature = "jobs")),
("scheduler", cfg!(feature = "scheduler")),
("notifications", cfg!(feature = "notifications")),
("no_such_feature", false),
] {
assert_eq!(compiled_feature(feature), on, "{feature}");
}
}
#[test]
fn worker_loads_dotenv_before_reading_the_url() {
let body = worker_template("Mailer", "mailer", "rustango", "my_app");
let dotenv = body.find("dotenvy::dotenv()").expect("loads .env");
assert!(
dotenv < body.find("database_url_from_env").unwrap(),
"{body}"
);
}
#[test]
fn worker_leaves_logging_to_the_main_macro() {
let body = worker_template("Mailer", "mailer", "rustango", "my_app");
assert!(!body.contains("logging::setup"), "{body}");
}
fn run_make(cmd: &str, args: &[&str], dirs: &[&str], read: &[&str]) -> (String, Vec<String>) {
let _guard = cwd_lock().lock().unwrap_or_else(|p| p.into_inner());
let dir = tempfile::tempdir().unwrap();
std::fs::write(
dir.path().join("Cargo.toml"),
"[package]\nname = \"shop\"\n\n[dependencies]\nrustango = \"0.61\"\n",
)
.unwrap();
for d in dirs {
std::fs::create_dir_all(dir.path().join(d)).unwrap();
}
std::fs::write(dir.path().join("src/lib.rs"), "pub mod urls;\n").unwrap();
let prev = std::env::current_dir().unwrap();
std::env::set_current_dir(dir.path()).unwrap();
let mut out = Vec::new();
let args: Vec<String> = args.iter().map(|a| (*a).to_owned()).collect();
let run = make_cmd(cmd, &args, &mut out, |_| true);
let files: Vec<String> = read
.iter()
.map(|f| std::fs::read_to_string(f).unwrap_or_default())
.collect();
let _ = std::env::set_current_dir(&prev);
run.unwrap();
(String::from_utf8(out).unwrap(), files)
}
#[test]
fn api_routes_hints_name_a_mount_that_compiles() {
let (out, files) = run_make(
"make:api_routes",
&["blog"],
&["src/blog"],
&["src/blog/api_routes.rs"],
);
assert!(
out.contains(r#".nest_with("/api", |pool| shop::blog::api_routes::api(pool))"#),
"{out}"
);
assert!(!out.contains(".merge("), "{out}");
let body = &files[0];
assert!(!body.contains("viewsets::"), "{body}");
assert!(body.contains("crate::<snake>::<Name>::router("), "{body}");
let (out, files) = run_make(
"make:api_routes",
&["blog", "--tenant"],
&["src/blog"],
&["src/blog/api_routes.rs"],
);
assert!(
out.contains(".merge(crate::blog::api_routes::api())"),
"{out}"
);
assert!(!files[0].contains("viewsets::"), "{}", files[0]);
assert!(
files[0].contains("crate::<snake>::router()"),
"{}",
files[0]
);
}
#[test]
fn make_form_imports_only_what_it_uses() {
let (_, files) = run_make("make:form", &["PostForm"], &["src"], &["src/post_form.rs"]);
assert!(
!files[0].contains("use rustango::forms::Form;"),
"{}",
files[0]
);
}
#[test]
fn make_scheduled_wires_into_a_worker() {
let (_, files) = run_make("make:scheduled", &["Tick"], &["src"], &["src/tick.rs"]);
let body = &files[0];
assert!(body.contains("make:worker"), "{body}");
assert!(!body.contains("in main.rs"), "{body}");
assert!(
body.contains("Arc::new(shop::tick::Tick { pool: pool.clone() })"),
"{body}"
);
}
#[test]
fn tenant_viewset_mounts_an_auth_layer() {
let body = viewset_template_tenant(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango",
);
assert!(
body.contains(".require_auth(vec![Arc::new(ModelBackend)])"),
"{body}"
);
assert!(body.contains("use rustango::tenancy::{ModelBackend, RouterAuthExt as _};"));
}
#[test]
fn startapp_report_does_not_ask_for_the_mod_it_added() {
let report = super::super::scaffold::StartAppReport {
written: vec!["src/shop/mod.rs".into()],
patched: vec!["src/lib.rs".into(), "src/urls.rs".into()],
..Default::default()
};
let mut out = Vec::new();
write_startapp_report(&mut out, "shop", &report).unwrap();
let out = String::from_utf8(out).unwrap();
assert!(!out.contains("add `mod"), "{out}");
assert!(out.contains("makemigrations"), "{out}");
}
#[test]
fn parse_dumpdata_args_defaults() {
let p = parse_dumpdata_args(&[]).unwrap();
assert!(p.model_filters.is_empty());
assert_eq!(p.indent, 2);
assert!(!p.help);
}
#[test]
fn parse_dumpdata_args_help_flag() {
let p = parse_dumpdata_args(&["--help".into()]).unwrap();
assert!(p.help);
}
#[test]
fn parse_dumpdata_args_collects_multiple_models() {
let args: Vec<String> = vec![
"--model".into(),
"blog.Article".into(),
"--model".into(),
"Author".into(),
];
let p = parse_dumpdata_args(&args).unwrap();
assert_eq!(p.model_filters, vec!["blog.Article", "Author"]);
}
#[test]
fn parse_dumpdata_args_indent_accepts_zero() {
let p = parse_dumpdata_args(&["--indent".into(), "0".into()]).unwrap();
assert_eq!(p.indent, 0);
}
#[test]
fn parse_dumpdata_args_rejects_unknown_flag() {
let r = parse_dumpdata_args(&["--unknown".into()]);
assert!(r.is_err());
assert!(format!("{}", r.unwrap_err()).contains("unknown flag"));
}
#[test]
fn parse_dumpdata_args_rejects_missing_model_value() {
let r = parse_dumpdata_args(&["--model".into()]);
assert!(r.is_err());
}
#[test]
fn parse_dumpdata_args_rejects_non_integer_indent() {
let r = parse_dumpdata_args(&["--indent".into(), "abc".into()]);
assert!(r.is_err());
}
#[test]
fn parse_dumpdata_args_rejects_positional() {
let r = parse_dumpdata_args(&["unexpected".into()]);
assert!(r.is_err());
assert!(format!("{}", r.unwrap_err()).contains("positional"));
}
#[test]
fn parse_loaddata_args_requires_path() {
let r = parse_loaddata_args(&[]);
assert!(r.is_err());
assert!(format!("{}", r.unwrap_err()).contains("missing"));
}
#[test]
fn parse_loaddata_args_path_then_flag() {
let p = parse_loaddata_args(&["fixtures.json".into(), "--fail-fast".into()]).unwrap();
assert_eq!(p.file, "fixtures.json");
assert!(p.fail_fast);
}
#[test]
fn parse_loaddata_args_flag_then_path() {
let p = parse_loaddata_args(&["--fail-fast".into(), "fixtures.json".into()]).unwrap();
assert_eq!(p.file, "fixtures.json");
assert!(p.fail_fast);
}
#[test]
fn parse_loaddata_args_help_flag_short_circuits() {
let p = parse_loaddata_args(&["--help".into()]).unwrap();
assert!(p.help);
}
#[test]
fn parse_loaddata_args_rejects_two_positionals() {
let r = parse_loaddata_args(&["a.json".into(), "b.json".into()]);
assert!(r.is_err());
}
#[test]
fn parse_loaddata_args_rejects_unknown_flag() {
let r = parse_loaddata_args(&["--unknown".into(), "a.json".into()]);
assert!(r.is_err());
}
fn field(name: &'static str, ty: crate::core::FieldType) -> crate::core::FieldSchema {
crate::core::FieldSchema {
name,
column: name,
ty,
nullable: true,
primary_key: false,
relation: None,
max_length: None,
min: None,
max: None,
default: None,
auto: false,
auto_now: false,
unique: false,
generated_as: None,
help_text: None,
choices: None,
db_comment: None,
verbose_name: None,
editable: true,
blank: false,
case_insensitive: false,
fk_on_delete: None,
validators: &[],
cast: None,
}
}
#[test]
fn fixture_writer_matches_serde_json() {
let entries = vec![
serde_json::json!({"model": "a.B", "pk": 1, "fields": {"s": "x\ny", "l": [1, 2]}}),
serde_json::json!({"model": "a.B", "pk": 2, "fields": {}}),
];
for n in 0..=entries.len() {
let some = &entries[..n];
for indent in [0, 2] {
let mut buf = Vec::new();
let mut w = FixtureWriter::new(&mut buf, indent);
for e in some {
w.entry(e).unwrap();
}
w.finish().unwrap();
let whole = if indent == 0 {
serde_json::to_string(some)
} else {
serde_json::to_string_pretty(some)
}
.unwrap();
assert_eq!(String::from_utf8(buf).unwrap(), format!("{whole}\n"));
}
}
}
#[test]
fn only_loaddata_reads_the_json_tag() {
use crate::core::{FieldType, SqlValue};
use serde_json::{json, Value};
let mut f = field("j", FieldType::Json);
f.nullable = false;
let tagged = json!({"$json": 1});
assert!(
matches!(json_to_sql_value(&tagged, &f), Ok(SqlValue::Json(ref v)) if *v == tagged),
"a seed fixture keeps the literal object"
);
assert!(matches!(
json_to_sql_value(&Value::Null, &f),
Ok(SqlValue::Null)
));
assert!(matches!(loaddata_value(&tagged, &f), Ok(SqlValue::Json(ref v)) if *v == json!(1)));
assert!(matches!(
loaddata_value(&Value::Null, &f),
Ok(SqlValue::Json(Value::Null))
));
f.nullable = true;
assert!(matches!(
loaddata_value(&Value::Null, &f),
Ok(SqlValue::Null)
));
}
#[test]
fn json_to_sql_value_null_passes_through() {
let f = field("x", crate::core::FieldType::I64);
let v = json_to_sql_value(&serde_json::Value::Null, &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::Null));
}
#[test]
fn json_to_sql_value_i32_range_check() {
let f = field("x", crate::core::FieldType::I32);
let v = json_to_sql_value(&serde_json::json!(42), &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::I32(42)));
let r = json_to_sql_value(&serde_json::json!(99999999999_i64), &f);
assert!(r.is_err());
}
#[test]
fn json_to_sql_value_reads_what_dumpdata_writes() {
let f = field("x", crate::core::FieldType::I64);
let v = json_to_sql_value(&serde_json::json!("42"), &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::I64(42)), "{v:?}");
assert!(json_to_sql_value(&serde_json::json!("4x"), &f).is_err());
let t = field("t", crate::core::FieldType::Time);
let v = json_to_sql_value(&serde_json::json!("12:34:56.789"), &t).unwrap();
let want = chrono::NaiveTime::from_hms_milli_opt(12, 34, 56, 789).unwrap();
assert!(
matches!(v, crate::core::SqlValue::Time(x) if x == want),
"{v:?}"
);
}
#[test]
fn json_to_sql_value_string_passes_through() {
let f = field("x", crate::core::FieldType::String);
let v = json_to_sql_value(&serde_json::json!("hello"), &f).unwrap();
match v {
crate::core::SqlValue::String(s) => assert_eq!(s, "hello"),
other => panic!("expected SqlValue::String, got {other:?}"),
}
}
#[test]
fn json_to_sql_value_date_parses_iso() {
let f = field("x", crate::core::FieldType::Date);
let v = json_to_sql_value(&serde_json::json!("2025-01-02"), &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::Date(_)));
let r = json_to_sql_value(&serde_json::json!("not a date"), &f);
assert!(r.is_err());
}
#[test]
fn json_to_sql_value_datetime_parses_rfc3339() {
let f = field("x", crate::core::FieldType::DateTime);
let v = json_to_sql_value(&serde_json::json!("2025-01-02T12:00:00+00:00"), &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::DateTime(_)));
}
#[test]
fn json_to_sql_value_decimal_parses_string_and_number() {
let f = field("x", crate::core::FieldType::Decimal);
let v = json_to_sql_value(&serde_json::json!("123.45"), &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::Decimal(_)));
let v = json_to_sql_value(&serde_json::json!(42), &f).unwrap();
assert!(matches!(v, crate::core::SqlValue::Decimal(_)));
}
#[test]
fn json_to_sql_value_binary_hex_round_trip() {
let f = field("x", crate::core::FieldType::Binary);
let v = json_to_sql_value(&serde_json::json!("deadbeef"), &f).unwrap();
match v {
crate::core::SqlValue::Binary(b) => assert_eq!(b, vec![0xde, 0xad, 0xbe, 0xef]),
other => panic!("expected Binary, got {other:?}"),
}
let r = json_to_sql_value(&serde_json::json!("abc"), &f);
assert!(r.is_err());
}
#[test]
fn json_to_sql_value_json_field_stays_value() {
let f = field("x", crate::core::FieldType::Json);
let v = json_to_sql_value(&serde_json::json!({"k": "v"}), &f).unwrap();
match v {
crate::core::SqlValue::Json(j) => assert_eq!(j, serde_json::json!({"k": "v"})),
other => panic!("expected Json, got {other:?}"),
}
}
#[test]
fn showurls_help_emits_usage_line() {
let mut buf: Vec<u8> = Vec::new();
let r = showurls_cmd(&["--help".into()], &mut buf);
assert!(r.is_ok());
let s = String::from_utf8(buf).unwrap();
assert!(s.contains("showurls"), "help text mentions verb: {s}");
assert!(s.contains("plain"));
assert!(s.contains("json"));
}
#[test]
fn showurls_rejects_unknown_format() {
let mut buf: Vec<u8> = Vec::new();
let r = showurls_cmd(&["--format".into(), "xml".into()], &mut buf);
assert!(r.is_err());
let e = format!("{}", r.unwrap_err());
assert!(e.contains("unknown value"), "error: {e}");
}
#[test]
fn showurls_rejects_unknown_flag() {
let mut buf: Vec<u8> = Vec::new();
let r = showurls_cmd(&["--badflag".into()], &mut buf);
assert!(r.is_err());
}
#[test]
fn showurls_rejects_positional() {
let mut buf: Vec<u8> = Vec::new();
let r = showurls_cmd(&["extra".into()], &mut buf);
assert!(r.is_err());
let e = format!("{}", r.unwrap_err());
assert!(e.contains("positional"));
}
#[test]
fn showurls_plain_format_runs_without_error() {
let mut buf: Vec<u8> = Vec::new();
let r = showurls_cmd(&[], &mut buf);
assert!(r.is_ok(), "showurls plain shouldn't error: {r:?}");
}
#[test]
fn showurls_json_format_emits_json_array() {
let mut buf: Vec<u8> = Vec::new();
let r = showurls_cmd(&["--format".into(), "json".into()], &mut buf);
assert!(r.is_ok());
let s = String::from_utf8(buf).unwrap();
let parsed: serde_json::Value = serde_json::from_str(s.trim()).unwrap();
assert!(parsed.is_array(), "expected JSON array, got: {s}");
}
#[test]
fn showmodels_help_short_circuits() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&["--help".into()], &mut buf);
assert!(r.is_ok());
let s = String::from_utf8(buf).unwrap();
assert!(s.contains("showmodels"));
assert!(s.contains("--app"));
}
#[test]
fn showmodels_rejects_unknown_format() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&["--format".into(), "xml".into()], &mut buf);
assert!(r.is_err());
let e = format!("{}", r.unwrap_err());
assert!(e.contains("unknown value"));
}
#[test]
fn showmodels_rejects_unknown_flag() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&["--badflag".into()], &mut buf);
assert!(r.is_err());
}
#[test]
fn showmodels_rejects_positional() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&["unexpected".into()], &mut buf);
assert!(r.is_err());
}
#[test]
fn showmodels_app_filter_requires_value() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&["--app".into()], &mut buf);
assert!(r.is_err());
}
#[test]
fn showmodels_plain_runs_clean() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&[], &mut buf);
assert!(r.is_ok(), "showmodels should not error: {r:?}");
}
#[test]
fn showmodels_json_emits_parseable_array() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(&["--format".into(), "json".into()], &mut buf);
assert!(r.is_ok());
let s = String::from_utf8(buf).unwrap();
let parsed: serde_json::Value = serde_json::from_str(s.trim()).unwrap();
assert!(parsed.is_array());
}
#[test]
fn parse_flush_args_defaults() {
let p = parse_flush_args(&[]).unwrap();
assert!(!p.yes);
assert!(p.apps.is_empty());
assert!(p.models.is_empty());
assert!(!p.help);
}
#[test]
fn parse_flush_args_yes_flag() {
let p = parse_flush_args(&["--yes".into()]).unwrap();
assert!(p.yes);
}
#[test]
fn parse_flush_args_collects_multiple_apps_and_models() {
let args: Vec<String> = vec![
"--app".into(),
"blog".into(),
"--model".into(),
"Article".into(),
"--app".into(),
"shop".into(),
"--model".into(),
"shop.Order".into(),
];
let p = parse_flush_args(&args).unwrap();
assert_eq!(p.apps, vec!["blog", "shop"]);
assert_eq!(p.models, vec!["Article", "shop.Order"]);
}
#[test]
fn parse_flush_args_help_short_circuits() {
let p = parse_flush_args(&["--help".into()]).unwrap();
assert!(p.help);
}
#[test]
fn flush_plan_breaks_cycles_at_nullable_links() {
use crate::core::{FieldSchema, FieldType, ModelSchema, Relation};
fn table(name: &'static str, to: &'static str, nullable: bool) -> &'static ModelSchema {
let mut f = FieldSchema::new("link", "link", FieldType::I64);
f.relation = Some(Relation::fk(to, "id"));
f.nullable = nullable;
let mut s = ModelSchema::new(name, name);
s.fields = Box::leak(Box::new([f]));
Box::leak(Box::new(s))
}
let stuck = flush_plan(&[table("a", "b", false), table("b", "a", false)]);
assert_eq!(stuck.stuck, ["a", "b"]);
let plan = flush_plan(&[table("a", "b", true), table("b", "a", false)]);
assert!(plan.stuck.is_empty());
let unlinked: Vec<_> = plan
.unlink
.iter()
.map(|(s, c)| (s.table, c.clone()))
.collect();
assert_eq!(unlinked, [("a", vec!["link"])]);
let order: Vec<_> = plan.order.iter().map(|s| s.table).collect();
assert_eq!(order, ["b", "a"]);
}
#[test]
fn parse_flush_args_rejects_unknown_flag() {
let r = parse_flush_args(&["--badflag".into()]);
assert!(r.is_err());
}
#[test]
fn parse_flush_args_rejects_positional() {
let r = parse_flush_args(&["unexpected".into()]);
assert!(r.is_err());
}
#[test]
fn parse_flush_args_app_requires_value() {
let r = parse_flush_args(&["--app".into()]);
assert!(r.is_err());
}
#[test]
fn parse_flush_args_model_requires_value() {
let r = parse_flush_args(&["--model".into()]);
assert!(r.is_err());
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_defaults_empty() {
let p = parse_sendtestemail_args(&[]).unwrap();
assert!(p.to.is_none());
assert!(p.from.is_none());
assert!(p.subject.is_none());
assert!(!p.help);
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_collects_to_from_subject() {
let args: Vec<String> = vec![
"--to".into(),
"ops@example.com".into(),
"--from".into(),
"bot@example.com".into(),
"--subject".into(),
"ping".into(),
];
let p = parse_sendtestemail_args(&args).unwrap();
assert_eq!(p.to.as_deref(), Some("ops@example.com"));
assert_eq!(p.from.as_deref(), Some("bot@example.com"));
assert_eq!(p.subject.as_deref(), Some("ping"));
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_help_short_circuits() {
let p = parse_sendtestemail_args(&["--help".into()]).unwrap();
assert!(p.help);
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_rejects_unknown_flag() {
let r = parse_sendtestemail_args(&["--bogus".into()]);
assert!(r.is_err());
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_rejects_positional() {
let r = parse_sendtestemail_args(&["unexpected".into()]);
assert!(r.is_err());
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_to_requires_value() {
let r = parse_sendtestemail_args(&["--to".into()]);
assert!(r.is_err());
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_from_requires_value() {
let r = parse_sendtestemail_args(&["--from".into()]);
assert!(r.is_err());
}
#[cfg(all(feature = "config", feature = "email"))]
#[test]
fn parse_sendtestemail_args_subject_requires_value() {
let r = parse_sendtestemail_args(&["--subject".into()]);
assert!(r.is_err());
}
#[cfg(all(feature = "config", feature = "email"))]
#[tokio::test]
async fn sendtestemail_help_short_circuits_without_settings_lookup() {
let mut buf: Vec<u8> = Vec::new();
let r = sendtestemail_cmd(&["--help".into()], &mut buf).await;
assert!(r.is_ok());
let s = String::from_utf8(buf).unwrap();
assert!(s.contains("sendtestemail"));
assert!(s.contains("--to"));
}
#[cfg(feature = "config")]
#[tokio::test]
async fn sendtestemail_errors_when_to_missing() {
let mut buf: Vec<u8> = Vec::new();
let r = sendtestemail_cmd(&[], &mut buf).await;
assert!(r.is_err(), "expected Validation error for missing --to");
}
#[test]
fn showmodels_app_filter_excludes_other_apps() {
let mut buf: Vec<u8> = Vec::new();
let r = showmodels_cmd(
&["--app".into(), "xyz_definitely_not_an_app".into()],
&mut buf,
);
assert!(r.is_ok());
let s = String::from_utf8(buf).unwrap();
assert!(
s.contains("no models registered"),
"expected empty-output marker for unknown app, got: {s}"
);
}
#[test]
fn viewset_template_pool_emits_derive_macro() {
let body = viewset_template_pool(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango",
"app",
);
assert!(
body.contains("#[derive(ViewSet)]"),
"expected derive macro, got: {body}"
);
assert!(
body.contains("PostViewSet::router"),
"expected `Name::router(...)` mount hint, got: {body}"
);
assert!(
!body.contains("tenant_router"),
"pool template must NOT reference tenant_router, got: {body}"
);
assert!(
body.contains("use rustango::ViewSet;"),
"default crate root must emit `use rustango::ViewSet;`, got: {body}"
);
}
#[test]
fn viewset_templates_guard_writes() {
let live =
|body: &str, needle: &str| body.lines().any(|l| l.trim_start().starts_with(needle));
let tenant = viewset_template_tenant(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango",
);
assert!(
live(&tenant, ".permissions_for_model::<Post>()"),
"{tenant}"
);
let pool = viewset_template_pool(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango",
"app",
);
assert!(live(&pool, "read_only,"), "{pool}");
assert!(
live(&tenant, "ViewSet::for_model(Post::SCHEMA, &[\"id\"])"),
"{tenant}"
);
assert!(live(&pool, "fields = \"id"), "{pool}");
}
#[test]
fn viewset_template_pool_threads_renamed_crate_root() {
let body = viewset_template_pool(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango_orm",
"app",
);
assert!(
body.contains("use rustango_orm::ViewSet;"),
"renamed crate root must propagate to `use` line, got: {body}"
);
assert!(
!body.contains("use rustango::"),
"no bare `use rustango::` should remain after rename, got: {body}"
);
assert!(
body.contains("#[viewset("),
"derive attribute name stays literal across renames, got: {body}"
);
}
#[test]
fn viewset_template_tenant_threads_renamed_crate_root() {
let body = viewset_template_tenant(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango_orm",
);
assert!(body.contains("use rustango_orm::core::Model as _;"));
assert!(body.contains("use rustango_orm::viewset::ViewSet;"));
assert!(!body.contains("use rustango::"));
}
#[test]
fn api_routes_template_pool_threads_renamed_crate_root() {
let body = api_routes_template_pool("blog", "rustango_orm");
assert!(body.contains("use rustango_orm::sql::Pool;"));
assert!(!body.contains("use rustango::"));
}
#[test]
fn viewset_template_tenant_uses_tenant_router() {
let body = viewset_template_tenant(
"PostViewSet",
&ModelRef::bare("Post"),
"post_view_set",
"rustango",
);
assert!(
body.contains("ViewSet::for_model"),
"expected runtime ViewSet::for_model builder, got: {body}"
);
assert!(
body.contains(".tenant_router("),
"expected `.tenant_router(...)` call, got: {body}"
);
assert!(
!body.contains("#[derive(ViewSet)]"),
"tenant template must NOT use the derive macro (pool-coupled), got: {body}"
);
assert!(
body.contains("/api/post_view_set"),
"expected snake-cased path, got: {body}"
);
assert!(
body.contains("pub fn router()"),
"expected `pub fn router()` so api_routes.rs can `.merge(...)`, got: {body}"
);
assert!(
!body.contains("v1 scope"),
"v0.30 unification removed the v1 scope caveat — \
template must reflect full feature parity, got: {body}"
);
for knob in [
".filter_fields(",
".search_fields(",
".ordering(",
".page_size(",
".permissions_for_model::",
] {
assert!(
body.contains(knob),
"expected `{knob}` in tenant template (commented hint), got: {body}"
);
}
}
fn cwd_lock() -> &'static std::sync::Mutex<()> {
static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
LOCK.get_or_init(|| std::sync::Mutex::new(()))
}
#[test]
fn package_name_reads_the_package_table() {
let manifest = r#"[package]
name = "my-shop"
version = "0.1.0"
[dependencies]
name = "not-this-one"
rustango = { version = "0.61", features = ["batteries"] }
"#;
assert_eq!(
package_name_from_cargo_toml(manifest).as_deref(),
Some("my_shop"),
"must read [package].name and fold hyphens, ignoring `name` keys \
in any other table"
);
}
#[test]
fn package_name_tolerates_spacing_and_missing_tables() {
assert_eq!(
package_name_from_cargo_toml("[package]\n name = \"probe\"\n").as_deref(),
Some("probe")
);
assert_eq!(
package_name_from_cargo_toml("[dependencies]\nname = \"x\"\n"),
None,
"no [package] table → no name to report"
);
assert_eq!(package_name_from_cargo_toml(""), None);
}
#[test]
fn project_uses_tenancy_detects_inline_features_array() {
let _guard = cwd_lock().lock().unwrap_or_else(|p| p.into_inner());
let dir = tempfile::tempdir().unwrap();
let cargo = dir.path().join("Cargo.toml");
std::fs::write(
&cargo,
r#"[package]
name = "demo"
version = "0.1.0"
edition = "2021"
[dependencies]
rustango = { version = "0.30", features = ["tenancy", "manage"] }
"#,
)
.unwrap();
let prev = std::env::current_dir().unwrap();
std::env::set_current_dir(dir.path()).unwrap();
let detected = project_uses_tenancy();
let _ = std::env::set_current_dir(&prev);
assert!(
detected,
"inline-table dep with `tenancy` in features should auto-detect"
);
}
#[test]
fn project_uses_tenancy_false_when_feature_absent() {
let _guard = cwd_lock().lock().unwrap_or_else(|p| p.into_inner());
let dir = tempfile::tempdir().unwrap();
let cargo = dir.path().join("Cargo.toml");
std::fs::write(
&cargo,
r#"[package]
name = "demo"
version = "0.1.0"
edition = "2021"
[dependencies]
rustango = { version = "0.30", features = ["postgres", "manage"] }
"#,
)
.unwrap();
let prev = std::env::current_dir().unwrap();
std::env::set_current_dir(dir.path()).unwrap();
let detected = project_uses_tenancy();
let _ = std::env::set_current_dir(&prev);
assert!(
!detected,
"no tenancy feature → must default to single-tenant scaffold"
);
}
#[test]
fn check_dirs_scans_apps_beside_a_bare_migrations_dir() {
let _guard = cwd_lock().lock().unwrap_or_else(|p| p.into_inner());
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir_all(dir.path().join("migrations")).unwrap();
std::fs::create_dir_all(dir.path().join("blog/migrations")).unwrap();
let prev = std::env::current_dir().unwrap();
std::env::set_current_dir(dir.path()).unwrap();
let labels: Vec<String> = check_dirs(Path::new("migrations"), true)
.into_iter()
.map(|(app, _)| app)
.collect();
let _ = std::env::set_current_dir(&prev);
assert_eq!(labels, ["", "blog/"]);
}
#[test]
fn project_uses_tenancy_false_when_cargo_toml_missing() {
let _guard = cwd_lock().lock().unwrap_or_else(|p| p.into_inner());
let dir = tempfile::tempdir().unwrap();
let prev = std::env::current_dir().unwrap();
std::env::set_current_dir(dir.path()).unwrap();
let detected = project_uses_tenancy();
let _ = std::env::set_current_dir(&prev);
assert!(!detected);
}
#[test]
fn resolve_viewset_tenant_mode_decision_table() {
let cases: &[(bool, bool, bool, bool, bool)] = &[
(false, false, false, false, false),
(false, false, true, true, true),
(true, false, false, true, false),
(true, false, true, true, false),
(false, true, false, false, false),
(false, true, true, false, false),
(true, true, true, false, false),
];
for &(et, ent, pt, want_tenant, want_echo) in cases {
let (tenant, echo) = resolve_viewset_tenant_mode(et, ent, pt);
assert_eq!(
(tenant, echo),
(want_tenant, want_echo),
"case (explicit_tenant={et}, explicit_no_tenant={ent}, project_tenancy={pt})"
);
}
}
#[test]
fn api_routes_template_tenant_emits_no_arg_fn() {
let body = api_routes_template_tenant("regions");
assert!(
body.contains("pub fn api() -> Router<()>"),
"expected no-arg `pub fn api()`, got: {body}"
);
assert!(
body.contains(".merge(crate::<snake>::router())"),
"expected a make:viewset --tenant mount hint, got: {body}"
);
assert!(
!body.contains("PgPool"),
"tenant template must NOT thread PgPool through api(), got: {body}"
);
}
#[test]
fn api_routes_template_pool_threads_a_dialect_agnostic_pool() {
let body = api_routes_template_pool("blog", "rustango");
assert!(
body.contains("pub fn api(pool: Pool) -> Router<()>"),
"expected pool-arg api fn, got: {body}"
);
assert!(
body.contains("use rustango::sql::Pool;"),
"default crate root must emit `use rustango::sql::Pool;`, got: {body}"
);
let code: String = body
.lines()
.map(|l| match l.find("//") {
Some(i) => &l[..i],
None => l,
})
.collect::<Vec<_>>()
.join("\n");
assert!(
!code.contains("PgPool"),
"a generated project must not be pinned to Postgres here, got: {code}"
);
}
#[test]
fn is_valid_app_name_snake_case_only() {
assert!(is_valid_app_name("regions"));
assert!(is_valid_app_name("blog_posts"));
assert!(is_valid_app_name("_internal"));
assert!(!is_valid_app_name(""));
assert!(!is_valid_app_name("Regions"));
assert!(!is_valid_app_name("region-app"));
assert!(!is_valid_app_name("9_apps"));
}
#[cfg(feature = "cache")]
#[test]
fn deploy_audit_flags_a_process_local_lockout_only() {
use crate::account_lockout::Lockout;
use std::sync::Arc;
let mut out = DeployAuditFindings::default();
login_store_audit(
&Lockout::new(Arc::new(crate::cache::InMemoryCache::new())),
&mut out,
);
assert!(
out.info.iter().any(|w| w.contains("account lockout")),
"{out:?}"
);
assert!(
!out.warnings.iter().any(|w| w.contains("account lockout")),
"{out:?}"
);
let scoped = crate::cache::ScopedCache::for_tenant(
Arc::new(crate::cache::InMemoryCache::new()),
"acme",
);
assert!(Lockout::new(Arc::new(scoped)).is_process_local());
let mut out = DeployAuditFindings::default();
login_store_audit(&Lockout::new(Arc::new(SharedStandIn)), &mut out);
assert!(
!out.info.iter().any(|w| w.contains("account lockout")),
"{out:?}"
);
let mut out = DeployAuditFindings::default();
login_store_audit(&Lockout::new(Arc::new(crate::cache::NullCache)), &mut out);
assert!(
out.warnings.iter().any(|w| w.contains("stores nothing")),
"{out:?}"
);
let scoped =
crate::cache::ScopedCache::for_tenant(Arc::new(crate::cache::NullCache), "acme");
let mut out = DeployAuditFindings::default();
login_store_audit(&Lockout::new(Arc::new(scoped)), &mut out);
assert!(
out.warnings.iter().any(|w| w.contains("stores nothing")),
"{out:?}"
);
}
#[cfg(feature = "cache")]
struct SharedStandIn;
#[cfg(feature = "cache")]
#[async_trait::async_trait]
impl crate::cache::Cache for SharedStandIn {
async fn get(&self, _: &str) -> Result<Option<String>, crate::cache::CacheError> {
Ok(None)
}
async fn set(
&self,
_: &str,
_: &str,
_: Option<std::time::Duration>,
) -> Result<(), crate::cache::CacheError> {
Ok(())
}
async fn delete(&self, _: &str) -> Result<(), crate::cache::CacheError> {
Ok(())
}
async fn exists(&self, _: &str) -> Result<bool, crate::cache::CacheError> {
Ok(false)
}
async fn clear(&self) -> Result<(), crate::cache::CacheError> {
Ok(())
}
}
#[cfg(all(feature = "cache", feature = "sqlite"))]
#[tokio::test]
async fn check_deploy_notes_the_default_in_memory_lockout() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let mut buf: Vec<u8> = Vec::new();
let _ = check_cmd(
&pool,
Path::new("/nonexistent"),
&["--deploy".into()],
&mut buf,
)
.await;
let s = String::from_utf8(buf).unwrap();
assert!(s.contains("[info] account lockout"), "{s}");
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn check_refuses_an_unknown_argument() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let mut buf: Vec<u8> = Vec::new();
let err = check_cmd(
&pool,
Path::new("/nonexistent"),
&["--deplyo".into()],
&mut buf,
)
.await
.expect_err("--deplyo must fail");
assert!(err.to_string().contains("--deplyo"), "{err}");
}
#[cfg(feature = "sqlite")]
fn create_mig(name: &str, prev: Option<&str>, table: &str) -> Migration {
Migration {
name: name.to_owned(),
created_at: "2026-10-14T00:00:00Z".into(),
prev: prev.map(str::to_owned),
atomic: true,
scope: super::super::MigrationScope::Tenant,
replaces: Vec::new(),
merges: Vec::new(),
snapshot: serde_json::from_value(serde_json::json!({
"tables": [{"name": table, "model": "T",
"fields": [{"name": "id", "column": "id", "ty": "i64",
"nullable": false, "primary_key": true}]}]
}))
.unwrap(),
forward: vec![Operation::Schema(super::super::SchemaChange::CreateTable(
table.into(),
))],
}
}
#[cfg(feature = "sqlite")]
async fn manage_out(pool: &Pool, dir: &Path, args: &[&str]) -> Result<String, MigrateError> {
let mut buf: Vec<u8> = Vec::new();
let args: Vec<String> = args.iter().map(|s| (*s).to_owned()).collect();
run_with_writer(pool, dir, args, &mut buf).await?;
Ok(String::from_utf8(buf).unwrap())
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn previews_cover_the_system_chain() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"]).await;
let out = out.unwrap();
assert!(out.contains("not on disk yet"), "{out}");
manage_out(&pool, &dir, &["migrate"])
.await
.expect("migrate");
let step = file::list_dir(&root.path().join("system").join("migrations"))
.unwrap()
.into_iter()
.find(|m| m.scope == super::super::MigrationScope::Tenant)
.expect("a system step");
crate::sql::raw_execute_pool(
&pool,
"DELETE FROM __rustango_system_migrations__ WHERE name = ?",
vec![crate::core::SqlValue::String(step.name.clone())],
)
.await
.unwrap();
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"]).await;
let out = out.unwrap();
assert!(out.contains(&format!("-- {} (system,", step.name)), "{out}");
let out = manage_out(&pool, &dir, &["showmigrations"]).await;
let out = out.unwrap();
assert!(out.contains(&format!("[ ] {}", step.name)), "{out}");
let out = manage_out(&pool, &dir, &["sqlmigrate", &step.name]).await;
assert!(out.expect("sqlmigrate").contains(&step.name));
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn dry_run_leaves_project_owned_tables_to_the_project() {
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
let scratch = Pool::connect("sqlite::memory:").await.unwrap();
manage_out(&scratch, &dir, &["migrate"])
.await
.expect("migrate");
let table = file::list_dir(&root.path().join("system").join("migrations"))
.unwrap()
.into_iter()
.filter(|m| m.scope == super::super::MigrationScope::Tenant)
.find_map(|m| {
m.forward.iter().find_map(|op| match op {
Operation::Schema(super::super::SchemaChange::CreateTable(t)) => {
Some(t.clone())
}
_ => None,
})
})
.expect("a system table");
file::write(
&dir.join("0001_own.json"),
&create_mig("0001_own", None, &table),
)
.unwrap();
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"]).await;
let out = out.unwrap();
let creates = out
.lines()
.filter(|l| l.starts_with("CREATE TABLE") && l.contains(&format!("\"{table}\"")))
.count();
assert_eq!(creates, 1, "{out}");
assert!(out.contains("(system,"), "{out}");
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn dry_run_shows_bootstrap_ddl() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
let creates_translations = |out: &str| {
out.lines()
.any(|l| l.starts_with("CREATE TABLE") && l.contains("rustango_translations"))
};
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"])
.await
.unwrap();
assert!(creates_translations(&out), "{out}");
let tables: Vec<(i64,)> =
crate::sql::raw_query_pool("SELECT COUNT(*) FROM sqlite_master", Vec::new(), &pool)
.await
.unwrap();
assert_eq!(tables[0].0, 0, "the dry run wrote to the database");
manage_out(&pool, &dir, &["migrate"])
.await
.expect("migrate");
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"])
.await
.unwrap();
assert!(!creates_translations(&out), "{out}");
assert!(out.contains("nothing to migrate"), "{out}");
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn dry_run_shows_the_system_step_migrate_generates() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"])
.await
.unwrap();
assert!(out.contains("(system,"), "{out}");
assert!(
!root.path().join("system").exists(),
"a preview writes no system files"
);
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn dry_run_shows_the_converge_of_project_owned_tables() {
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
let scratch = Pool::connect("sqlite::memory:").await.unwrap();
manage_out(&scratch, &dir, &["migrate"])
.await
.expect("migrate");
let (table, column) = file::list_dir(&root.path().join("system").join("migrations"))
.unwrap()
.into_iter()
.filter(|m| m.scope == super::super::MigrationScope::Tenant)
.find_map(|m| {
m.forward.iter().find_map(|op| match op {
Operation::Schema(super::super::SchemaChange::CreateTable(t)) => {
let fields = &m.snapshot.table(t)?.fields;
let id = fields.iter().find(|f| f.primary_key)?;
let other = fields.iter().find(|f| !f.primary_key)?;
(id.column == "id").then(|| (t.clone(), other.column.clone()))
}
_ => None,
})
})
.expect("a system table with an id pk");
file::write(
&dir.join("0001_own.json"),
&create_mig("0001_own", None, &table),
)
.unwrap();
let pool = Pool::connect("sqlite::memory:").await.unwrap();
runner::migrate_pool(&pool, &dir)
.await
.expect("project chain");
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"])
.await
.unwrap();
let adds = format!("ALTER TABLE \"{table}\" ADD COLUMN \"{column}\"");
assert!(out.contains(&adds), "{adds} missing from:\n{out}");
}
#[test]
fn fixture_writer_buffers() {
struct Counting(usize, Vec<u8>);
impl Write for Counting {
fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
self.0 += 1;
self.1.extend_from_slice(b);
Ok(b.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
let mut sink = Counting(0, Vec::new());
let mut out = FixtureWriter::new(&mut sink, 2);
for pk in 0..100 {
out.entry(&serde_json::json!({"pk": pk})).unwrap();
}
out.finish().unwrap();
let text = String::from_utf8(sink.1).unwrap();
assert_eq!(
serde_json::from_str::<serde_json::Value>(&text).unwrap()[99]["pk"],
99
);
assert!(sink.0 < 10, "{} writes for 100 entries", sink.0);
}
#[test]
fn fixture_writer_reports_a_failed_flush() {
struct Full;
impl Write for Full {
fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
Err(std::io::Error::other("disk full"))
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
let mut sink = Full;
let mut out = FixtureWriter::new(&mut sink, 0);
out.entry(&serde_json::json!({"pk": 1})).unwrap();
let err = out.finish().expect_err("lost write error");
assert!(err.to_string().contains("disk full"), "{err}");
}
#[allow(dead_code)]
#[derive(crate::Model, Debug)]
#[rustango(
table = "dl1399_post",
m2m(
name = "tags",
to = "dl1399_tag",
through = "dl1399_post_tags",
src = "post_id",
dst = "tag_id"
)
)]
pub struct DlPost {
#[rustango(primary_key)]
pub id: i64,
}
#[allow(dead_code)]
#[derive(crate::Model, Debug)]
#[rustango(table = "dl1399_tag")]
pub struct DlTag {
#[rustango(primary_key)]
pub id: i64,
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn m2m_links_reads_only_the_page() {
use crate::core::Model as _;
let pool = Pool::connect("sqlite::memory:").await.unwrap();
for sql in [
"CREATE TABLE dl1399_post_tags (post_id BIGINT NOT NULL, tag_id BIGINT NOT NULL)",
"INSERT INTO dl1399_post_tags VALUES (1, 10), (2, 20)",
] {
crate::sql::raw_execute_pool(&pool, sql, Vec::new())
.await
.unwrap();
}
let page = [crate::core::SqlValue::I64(1)];
let links = m2m_links(&pool, DlPost::SCHEMA, &page).await.unwrap();
let [(_, by_src)] = links.as_slice() else {
panic!("one relation: {links:?}");
};
let srcs: Vec<&String> = by_src.keys().collect();
assert_eq!(srcs, ["1"], "read links of rows outside the page");
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn a_squashed_file_is_not_pending() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
let first = create_mig("0001_a", None, "sq1311_a");
file::write(&dir.join("0001_a.json"), &first).unwrap();
manage_out(&pool, &dir, &["migrate"])
.await
.expect("migrate");
let mut squash = create_mig("0002_sq", Some("0001_a"), "sq1311_a");
squash.replaces = vec!["0001_a".into()];
file::write(&dir.join("0002_sq.json"), &squash).unwrap();
manage_out(&pool, &dir, &["migrate"]).await.expect("squash");
let out = manage_out(&pool, &dir, &["showmigrations"]).await;
let out = out.unwrap();
assert!(out.contains("[X] 0001_a"), "{out}");
let out = manage_out(&pool, &dir, &["migrate", "--dry-run"]).await;
let out = out.unwrap();
assert!(!out.contains("0001_a"), "{out}");
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn squash_refuses_hand_written_ops() {
use super::super::{CallbackOp, SchemaChange};
let hand = [
Operation::Data(DataOp {
sql: "UPDATE t SET x = 1".into(),
reverse_sql: None,
reversible: false,
}),
Operation::Callback(CallbackOp {
name: "fill".into(),
reverse_name: None,
}),
Operation::Schema(SchemaChange::RenameTable {
old_name: "sq_a".into(),
new_name: "sq_b".into(),
}),
Operation::Schema(SchemaChange::RenameColumn {
table: "sq_a".into(),
old_column: "x".into(),
new_column: "y".into(),
}),
];
for op in hand {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let root = tempfile::tempdir().unwrap();
let dir = root.path().join("migrations");
std::fs::create_dir_all(&dir).unwrap();
file::write(
&dir.join("0001_a.json"),
&create_mig("0001_a", None, "sq_a"),
)
.unwrap();
let mut second = create_mig("0002_b", Some("0001_a"), "sq_a");
second.atomic = false;
second.forward = vec![op.clone()];
file::write(&dir.join("0002_b.json"), &second).unwrap();
let err = manage_out(&pool, &dir, &["migrate", "--squash"])
.await
.expect_err("a re-diff would lose the op");
assert!(err.to_string().contains("0002_b"), "{op:?}: {err}");
assert!(!err.to_string().contains("UPDATE"), "the SQL leaked: {err}");
assert!(
dir.join("0002_b.json").exists(),
"{op:?}: the file was deleted"
);
}
}
#[test]
fn missing_time_zone_tables_warn() {
assert!(time_zone_warning(None).is_some());
assert!(time_zone_warning(Some("2000-01-01 00:00:00")).is_none());
}
#[test]
fn ci_and_bin_collations_warn() {
assert!(collation_warning("utf8mb4_0900_ai_ci").is_some());
assert!(collation_warning("utf8mb4_general_ci").is_some());
assert!(collation_warning("utf8mb4_0900_as_cs").is_none());
assert!(collation_warning("utf8mb4_bin").is_some());
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn check_deploy_has_no_collation_warning_on_sqlite() {
let pool = Pool::connect("sqlite::memory:").await.unwrap();
let mut buf: Vec<u8> = Vec::new();
let _ = check_cmd(
&pool,
Path::new("/nonexistent"),
&["--deploy".into()],
&mut buf,
)
.await;
let s = String::from_utf8(buf).unwrap();
assert!(!s.contains("collation"), "{s}");
}
fn good_prod_env() -> DeployAuditEnv {
DeployAuditEnv {
rustango_env: Some("prod".into()),
session_secret: Some("a".repeat(48)),
database_url: Some("postgres://app:s3cr3t@db.example.com/app_prod".into()),
database_style_conflict: None,
apex_domain: ApexDomain::Tenancy(Some("app.example.com".into())),
bind: Some("0.0.0.0:8080".into()),
ungated_admin: false,
}
}
#[test]
fn deploy_audit_says_nothing_about_the_tenant_scheme() {
let r = run(&good_prod_env());
assert!(
!r.warnings
.iter()
.any(|w| w.contains("RUSTANGO_TENANT_SCHEME")),
"{:?}",
r.warnings
);
}
#[test]
fn deploy_audit_warns_on_an_ungated_admin() {
let env = DeployAuditEnv {
ungated_admin: true,
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings.iter().any(|w| w.contains("with_session_auth")),
"{:?}",
r.warnings
);
}
#[cfg(all(feature = "admin", feature = "sqlite"))]
#[tokio::test]
async fn only_an_ungated_admin_reaches_the_deploy_audit() {
let _g = crate::admin::ungated_flag_lock().lock().await;
crate::admin::reset_ungated_admin_built();
let pool = crate::sql::Pool::connect("sqlite::memory:")
.await
.expect("sqlite");
let secret = crate::session::SessionSecret::from_bytes(vec![7; 32]);
let _gated = crate::admin::Builder::new(pool.clone())
.with_session_auth(secret)
.build();
let gated_clear = !deploy_audit_env(false).ungated_admin;
let _open = crate::admin::Builder::new(pool.clone())
.tenant_mode()
.build();
let flagged = deploy_audit_env(false).ungated_admin;
crate::admin::reset_ungated_admin_built();
assert!(gated_clear, "a gated admin set the ungated flag");
assert!(
flagged,
"a `tenant_mode` admin without auth was not flagged"
);
}
fn run(env: &DeployAuditEnv) -> DeployAuditFindings {
let mut out = DeployAuditFindings::default();
run_deploy_audit(env, &mut out);
out
}
#[test]
fn deploy_audit_clean_prod_env_has_no_warnings_or_errors() {
let r = run(&good_prod_env());
assert!(
r.errors.is_empty(),
"expected no errors in clean prod env, got: {:?}",
r.errors
);
assert!(
r.warnings.is_empty(),
"expected no warnings in clean prod env, got: {:?}",
r.warnings
);
}
#[test]
fn deploy_audit_unset_session_secret_errors() {
let env = DeployAuditEnv {
session_secret: None,
..good_prod_env()
};
let r = run(&env);
assert!(
r.errors
.iter()
.any(|e| e.contains("RUSTANGO_SESSION_SECRET")),
"expected error for unset RUSTANGO_SESSION_SECRET, got: {:?}",
r.errors
);
}
#[test]
fn deploy_audit_unset_session_secret_says_what_happens() {
let env = DeployAuditEnv {
session_secret: None,
..good_prod_env()
};
let r = run(&env);
let e = r
.errors
.iter()
.find(|e| e.contains("RUSTANGO_SESSION_SECRET is unset"))
.expect("unset error");
assert!(e.contains("will not start") && e.contains("./var/"), "{e}");
assert!(!e.contains("ephemeral"), "{e}");
assert_eq!(e.contains("tenant"), cfg!(feature = "tenancy"), "{e}");
}
#[cfg(any(feature = "admin", feature = "tenancy"))]
#[test]
fn deploy_audit_short_session_secret_errors() {
let env = DeployAuditEnv {
session_secret: Some("AAAAAAAA".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.errors.iter().any(|e| e.contains("decoded to 6 bytes")),
"expected a decoded-length error, got: {:?}",
r.errors
);
}
#[cfg(any(feature = "admin", feature = "tenancy"))]
#[test]
fn deploy_audit_non_base64_session_secret_errors() {
let env = DeployAuditEnv {
session_secret: Some("correct-horse-battery-staple-and-then-som".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.errors.iter().any(|e| e.contains("not valid base64")),
"a long non-base64 value is still not a key, got: {:?}",
r.errors
);
}
#[cfg(any(feature = "admin", feature = "tenancy"))]
#[test]
fn deploy_audit_rejects_a_32_char_base64_secret_the_runtime_refuses() {
let trap = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
assert_eq!(
trap.len(),
32,
"the premise: 32 characters, so the old check passed"
);
assert!(
crate::session::SessionSecret::from_b64(trap).is_err(),
"the premise: the runtime refuses it"
);
let env = DeployAuditEnv {
session_secret: Some(trap.into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.errors.iter().any(|e| e.contains("decoded to 24 bytes")),
"check --deploy must refuse what the runtime refuses, got: {:?}",
r.errors
);
assert!(
!r.info.iter().any(|i| i.contains("SESSION_SECRET")),
"and must not also report it OK, got: {:?}",
r.info
);
}
#[test]
fn deploy_audit_placeholder_session_secret_errors() {
let env = DeployAuditEnv {
session_secret: Some("change-me-base64-encoded-32-bytes-or-more".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.errors
.iter()
.any(|e| e.contains("placeholder") || e.contains("change-me")),
"expected error for unchanged placeholder secret, got: {:?}",
r.errors
);
}
#[test]
fn deploy_audit_unset_rustango_env_warns() {
let env = DeployAuditEnv {
rustango_env: None,
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings.iter().any(|w| w.contains("RUSTANGO_ENV")),
"expected warning for unset RUSTANGO_ENV, got: {:?}",
r.warnings
);
}
#[test]
fn deploy_audit_dev_rustango_env_warns() {
let env = DeployAuditEnv {
rustango_env: Some("dev".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings.iter().any(|w| w.contains("`dev`")),
"expected warning for non-prod RUSTANGO_ENV, got: {:?}",
r.warnings
);
}
#[test]
fn deploy_audit_unset_database_url_errors() {
let env = DeployAuditEnv {
database_url: None,
..good_prod_env()
};
let r = run(&env);
assert!(
r.errors.iter().any(|e| e.contains("DATABASE_URL")),
"expected error for unset DATABASE_URL, got: {:?}",
r.errors
);
}
#[test]
fn deploy_audit_reports_both_database_styles() {
let env = DeployAuditEnv {
database_style_conflict: Some("both DATABASE_URL and DB_USER are set".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings.iter().any(|w| w.contains("DB_USER")),
"{:?}",
r.warnings
);
}
#[test]
fn deploy_audit_localhost_database_url_warns() {
let env = DeployAuditEnv {
database_url: Some("postgres://app:p@localhost/db".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings
.iter()
.any(|w| w.contains("DATABASE_URL") && w.contains("localhost")),
"expected warning for localhost DATABASE_URL, got: {:?}",
r.warnings
);
}
#[test]
fn deploy_audit_single_tenant_skips_the_apex() {
let env = DeployAuditEnv {
apex_domain: ApexDomain::SingleTenant,
..good_prod_env()
};
let r = run(&env);
assert!(r.warnings.is_empty(), "{:?}", r.warnings);
assert!(!r.info.iter().any(|i| i.contains("RUSTANGO_APEX_DOMAIN")));
}
#[test]
fn deploy_audit_localhost_apex_warns_for_tenancy() {
let env = DeployAuditEnv {
apex_domain: ApexDomain::Tenancy(Some("localhost".into())),
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings
.iter()
.any(|w| w.contains("RUSTANGO_APEX_DOMAIN")),
"expected warning for localhost apex, got: {:?}",
r.warnings
);
}
#[test]
fn deploy_audit_loopback_bind_warns() {
let env = DeployAuditEnv {
bind: Some("127.0.0.1:8080".into()),
..good_prod_env()
};
let r = run(&env);
assert!(
r.warnings
.iter()
.any(|w| w.contains("RUSTANGO_BIND") && w.contains("loopback")),
"expected warning for loopback bind, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
fn settings_run(env_tier: &str, s: &crate::config::Settings) -> DeployAuditFindings {
let mut out = DeployAuditFindings::default();
settings_audit_check(env_tier, s, &mut out);
out
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_dev_tier_is_quiet() {
let mut s = crate::config::Settings::default();
s.security.headers_preset = Some("dev".into());
s.security.hsts_max_age_secs = Some(0);
s.auth.argon2_memory_kib = Some(1024); let r = settings_run("dev", &s);
assert!(
r.warnings.is_empty(),
"dev tier should be quiet, got: {:?}",
r.warnings
);
assert!(
r.info.is_empty(),
"dev tier should be quiet, got: {:?}",
r.info
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_reports_insecure_cookies_presets_and_missing_config() {
let mut s = crate::config::Settings::default();
s.security.secure_cookies = Some(false);
let r = settings_run("prod", &s);
assert!(
r.warnings.iter().any(|w| w.contains("secure_cookies")),
"{:?}",
r.warnings
);
for preset in ["empty", "strixt"] {
let mut s = crate::config::Settings::default();
s.security.headers_preset = Some(preset.into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("headers_preset") && w.contains(preset)),
"{preset}: {:?}",
r.warnings
);
}
let mut s = crate::config::Settings::default();
s.security.headers_preset = Some("relaxed".into());
let r = settings_run("prod", &s);
assert!(
!r.warnings.iter().any(|w| w.contains("headers_preset")),
"{:?}",
r.warnings
);
let empty = tempfile::tempdir().unwrap();
let missing = crate::config::Settings::load_from(empty.path(), "prod");
let mut out = DeployAuditFindings::default();
settings_audit_loaded("prod", missing, &mut out);
assert!(
out.warnings.iter().any(|w| w.contains("default.toml")),
"{:?}",
out.warnings
);
let missing = crate::config::Settings::load_from(empty.path(), "dev");
let mut out = DeployAuditFindings::default();
settings_audit_loaded("dev", missing, &mut out);
assert!(out.warnings.is_empty(), "{:?}", out.warnings);
}
#[cfg(feature = "config")]
#[test]
fn deploy_audit_treats_every_prod_spelling_as_prod() {
let mut s = crate::config::Settings::default();
s.security.hsts_max_age_secs = Some(0);
for tier in ["PROD", "Production", " prod "] {
let r = settings_run(tier, &s);
assert!(
r.warnings.iter().any(|w| w.contains("hsts_max_age_secs")),
"{tier:?}: {:?}",
r.warnings
);
let mut env = good_prod_env();
env.rustango_env = Some(tier.into());
let r = run(&env);
assert!(r.warnings.is_empty(), "{tier:?}: {:?}", r.warnings);
}
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_prod_with_dev_headers_preset_warns() {
let mut s = crate::config::Settings::default();
s.security.headers_preset = Some("dev".into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("headers_preset") && w.contains("dev")),
"expected warning for dev headers in prod, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_prod_with_zero_hsts_warns() {
let mut s = crate::config::Settings::default();
s.security.hsts_max_age_secs = Some(0);
let r = settings_run("prod", &s);
assert!(
r.warnings.iter().any(|w| w.contains("hsts_max_age_secs")),
"expected HSTS warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_prod_low_argon2_warns_but_unset_is_quiet() {
let mut s = crate::config::Settings::default();
s.auth.argon2_memory_kib = Some(4096); let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("argon2_memory_kib") && w.contains("19456")),
"expected argon2 floor warning, got: {:?}",
r.warnings
);
let s_default = crate::config::Settings::default();
let r = settings_run("prod", &s_default);
assert!(
!r.warnings.iter().any(|w| w.contains("argon2")),
"default argon2 should be quiet, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_prod_long_jwt_access_ttl_warns() {
let mut s = crate::config::Settings::default();
s.auth.jwt.access_ttl_secs = Some(86400); let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("access_ttl_secs") && w.contains("86400")),
"expected access TTL warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_prod_loopback_bind_warns() {
let mut s = crate::config::Settings::default();
s.server.bind = Some("127.0.0.1:8080".into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("[server] bind") && w.contains("loopback")),
"expected loopback warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_legacy_preset_and_unset_retention_are_info() {
let mut s = crate::config::Settings::default();
s.routes.legacy_preset = Some(true);
s.security.allowed_hosts = vec!["example.com".into()];
s.mail.backend = Some("smtp".into());
s.server.request_timeout_secs = Some(30);
let r = settings_run("prod", &s);
assert!(
r.info.iter().any(|i| i.contains("legacy_preset")),
"expected legacy_preset info, got: {:?}",
r.info
);
assert!(
r.info.iter().any(|i| i.contains("retention_days")),
"expected retention_days info, got: {:?}",
r.info
);
assert!(
r.warnings.is_empty(),
"neither should be warnings, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_empty_trusted_origins_does_not_claim_check_disabled() {
let s = crate::config::Settings::default();
let r = settings_run("prod", &s);
let line = r
.info
.iter()
.find(|i| i.contains("csrf_trusted_origins"))
.expect("csrf_trusted_origins info line");
assert!(!line.contains("disabled"), "{line}");
assert!(line.contains("own Host"), "{line}");
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_csrf_insecure_warns_in_prod() {
let mut s = crate::config::Settings::default();
s.admin.csrf_cookie_secure = Some(false);
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("[admin]") && w.contains("csrf_cookie_secure")),
"expected CSRF secure warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_bad_primary_color_warns() {
let mut s = crate::config::Settings::default();
s.admin.primary_color = Some("not-a-color".into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("[admin]") && w.contains("primary_color")),
"expected primary_color warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_valid_hex_color_is_quiet() {
for hex in ["#abc", "#2c6fb0", "#2c6fb0ff"] {
let mut s = crate::config::Settings::default();
s.admin.primary_color = Some(hex.into());
let r = settings_run("prod", &s);
assert!(
!r.warnings
.iter()
.any(|w| w.contains("[admin]") && w.contains("primary_color")),
"expected `{hex}` to be quiet, got: {:?}",
r.warnings
);
}
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_unknown_theme_mode_warns() {
let mut s = crate::config::Settings::default();
s.admin.theme_mode = Some("midnight".into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("[admin]") && w.contains("theme_mode")),
"expected theme_mode warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_zero_session_timeout_is_info() {
let mut s = crate::config::Settings::default();
s.admin.session_timeout_minutes = Some(0);
let r = settings_run("prod", &s);
assert!(
r.info
.iter()
.any(|i| i.contains("[admin]") && i.contains("session_timeout_minutes")),
"expected session_timeout_minutes info, got: {:?}",
r.info
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_trailing_slash_url_prefix_warns() {
let mut s = crate::config::Settings::default();
s.admin.url_prefix = Some("/admin/".into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("[admin]") && w.contains("url_prefix")),
"expected url_prefix warning, got: {:?}",
r.warnings
);
}
#[cfg(feature = "config")]
#[test]
fn settings_audit_admin_empty_url_prefix_is_quiet() {
let mut s = crate::config::Settings::default();
s.admin.url_prefix = Some("".into());
let r = settings_run("prod", &s);
assert!(
!r.warnings
.iter()
.any(|w| w.contains("[admin]") && w.contains("url_prefix")),
"empty url_prefix should be quiet, got: {:?}",
r.warnings
);
}
#[cfg(all(
feature = "config",
feature = "tenancy",
feature = "postgres",
any(feature = "sqlite", feature = "mysql"),
))]
#[test]
fn settings_audit_sqlite_backend_with_tenancy_warns_in_prod() {
let mut s = crate::config::Settings::default();
s.database.backend = Some("sqlite".into());
let r = settings_run("prod", &s);
assert!(
r.warnings
.iter()
.any(|w| w.contains("[database]") && w.contains("tenancy")),
"expected backend × tenancy warning, got: {:?}",
r.warnings
);
}
}
#[cfg(test)]
mod db_cmd_tests {
use super::*;
fn args(s: &[&str]) -> Vec<String> {
s.iter().map(|x| (*x).to_owned()).collect()
}
#[test]
fn dump_no_flags_defaults() {
let p = parse_db_dump_args(&[]).unwrap();
assert!(p.out.is_none());
assert!(!p.data_only);
assert!(!p.schema_only);
assert!(!p.no_owner);
}
#[test]
fn dump_out_flag_with_value() {
let p = parse_db_dump_args(&args(&["--out", "/tmp/db.sql"])).unwrap();
assert_eq!(p.out.as_deref(), Some("/tmp/db.sql"));
}
#[test]
fn dump_short_o_flag() {
let p = parse_db_dump_args(&args(&["-o", "/tmp/db.sql"])).unwrap();
assert_eq!(p.out.as_deref(), Some("/tmp/db.sql"));
}
#[test]
fn dump_data_only_flag() {
let p = parse_db_dump_args(&args(&["--data-only"])).unwrap();
assert!(p.data_only);
assert!(!p.schema_only);
}
#[test]
fn dump_schema_only_flag() {
let p = parse_db_dump_args(&args(&["--schema-only"])).unwrap();
assert!(p.schema_only);
assert!(!p.data_only);
}
#[test]
fn dump_no_owner_flag() {
let p = parse_db_dump_args(&args(&["--no-owner"])).unwrap();
assert!(p.no_owner);
}
#[test]
fn dump_out_without_value_errors() {
let r = parse_db_dump_args(&args(&["--out"]));
assert!(r.is_err());
}
#[test]
fn dump_data_and_schema_only_conflict() {
let r = parse_db_dump_args(&args(&["--data-only", "--schema-only"]));
assert!(r.is_err());
}
#[test]
fn dump_unknown_flag_errors() {
let r = parse_db_dump_args(&args(&["--bogus"]));
assert!(r.is_err());
}
#[test]
fn dump_argv_starts_with_a_password_free_dbname() {
let parsed = DbDumpArgs {
out: None,
data_only: false,
schema_only: false,
no_owner: false,
};
let cmd = pg_dump_command(&parsed, "postgres://u:s3cret@h/db").unwrap();
let argv: Vec<_> = cmd.get_args().map(|a| a.to_string_lossy()).collect();
assert_eq!(argv[..2], ["--dbname", "postgres://u@h/db"]);
}
#[test]
fn dump_and_restore_keep_the_password_out_of_argv() {
let url = "postgres://u:s3cret@h/db";
let dump = DbDumpArgs {
out: None,
data_only: false,
schema_only: false,
no_owner: false,
};
let plan = RestorePlan {
file: "/tmp/x.sql".into(),
clean: false,
};
for cmd in [
pg_dump_command(&dump, url).unwrap(),
psql_restore_command("psql", &plan, url).unwrap(),
] {
assert!(
!cmd.get_args()
.any(|a| a.to_string_lossy().contains("s3cret")),
"{cmd:?}"
);
let env: Vec<_> = cmd.get_envs().collect();
assert_eq!(
env,
[(
std::ffi::OsStr::new("PGPASSWORD"),
Some(std::ffi::OsStr::new("s3cret"))
)]
);
}
}
#[test]
fn dump_argv_includes_chosen_flags() {
let parsed = DbDumpArgs {
out: Some("/tmp/x.sql".into()),
data_only: true,
schema_only: false,
no_owner: true,
};
let argv = build_pg_dump_argv(&parsed);
assert!(argv.contains(&"--data-only".to_owned()));
assert!(argv.contains(&"--no-owner".to_owned()));
assert!(argv.contains(&"--file".to_owned()));
assert!(argv.contains(&"/tmp/x.sql".to_owned()));
assert!(!argv.contains(&"--schema-only".to_owned()));
}
#[test]
fn restore_requires_file() {
let r = parse_db_restore_args(&[]);
assert!(r.is_err());
}
#[test]
fn restore_positional_file() {
let p = parse_db_restore_args(&args(&["/tmp/db.sql"])).unwrap();
assert_eq!(p.file, "/tmp/db.sql");
assert!(!p.clean);
}
#[test]
fn restore_with_clean_flag() {
let p = parse_db_restore_args(&args(&["--clean", "/tmp/db.sql"])).unwrap();
assert!(p.clean);
assert_eq!(p.file, "/tmp/db.sql");
}
#[test]
fn restore_clean_after_file() {
let p = parse_db_restore_args(&args(&["/tmp/db.sql", "--clean"])).unwrap();
assert!(p.clean);
}
#[test]
fn restore_two_files_errors() {
let r = parse_db_restore_args(&args(&["a.sql", "b.sql"]));
assert!(r.is_err());
}
fn restore_args(file: &str, clean: bool, yes: bool) -> DbRestoreArgs {
DbRestoreArgs {
file: file.into(),
clean,
yes,
}
}
#[test]
fn restore_refuses_a_missing_file_before_asking() {
let dir = tempfile::tempdir().unwrap();
let missing = dir.path().join("nope.sql");
let missing = missing.to_str().unwrap();
let err = RestorePlan::check(restore_args(missing, true, false), |_| {
panic!("asked to confirm a file that does not exist")
})
.unwrap_err();
assert!(err.to_string().contains("cannot read"), "{err}");
let dir_path = dir.path().to_str().unwrap();
assert!(RestorePlan::check(restore_args(dir_path, true, true), |_| true).is_err());
}
#[test]
fn restore_clean_needs_yes_or_a_confirm() {
let f = tempfile::NamedTempFile::new().unwrap();
std::fs::write(f.path(), "SELECT 1;\n").unwrap();
let path = f.path().to_str().unwrap();
let err = RestorePlan::check(restore_args(path, true, false), |_| false).unwrap_err();
assert!(err.to_string().contains("--yes"), "{err}");
assert!(RestorePlan::check(restore_args(path, true, false), |_| true).is_ok());
assert!(RestorePlan::check(restore_args(path, true, true), |_| false).is_ok());
assert!(RestorePlan::check(restore_args(path, false, false), |_| false).is_ok());
let p = parse_db_restore_args(&args(&["--clean", "--yes", path])).unwrap();
assert!(p.clean && p.yes);
}
#[test]
fn restore_clean_refuses_an_empty_file() {
let f = tempfile::NamedTempFile::new().unwrap();
let path = f.path().to_str().unwrap();
let err = RestorePlan::check(restore_args(path, true, true), |_| true).unwrap_err();
assert!(err.to_string().contains("empty"), "{err}");
assert!(RestorePlan::check(restore_args(path, false, false), |_| false).is_ok());
}
#[cfg(unix)]
#[test]
fn restore_takes_a_pipe_only_without_clean() {
assert!(RestorePlan::check(restore_args("/dev/stdin", false, false), |_| false).is_ok());
let err = RestorePlan::check(restore_args("/dev/stdin", true, true), |_| true).unwrap_err();
assert!(err.to_string().contains("regular file"), "{err}");
}
#[test]
fn restore_prompt_needs_a_typed_yes() {
assert!(answered_yes(Ok(Some("yes".into()))));
for no in [Ok(Some("y".into())), Ok(Some("YES".into())), Ok(None)] {
assert!(!answered_yes(no));
}
assert!(!answered_yes(Err(std::io::Error::other("tty"))));
}
#[cfg(feature = "postgres")]
#[tokio::test]
async fn restore_clean_rolls_back_a_bad_dump_on_a_private_database() {
let Ok(url) = std::env::var("DATABASE_URL") else {
eprintln!("DATABASE_URL not set — skipping");
return;
};
let runs = |p: &str| {
std::process::Command::new(p)
.arg("--version")
.output()
.is_ok_and(|o| o.status.success())
};
let psql = std::env::var("RUSTANGO_TEST_PSQL").unwrap_or_else(|_| "psql".to_owned());
if !runs(&psql) {
assert!(
std::env::var_os("CI").is_none(),
"db:restore needs `psql` on PATH in CI"
);
eprintln!("SKIPPED: `{psql}` does not run (set RUSTANGO_TEST_PSQL); db:restore is unproven here");
return;
}
let db = format!("rustango_restore_{}", std::process::id());
let (base, _) = url.rsplit_once('/').expect("DATABASE_URL names a database");
let private = format!("{base}/{db}");
let admin = sqlx::PgPool::connect(&url).await.expect("connect");
let drop_db = format!(r#"DROP DATABASE IF EXISTS "{db}" WITH (FORCE)"#);
sqlx::query(&drop_db).execute(&admin).await.unwrap();
sqlx::query(&format!(r#"CREATE DATABASE "{db}""#))
.execute(&admin)
.await
.unwrap();
let p = sqlx::PgPool::connect(&private).await.expect("private db");
for sql in ["CREATE TABLE keep (id int)", "INSERT INTO keep VALUES (1)"] {
sqlx::query(sql).execute(&p).await.unwrap();
}
let dir = tempfile::tempdir().unwrap();
let bad = dir.path().join("bad.sql");
std::fs::write(
&bad,
"CREATE TABLE half (id int);\nSELECT * FROM no_such_table;\n",
)
.unwrap();
let plan = RestorePlan::check(restore_args(bad.to_str().unwrap(), true, true), |_| true)
.expect("plan");
let res = run_psql_restore(&psql, &plan, &private, &mut Vec::new());
let kept: Result<i64, _> = sqlx::query_scalar("SELECT COUNT(*)::bigint FROM keep")
.fetch_one(&p)
.await;
p.close().await;
sqlx::query(&drop_db).execute(&admin).await.unwrap();
assert_eq!(kept.ok(), Some(1), "the bad dump left the DROP committed");
assert!(res.unwrap_err().to_string().contains("psql exited"));
}
#[test]
fn restore_argv_includes_on_error_stop() {
let parsed = RestorePlan {
file: "/tmp/x.sql".into(),
clean: false,
};
let argv = build_psql_argv(&parsed);
assert!(argv.contains(&"ON_ERROR_STOP=1".to_owned()));
assert!(argv.contains(&"--single-transaction".to_owned()));
assert!(argv.contains(&"-f".to_owned()));
assert!(argv.contains(&"/tmp/x.sql".to_owned()));
assert!(!argv.iter().any(|a| a.contains("DROP SCHEMA")));
}
#[test]
fn restore_argv_with_clean_drops_schema() {
let parsed = RestorePlan {
file: "/tmp/x.sql".into(),
clean: true,
};
let argv = build_psql_argv(&parsed);
assert!(argv.iter().any(|a| a.contains("DROP SCHEMA")));
assert!(argv.iter().any(|a| a.contains("CREATE SCHEMA")));
}
#[test]
fn redact_masks_password_in_postgres_url() {
assert_eq!(
redact_url("postgres://alice:supersecret@localhost:5432/mydb"),
"postgres://alice:***@localhost:5432/mydb"
);
}
#[test]
fn redact_passes_through_url_without_credentials() {
assert_eq!(
redact_url("postgres://localhost:5432/mydb"),
"postgres://localhost:5432/mydb"
);
}
#[test]
fn redact_passes_through_non_urls() {
assert_eq!(redact_url("--data-only"), "--data-only");
assert_eq!(redact_url("/tmp/db.sql"), "/tmp/db.sql");
}
#[test]
fn redact_handles_url_with_only_user() {
assert_eq!(
redact_url("postgres://alice@localhost/db"),
"postgres://alice@localhost/db"
);
}
#[test]
fn about_masks_a_query_password() {
assert_eq!(
about_db_url(Some("postgres://db/app?password=x".into())),
"postgres://db/app?password=***"
);
assert_eq!(about_db_url(None), "(unset)");
}
#[test]
fn redact_masks_a_password_holding_an_at_sign() {
assert_eq!(
redact_url("postgres://alice:p@ss@localhost/db"),
"postgres://alice:***@localhost/db"
);
}
}