134 lines
5.4 KiB
Python
134 lines
5.4 KiB
Python
"""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
|
|
|
|
def test_a_missing_document_store_makes_an_otherwise_valid_run_incomplete(
|
|
self, monkeypatch, tmp_path
|
|
):
|
|
monkeypatch.setenv('DATABASE_URL', URL)
|
|
monkeypatch.setenv('DOCUMENTS_ROOT', str(tmp_path / 'missing-documents'))
|
|
monkeypatch.setattr(backup_module, 'BACKUP_DIR', str(tmp_path / 'backups'))
|
|
monkeypatch.setattr(backup_module, 'backup_database', lambda conn: 'database.dump')
|
|
monkeypatch.setattr(backup_module, 'verify_backup', lambda path: True)
|
|
monkeypatch.setattr(backup_module, 'cleanup_old_backups', lambda: None)
|
|
|
|
assert backup_module.main([]) == 1
|