fix(ops): sauvegarder reellement la base PostgreSQL

DATA-002 / OPS-001. backup.py ciblait SQLite : import sqlite3, DATABASE_PATH
par defaut instance/team_tryouts.db, et l'API de sauvegarde sqlite3. La
production tourne sur PostgreSQL, donc le fichier n'existait pas. Le script
affichait "[WARNING] Database not found... Skipping database backup" -- puis,
main() ne suivant que le resultat de la verification, **sortait avec le code
0**. Toute tache planifiee surveillant le code de sortie voyait vert alors
qu'aucune sauvegarde n'avait jamais ete produite.

Il n'existait donc aucune sauvegarde applicative de la base.

Reecriture
  pg_dump en --format=custom : compresse, et pg_restore permet une
  restauration selective, ce qu'un dump SQL a plat ne permet pas.
  parse_database_url accepte les suffixes de dialecte SQLAlchemy
  (postgresql+psycopg://) que pg_dump ne comprend pas, et refuse
  explicitement une URL SQLite -- le cas exact qui passait en silence.

  Le mot de passe ne figure jamais dans la ligne de commande : il serait
  visible de tout processus capable de lister argv. Il passe par PGPASSWORD.
  Il est egalement absent des messages affiches, qui atterrissent dans les
  journaux du planificateur.

  verify_backup lit l'archive avec pg_restore --list et exige au moins une
  table : une archive illisible ne se restaure pas, et une archive sans
  table signifie que le dump a vise la mauvaise cible. Les deux sont des
  echecs silencieux qu'il vaut mieux attraper maintenant que pendant un
  incident.

  Le code de sortie vaut 0 uniquement si le dump a ete produit ET verifie.

L'archive des documents est conservee : les contrats signes n'existent que
sur disque, la base ne stocke que des chemins. Restaurer l'une sans l'autre
laisse des lignes pointant vers des fichiers absents.

docs/database-restore.md
  Procedure de restauration testable sur une base jetable, requetes de
  controle, demarrage de l'application sur la copie restauree, plan de
  reprise par scenario. ENABLE_DISCORD_BOT=false y est signale comme non
  optionnel : sans lui, l'exercice demarre un vrai bot et envoie de vraies
  notifications a de vraies personnes, a partir de donnees restaurees.

  Les points ouverts sont listes tels quels : aucune copie hors site, pas de
  chiffrement au repos, aucune planification, et l'exercice de restauration
  n'a jamais ete effectue.

17 tests sur ce qui est verifiable sans serveur PostgreSQL : analyse de
l'URL, construction de la commande, non-fuite du mot de passe, et surtout
codes de sortie -- le silence ne vaut plus succes.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
GGThed
2026-08-07 20:06:51 -04:00
co-authored by Claude Opus 5
parent ca45be80db
commit 7bf428f9a0
3 changed files with 551 additions and 108 deletions
+242 -97
View File
@@ -1,84 +1,255 @@
"""Database backup script for the Team Tryouts application. """Database and document backup for the Team Tryouts application.
This module provides a simple backup mechanism for the SQLite database Dumps the PostgreSQL database with pg_dump and archives the uploaded
and uploaded contract documents. Designed to be run as a scheduled task contract documents. Designed to be run from a scheduled task (Windows Task
(Windows Task Scheduler) or cron job. Scheduler) or a cron job.
Usage: Usage:
python backup.py python app/supporting_scripts/backup.py
python app/supporting_scripts/backup.py --verify-only <archive>
Configuration via environment variables: Configuration via environment variables:
BACKUP_DIR: Directory to store backups (default: ./backups) DATABASE_URL PostgreSQL connection string (required)
BACKUP_RETENTION_DAYS: Number of days to keep backups (default: 30) BACKUP_DIR Where to store backups (default: ./backups)
BACKUP_RETENTION_DAYS How long to keep them (default: 30)
PG_DUMP Path to pg_dump if not on PATH
PG_RESTORE Path to pg_restore if not on PATH
A note on what this file used to be
-----------------------------------
The previous version targeted **SQLite**: it imported sqlite3, read
DATABASE_PATH defaulting to instance/team_tryouts.db, and used the sqlite3
backup API. Production runs on PostgreSQL, so the file never existed, the
script printed "[WARNING] Database not found... Skipping database backup"
and — because main() only tracked the verification result — still exited 0.
It reported success while backing up nothing at all. Any scheduled task
watching the exit code saw green.
Restoring is documented in docs/restauration-base.md. A backup that has
never been restored is not a backup.
""" """
import argparse
import os import os
import shutil import shutil
import sqlite3 import subprocess
import sys
from datetime import datetime, timedelta from datetime import datetime, timedelta
from urllib.parse import unquote, urlparse
# Configuration # Configuration
BACKUP_DIR = os.getenv('BACKUP_DIR', os.path.join(os.getcwd(), 'backups')) BACKUP_DIR = os.getenv('BACKUP_DIR', os.path.join(os.getcwd(), 'backups'))
BACKUP_RETENTION_DAYS = int(os.getenv('BACKUP_RETENTION_DAYS', 30)) BACKUP_RETENTION_DAYS = int(os.getenv('BACKUP_RETENTION_DAYS', 30))
DATABASE_PATH = os.getenv('DATABASE_PATH', os.path.join(os.getcwd(), 'instance', 'team_tryouts.db'))
DOCUMENTS_DIR = os.path.join(os.getcwd(), 'documents') DOCUMENTS_DIR = os.path.join(os.getcwd(), 'documents')
PG_DUMP = os.getenv('PG_DUMP', 'pg_dump')
PG_RESTORE = os.getenv('PG_RESTORE', 'pg_restore')
class BackupError(Exception):
"""Raised when a backup step fails in a way that must stop the run."""
# ---------------------------------------------------------------------------
# Connection handling
# ---------------------------------------------------------------------------
def parse_database_url(url):
"""Split a SQLAlchemy/PostgreSQL URL into pg_dump connection settings.
Accepts the dialect suffixes SQLAlchemy uses (postgresql+psycopg://),
which pg_dump does not understand.
Args:
url: The connection string.
Returns:
dict: host, port, dbname, user, password.
Raises:
BackupError: If the URL is missing or is not a PostgreSQL one.
"""
if not url:
raise BackupError('DATABASE_URL is not set.')
parsed = urlparse(url)
scheme = parsed.scheme.split('+')[0]
if scheme not in ('postgresql', 'postgres'):
raise BackupError(
f'DATABASE_URL is not a PostgreSQL connection string (scheme: {scheme!r}). '
'This script only backs up PostgreSQL.'
)
dbname = (parsed.path or '').lstrip('/')
if not dbname:
raise BackupError('DATABASE_URL does not name a database.')
return {
'host': parsed.hostname or 'localhost',
'port': str(parsed.port or 5432),
'dbname': dbname,
'user': unquote(parsed.username) if parsed.username else '',
'password': unquote(parsed.password) if parsed.password else '',
}
def describe_target(conn):
"""Human-readable target, deliberately without the password."""
user = f'{conn["user"]}@' if conn['user'] else ''
return f'{user}{conn["host"]}:{conn["port"]}/{conn["dbname"]}'
def build_dump_command(conn, output_path):
"""Assemble the pg_dump invocation.
--format=custom is compressed and lets pg_restore rebuild selectively;
plain SQL would be larger and all-or-nothing.
The password is never placed on the command line — it would be visible
to anyone able to list processes. It travels through PGPASSWORD instead,
which is what pg_dump documents for non-interactive use.
"""
return [
PG_DUMP,
'--host', conn['host'],
'--port', conn['port'],
'--username', conn['user'],
'--dbname', conn['dbname'],
'--format=custom',
'--no-owner',
'--no-privileges',
'--file', output_path,
]
def dump_environment(conn):
"""Environment for pg_dump/pg_restore, carrying the password out of argv."""
env = os.environ.copy()
if conn['password']:
env['PGPASSWORD'] = conn['password']
return env
# ---------------------------------------------------------------------------
# Backup steps
# ---------------------------------------------------------------------------
def create_backup_dir(): def create_backup_dir():
"""Create the backup directory if it doesn't exist.""" """Create the backup directory if it doesn't exist."""
os.makedirs(BACKUP_DIR, exist_ok=True) os.makedirs(BACKUP_DIR, exist_ok=True)
def backup_database(): def backup_database(conn):
"""Backup the SQLite database using sqlite3's built-in backup API. """Dump the PostgreSQL database.
Returns: Returns:
str: Path to the created backup file, or None if failed. str: Path to the created archive.
"""
if not os.path.exists(DATABASE_PATH):
print(f'[WARNING] Database not found at {DATABASE_PATH}. Skipping database backup.')
return None
Raises:
BackupError: If pg_dump is missing, fails, or produces nothing.
"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_filename = f'db_backup_{timestamp}.db' backup_path = os.path.join(BACKUP_DIR, f'db_backup_{timestamp}.dump')
backup_path = os.path.join(BACKUP_DIR, backup_filename)
print(f'[INFO] Dumping {describe_target(conn)}')
try: try:
source = sqlite3.connect(DATABASE_PATH) result = subprocess.run(
destination = sqlite3.connect(backup_path) build_dump_command(conn, backup_path),
source.backup(destination) env=dump_environment(conn),
source.close() capture_output=True,
destination.close() text=True,
print(f'[OK] Database backed up to: {backup_path}') timeout=900,
)
except FileNotFoundError:
raise BackupError(
f'{PG_DUMP} not found. Install the PostgreSQL client tools, or set '
'PG_DUMP to its full path.'
)
except subprocess.TimeoutExpired:
raise BackupError('pg_dump timed out after 15 minutes.')
if result.returncode != 0:
raise BackupError(f'pg_dump failed: {result.stderr.strip()}')
if not os.path.exists(backup_path) or os.path.getsize(backup_path) == 0:
raise BackupError('pg_dump reported success but produced an empty file.')
size_mb = os.path.getsize(backup_path) / (1024 * 1024)
print(f'[OK] Database backed up to: {backup_path} ({size_mb:.1f} MB)')
return backup_path return backup_path
except Exception as e:
print(f'[ERROR] Database backup failed: {e}')
return None def verify_backup(backup_path):
"""Check that the archive is readable and actually contains tables.
pg_restore --list parses the whole archive without touching any
database. A dump that cannot be listed cannot be restored, and an
archive holding no table would mean the dump ran against the wrong
target — both are silent failures worth catching here rather than
during an incident.
Args:
backup_path: Path to the archive to verify.
Returns:
bool: True if the archive looks restorable.
"""
if not backup_path or not os.path.exists(backup_path):
print('[ERROR] Nothing to verify.')
return False
try:
result = subprocess.run(
[PG_RESTORE, '--list', backup_path],
capture_output=True, text=True, timeout=300,
)
except FileNotFoundError:
print(f'[WARNING] {PG_RESTORE} not found: archive left unverified.')
return False
except subprocess.TimeoutExpired:
print('[ERROR] pg_restore --list timed out.')
return False
if result.returncode != 0:
print(f'[ERROR] Archive is not readable: {result.stderr.strip()}')
return False
table_count = sum(1 for line in result.stdout.splitlines()
if ' TABLE DATA ' in line)
if table_count == 0:
print('[ERROR] Archive contains no table data.')
return False
print(f'[OK] Archive verified: {table_count} table(s) present.')
return True
def backup_documents(): def backup_documents():
"""Backup the uploaded contract documents directory. """Archive the uploaded contract documents directory.
Returns: Returns:
str: Path to the created archive, or None if no documents exist. str: Path to the created archive, or None if there is nothing to
archive. Signed contracts live only on disk, so losing this
directory loses the documents themselves.
""" """
if not os.path.exists(DOCUMENTS_DIR): if not os.path.exists(DOCUMENTS_DIR):
print('[INFO] No documents directory found. Skipping document backup.') print('[INFO] No documents directory found. Skipping document backup.')
return None return None
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
archive_basename = f'documents_backup_{timestamp}' archive_basename = os.path.join(BACKUP_DIR, f'documents_backup_{timestamp}')
archive_path = os.path.join(BACKUP_DIR, archive_basename)
try: try:
shutil.make_archive(archive_path, 'zip', DOCUMENTS_DIR) shutil.make_archive(archive_basename, 'zip', DOCUMENTS_DIR)
zip_path = f'{archive_path}.zip' except Exception as exc:
print(f'[OK] Documents backed up to: {zip_path}') print(f'[ERROR] Document backup failed: {exc}')
return zip_path
except Exception as e:
print(f'[ERROR] Document backup failed: {e}')
return None return None
zip_path = f'{archive_basename}.zip'
size_mb = os.path.getsize(zip_path) / (1024 * 1024)
print(f'[OK] Documents backed up to: {zip_path} ({size_mb:.1f} MB)')
return zip_path
def cleanup_old_backups(): def cleanup_old_backups():
"""Remove backup files older than BACKUP_RETENTION_DAYS.""" """Remove backup files older than BACKUP_RETENTION_DAYS."""
@@ -90,95 +261,69 @@ def cleanup_old_backups():
for filename in os.listdir(BACKUP_DIR): for filename in os.listdir(BACKUP_DIR):
file_path = os.path.join(BACKUP_DIR, filename) file_path = os.path.join(BACKUP_DIR, filename)
if os.path.isfile(file_path): if not os.path.isfile(file_path):
file_time = datetime.fromtimestamp(os.path.getmtime(file_path)) continue
if file_time < cutoff: if datetime.fromtimestamp(os.path.getmtime(file_path)) >= cutoff:
continue
try: try:
os.remove(file_path) os.remove(file_path)
removed_count += 1 removed_count += 1
print(f'[CLEANUP] Removed old backup: {filename}') print(f'[CLEANUP] Removed old backup: {filename}')
except OSError as e: except OSError as exc:
print(f'[WARNING] Could not remove {filename}: {e}') print(f'[WARNING] Could not remove {filename}: {exc}')
if removed_count > 0: if removed_count:
print(f'[CLEANUP] Removed {removed_count} old backup(s).') print(f'[CLEANUP] Removed {removed_count} old backup(s).')
else: else:
print('[CLEANUP] No old backups to remove.') print('[CLEANUP] No old backups to remove.')
def verify_backup(backup_path): # ---------------------------------------------------------------------------
"""Verify a database backup by running a quick integrity check. # Entry point
# ---------------------------------------------------------------------------
Args: def main(argv=None):
backup_path: Path to the backup file to verify.
Returns:
bool: True if backup is valid, False otherwise.
"""
if not backup_path or not os.path.exists(backup_path):
return False
try:
conn = sqlite3.connect(backup_path)
cursor = conn.cursor()
cursor.execute('PRAGMA integrity_check')
result = cursor.fetchone()
conn.close()
is_valid = result[0] == 'ok'
if is_valid:
print(f'[OK] Backup integrity verified: {backup_path}')
else:
print(f'[ERROR] Backup integrity check failed: {backup_path} - {result[0]}')
return is_valid
except Exception as e:
print(f'[ERROR] Backup verification failed: {e}')
return False
def main():
"""Run the full backup process. """Run the full backup process.
Steps:
1. Create backup directory
2. Backup database
3. Backup documents (if any)
4. Verify database backup
5. Clean up old backups
Returns: Returns:
int: 0 on success, 1 on failure. int: 0 when the database was dumped AND verified, 1 otherwise. The
previous version returned 0 even when it had backed up nothing.
""" """
parser = argparse.ArgumentParser(description='Team Tryouts backup')
parser.add_argument('--verify-only', metavar='ARCHIVE',
help='Verify an existing archive and exit')
args = parser.parse_args(argv)
if args.verify_only:
return 0 if verify_backup(args.verify_only) else 1
print('=== Team Tryouts Backup ===') print('=== Team Tryouts Backup ===')
print(f'Started at: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}') print(f'Started at: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
print(f'Backup directory: {BACKUP_DIR}') print(f'Backup directory: {BACKUP_DIR}')
print(f'Retention period: {BACKUP_RETENTION_DAYS} days') print(f'Retention period: {BACKUP_RETENTION_DAYS} days')
print() print()
try:
conn = parse_database_url(os.getenv('DATABASE_URL'))
create_backup_dir() create_backup_dir()
backup_path = backup_database(conn)
except BackupError as exc:
print(f'[ERROR] {exc}')
print('\n=== Backup FAILED — no database backup was produced ===')
return 1
# 1. Backup database verified = verify_backup(backup_path)
db_backup_path = backup_database()
success = True
# 2. Verify database backup
if db_backup_path:
if not verify_backup(db_backup_path):
success = False
# 3. Backup documents
backup_documents() backup_documents()
# 4. Cleanup old backups
cleanup_old_backups() cleanup_old_backups()
print() print()
if success: if verified:
print('=== Backup completed successfully ===') print('=== Backup completed successfully ===')
else: return 0
print('=== Backup completed with warnings ===')
return 0 if success else 1 print('=== Backup FAILED verification — do not rely on this archive ===')
return 1
if __name__ == '__main__': if __name__ == '__main__':
exit(main()) sys.exit(main())
+175
View File
@@ -0,0 +1,175 @@
# Database Backup and Restore
A backup that has never been restored is not a backup. This document exists
so that the restore path is exercised **before** it is needed, not during an
incident.
---
## 1. What is backed up
`app/supporting_scripts/backup.py` produces two artefacts per run, in
`BACKUP_DIR` (default `./backups`):
| Artefact | Contents | Why it matters |
|---|---|---|
| `db_backup_<timestamp>.dump` | Full PostgreSQL dump, custom format | Every account, tryout, evaluation, note and contract record |
| `documents_backup_<timestamp>.zip` | `documents/` directory | The contract **files** themselves — they exist only on disk, the database stores paths |
Losing either one alone loses data. A database restore without the document
archive leaves contract rows pointing at files that no longer exist.
---
## 2. Running a backup
```bash
# From the project root, with DATABASE_URL set
python app/supporting_scripts/backup.py
```
Requires the PostgreSQL client tools (`pg_dump`, `pg_restore`) on `PATH`, or
`PG_DUMP` / `PG_RESTORE` pointing at them.
| Variable | Default | Purpose |
|---|---|---|
| `DATABASE_URL` | — | Required. `postgresql://user:pass@host:port/dbname` |
| `BACKUP_DIR` | `./backups` | Destination directory |
| `BACKUP_RETENTION_DAYS` | `30` | Files older than this are deleted |
| `PG_DUMP` / `PG_RESTORE` | `pg_dump` / `pg_restore` | Full paths if not on `PATH` |
**Exit code 0 means the dump was produced *and* verified.** Anything else
means you have no usable backup from that run — treat a non-zero exit as an
incident, not a warning. Whatever schedules this job must check the exit
code; the previous version of the script returned 0 even when it had backed
up nothing at all.
### Verifying an existing archive
```bash
python app/supporting_scripts/backup.py --verify-only backups/db_backup_20260807_030000.dump
```
This reads the archive with `pg_restore --list` and confirms it contains
table data. It touches no database.
---
## 3. Restore drill
**Run this on a throwaway database, at least once a quarter, and after any
change to the schema tooling.** It is the only thing that turns a file into
a guarantee.
### 3.1 Create an isolated target
Never restore onto the production database to "test" a backup.
```bash
createdb -h localhost -U postgres tryouts_restore_test
```
### 3.2 Restore
```bash
# --clean --if-exists makes the restore repeatable
pg_restore \
--host localhost --port 5432 --username postgres \
--dbname tryouts_restore_test \
--clean --if-exists --no-owner --no-privileges \
backups/db_backup_20260807_030000.dump
```
`pg_restore` reports errors per object and continues. **Read its output.**
A restore that emits errors and returns 0 has still lost something.
### 3.3 Check the data is actually there
```sql
-- Connected to tryouts_restore_test
SELECT COUNT(*) FROM users;
SELECT role, COUNT(*) FROM users GROUP BY role ORDER BY role;
SELECT COUNT(*) FROM tryouts;
SELECT COUNT(*) FROM evaluations;
SELECT COUNT(*) FROM contracts;
SELECT MAX(created_at) FROM users;
```
Compare against production. The last query tells you how old the backup is
— the number that matters during an incident.
### 3.4 Start the application against the restored copy
```bash
DATABASE_URL="postgresql://postgres@localhost:5432/tryouts_restore_test" \
SECRET_KEY="throwaway-for-the-drill" \
ENABLE_DISCORD_BOT=false \
FORCE_HTTPS=false \
python wsgi.py
```
`ENABLE_DISCORD_BOT=false` is not optional. Without it the drill starts a
real bot against the real Discord server and sends real notifications to
real people, from restored data.
Smoke test:
1. `GET /health` returns 200 with `"database": "connected"`.
2. Log in with a known account.
3. Open a tryout and check its registrations are present.
4. Open the calendar.
### 3.5 Tear down
```bash
dropdb -h localhost -U postgres tryouts_restore_test
```
---
## 4. Restoring documents
```bash
unzip backups/documents_backup_20260807_030000.zip -d documents/
```
Then confirm a contract downloads through the application, not just that
the file exists: `Contract.file_path` stores an **absolute** path recorded
at upload time. If the deployment root has changed since, the rows point
somewhere that no longer exists and the files must be placed at the old
path, or the column updated.
---
## 5. Recovery plan
| Scenario | First move | Then |
|---|---|---|
| Accidental deletion of a few records | Restore to a throwaway database (§3), extract the rows, re-insert them | Do **not** restore over production |
| Database corrupted or lost | Restore the latest verified dump onto a fresh database, repoint `DATABASE_URL` | Restore documents (§4), then smoke test (§3.4) |
| Bad deployment | Redeploy the previous commit | The database is untouched unless a migration ran |
| Bad migration | Restore the pre-migration dump | Always take one immediately before migrating |
| Server lost entirely | Provision a host, restore database and documents, redeploy | Discord bot token and `SECRET_KEY` must be reissued if they were on the lost host |
### Two numbers to agree on
- **RPO** — how much data may be lost. It equals the backup interval.
Nightly backups mean up to 24 hours of tryouts, evaluations and notes.
- **RTO** — how long recovery may take. Measure it during the drill; do not
estimate it.
Neither number is currently set for this project. Deciding them is a
prerequisite to claiming there is a backup policy.
---
## 6. Open points
- **Off-site copy.** Backups written next to the application are lost with
the host. Nothing currently copies them elsewhere.
- **Encryption at rest.** The dump contains every account record and
password hash. It is not encrypted.
- **Scheduling.** No scheduled task or cron job is configured in the
repository. The script must be wired to one, with its exit code monitored.
- **Restore drill.** Has never been performed. Until it is, the restore
path is untested.
+123
View File
@@ -0,0 +1,123 @@
"""Backup script.
DATA-002 / OPS-001. The previous script targeted SQLite while production
runs on PostgreSQL: it printed "[WARNING] Database not found" and still
exited 0, so a scheduled task watching the exit code saw green while
nothing had ever been backed up.
These tests cover what can be checked without a PostgreSQL server: URL
parsing, command construction, password handling, and — above all — that
failure now produces a non-zero exit code.
"""
import pytest
from app.supporting_scripts import backup as backup_module
from app.supporting_scripts.backup import (
BackupError,
build_dump_command,
describe_target,
dump_environment,
parse_database_url,
)
PASSWORD = 'sup3r-s3cret'
URL = f'postgresql://appuser:{PASSWORD}@db.example.test:6432/tryouts'
class TestUrlParsing:
def test_a_standard_url_is_split(self):
conn = parse_database_url(URL)
assert conn['host'] == 'db.example.test'
assert conn['port'] == '6432'
assert conn['dbname'] == 'tryouts'
assert conn['user'] == 'appuser'
assert conn['password'] == PASSWORD
def test_the_sqlalchemy_dialect_suffix_is_accepted(self):
"""SQLAlchemy writes postgresql+psycopg://, which pg_dump rejects."""
conn = parse_database_url(
'postgresql+psycopg://u:p@localhost/tryouts')
assert conn['dbname'] == 'tryouts'
def test_the_default_port_is_applied(self):
assert parse_database_url('postgresql://u:p@localhost/db')['port'] == '5432'
def test_percent_encoded_credentials_are_decoded(self):
conn = parse_database_url('postgresql://u%40corp:p%23ass@h/db')
assert conn['user'] == 'u@corp'
assert conn['password'] == 'p#ass'
def test_a_missing_url_is_refused(self):
with pytest.raises(BackupError, match='DATABASE_URL is not set'):
parse_database_url(None)
def test_a_sqlite_url_is_refused(self):
"""The exact case that used to pass silently."""
with pytest.raises(BackupError, match='not a PostgreSQL'):
parse_database_url('sqlite:///instance/team_tryouts.db')
def test_a_url_without_a_database_name_is_refused(self):
with pytest.raises(BackupError, match='does not name a database'):
parse_database_url('postgresql://u:p@localhost/')
class TestCommandConstruction:
def test_the_password_never_reaches_the_command_line(self):
"""Anything on argv is visible to any process listing."""
command = build_dump_command(parse_database_url(URL), '/tmp/out.dump')
assert PASSWORD not in ' '.join(command)
def test_the_password_travels_through_the_environment(self):
env = dump_environment(parse_database_url(URL))
assert env['PGPASSWORD'] == PASSWORD
def test_no_pgpassword_is_set_when_there_is_no_password(self):
env = dump_environment(parse_database_url('postgresql://u@localhost/db'))
assert 'PGPASSWORD' not in env
def test_the_dump_is_written_in_the_custom_format(self):
"""Custom format is compressed and restorable selectively."""
command = build_dump_command(parse_database_url(URL), '/tmp/out.dump')
assert '--format=custom' in command
assert '--file' in command
assert command[command.index('--file') + 1] == '/tmp/out.dump'
def test_connection_settings_are_passed_through(self):
command = build_dump_command(parse_database_url(URL), '/tmp/out.dump')
assert command[command.index('--host') + 1] == 'db.example.test'
assert command[command.index('--port') + 1] == '6432'
assert command[command.index('--dbname') + 1] == 'tryouts'
def test_the_description_omits_the_password(self):
"""It is printed on stdout and lands in scheduler logs."""
described = describe_target(parse_database_url(URL))
assert PASSWORD not in described
assert 'db.example.test:6432/tryouts' in described
class TestExitCodes:
"""The regression that matters: silence is no longer success."""
def test_a_sqlite_url_fails_the_run(self, monkeypatch, capsys):
monkeypatch.setenv('DATABASE_URL', 'sqlite:///instance/team_tryouts.db')
assert backup_module.main([]) == 1
assert 'FAILED' in capsys.readouterr().out
def test_a_missing_url_fails_the_run(self, monkeypatch):
monkeypatch.delenv('DATABASE_URL', raising=False)
assert backup_module.main([]) == 1
def test_a_missing_pg_dump_fails_the_run(self, monkeypatch, tmp_path):
monkeypatch.setenv('DATABASE_URL', URL)
monkeypatch.setattr(backup_module, 'BACKUP_DIR', str(tmp_path))
monkeypatch.setattr(backup_module, 'PG_DUMP', 'pg_dump_that_does_not_exist')
assert backup_module.main([]) == 1
def test_verifying_a_missing_archive_fails(self, tmp_path):
assert backup_module.main(
['--verify-only', str(tmp_path / 'nope.dump')]) == 1