This commit is contained in:
@@ -0,0 +1,285 @@
|
||||
"""Admin backup — create, download, delete, restore, and audit trail.
|
||||
|
||||
ADMIN-008. The backup buttons on the admin panel run pg_dump/pg_restore
|
||||
in production, but tests mock the subprocess boundary. These tests verify
|
||||
the route logic, the BackupRecord model, and audit logging.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from app.extensions import db
|
||||
from app.models import AuditLog, BackupRecord
|
||||
|
||||
|
||||
def _redirected(response):
|
||||
return response.status_code in (301, 302)
|
||||
|
||||
|
||||
class TestBackupCreate:
|
||||
def test_create_backup_succeeds(self, app, client, as_role, monkeypatch):
|
||||
import app.routes.admin as admin_module
|
||||
from app.supporting_scripts.backup import BackupError
|
||||
|
||||
def fake_create_backup_record(backup_type='manual', notes=None):
|
||||
from app.models import BackupRecord
|
||||
|
||||
record = BackupRecord(
|
||||
filename='db_backup_test.dump',
|
||||
file_path='/tmp/db_backup_test.dump',
|
||||
size_bytes=1234,
|
||||
backup_type=backup_type,
|
||||
notes=notes,
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
return record
|
||||
|
||||
monkeypatch.setattr(admin_module, '_create_backup_record', fake_create_backup_record)
|
||||
|
||||
as_role('admin')
|
||||
response = client.post(
|
||||
'/admin/backup/create', data={'notes': 'test backup'}, follow_redirects=False,
|
||||
)
|
||||
assert _redirected(response)
|
||||
|
||||
with app.app_context():
|
||||
record = BackupRecord.query.filter_by(filename='db_backup_test.dump').first()
|
||||
assert record is not None
|
||||
assert record.notes == 'test backup'
|
||||
|
||||
def test_create_backup_failure_is_handled(self, app, client, as_role, monkeypatch):
|
||||
import app.routes.admin as admin_module
|
||||
from app.supporting_scripts.backup import BackupError
|
||||
|
||||
def fake_create_backup_record(backup_type='manual', notes=None):
|
||||
raise BackupError('pg_dump not found')
|
||||
|
||||
monkeypatch.setattr(admin_module, '_create_backup_record', fake_create_backup_record)
|
||||
|
||||
as_role('admin')
|
||||
response = client.post(
|
||||
'/admin/backup/create', data={}, follow_redirects=True,
|
||||
)
|
||||
html = response.data.decode()
|
||||
assert 'Backup failed' in html
|
||||
|
||||
def test_create_backup_creates_audit_log(self, app, client, as_role, monkeypatch):
|
||||
import app.routes.admin as admin_module
|
||||
|
||||
def fake_create_backup_record(backup_type='manual', notes=None):
|
||||
from app.models import BackupRecord
|
||||
|
||||
record = BackupRecord(
|
||||
filename='db_backup_test.dump',
|
||||
file_path='/tmp/db_backup_test.dump',
|
||||
size_bytes=1234,
|
||||
backup_type=backup_type,
|
||||
notes=notes,
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
return record
|
||||
|
||||
monkeypatch.setattr(admin_module, '_create_backup_record', fake_create_backup_record)
|
||||
|
||||
as_role('admin')
|
||||
client.post('/admin/backup/create', data={}, follow_redirects=False)
|
||||
|
||||
with app.app_context():
|
||||
entry = AuditLog.query.filter_by(action='backup_created').first()
|
||||
assert entry is not None
|
||||
|
||||
|
||||
class TestBackupDownload:
|
||||
def test_download_existing_backup(self, app, client, as_role, tmp_path):
|
||||
from app.models import BackupRecord
|
||||
|
||||
backup_file = tmp_path / 'db_backup_test.dump'
|
||||
backup_file.write_text('fake dump content')
|
||||
|
||||
as_role('admin')
|
||||
with app.app_context():
|
||||
record = BackupRecord(
|
||||
filename='db_backup_test.dump',
|
||||
file_path=str(backup_file),
|
||||
size_bytes=18,
|
||||
backup_type='manual',
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
record_id = record.id
|
||||
|
||||
response = client.get(f'/admin/backup/{record_id}/download')
|
||||
assert response.status_code == 200
|
||||
assert response.data == b'fake dump content'
|
||||
|
||||
def test_download_missing_backup_file(self, app, client, as_role):
|
||||
from app.models import BackupRecord
|
||||
|
||||
as_role('admin')
|
||||
with app.app_context():
|
||||
record = BackupRecord(
|
||||
filename='missing.dump',
|
||||
file_path='/nonexistent/missing.dump',
|
||||
size_bytes=0,
|
||||
backup_type='manual',
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
record_id = record.id
|
||||
|
||||
response = client.get(
|
||||
f'/admin/backup/{record_id}/download', follow_redirects=True,
|
||||
)
|
||||
html = response.data.decode()
|
||||
assert 'missing from disk' in html.lower()
|
||||
|
||||
|
||||
class TestBackupDelete:
|
||||
def test_delete_backup_removes_record_and_file(self, app, client, as_role, tmp_path):
|
||||
from app.models import BackupRecord
|
||||
|
||||
backup_file = tmp_path / 'db_backup_delete.dump'
|
||||
backup_file.write_text('fake dump content')
|
||||
|
||||
as_role('admin')
|
||||
with app.app_context():
|
||||
record = BackupRecord(
|
||||
filename='db_backup_delete.dump',
|
||||
file_path=str(backup_file),
|
||||
size_bytes=18,
|
||||
backup_type='manual',
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
record_id = record.id
|
||||
|
||||
response = client.post(
|
||||
f'/admin/backup/{record_id}/delete', follow_redirects=False,
|
||||
)
|
||||
assert _redirected(response)
|
||||
|
||||
with app.app_context():
|
||||
assert BackupRecord.query.get(record_id) is None
|
||||
assert not backup_file.exists()
|
||||
|
||||
def test_delete_backup_creates_audit_log(self, app, client, as_role, tmp_path):
|
||||
from app.models import BackupRecord
|
||||
|
||||
backup_file = tmp_path / 'db_backup_delete2.dump'
|
||||
backup_file.write_text('fake')
|
||||
|
||||
as_role('admin')
|
||||
with app.app_context():
|
||||
record = BackupRecord(
|
||||
filename='db_backup_delete2.dump',
|
||||
file_path=str(backup_file),
|
||||
size_bytes=4,
|
||||
backup_type='manual',
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
record_id = record.id
|
||||
|
||||
client.post(f'/admin/backup/{record_id}/delete', follow_redirects=False)
|
||||
|
||||
with app.app_context():
|
||||
entry = AuditLog.query.filter_by(action='backup_deleted').first()
|
||||
assert entry is not None
|
||||
|
||||
|
||||
class TestBackupRestore:
|
||||
def test_restore_with_missing_file_fails(self, app, client, as_role):
|
||||
from app.models import BackupRecord
|
||||
|
||||
as_role('admin')
|
||||
with app.app_context():
|
||||
record = BackupRecord(
|
||||
filename='missing_restore.dump',
|
||||
file_path='/nonexistent/missing_restore.dump',
|
||||
size_bytes=0,
|
||||
backup_type='manual',
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
record_id = record.id
|
||||
|
||||
response = client.post(
|
||||
f'/admin/backup/{record_id}/restore', follow_redirects=True,
|
||||
)
|
||||
html = response.data.decode()
|
||||
assert 'missing from disk' in html.lower()
|
||||
|
||||
def test_restore_creates_safety_backup_first(
|
||||
self, app, client, as_role, monkeypatch, tmp_path
|
||||
):
|
||||
import app.routes.admin as admin_module
|
||||
|
||||
# The restore route parses DATABASE_URL outside its try block.
|
||||
monkeypatch.setenv('DATABASE_URL', 'postgresql://appuser:[email protected]:5432/tryouts')
|
||||
|
||||
backup_file = tmp_path / 'db_backup_restore.dump'
|
||||
backup_file.write_text('fake')
|
||||
|
||||
safety_file = tmp_path / 'db_backup_safety.dump'
|
||||
safety_file.write_text('fake safety')
|
||||
|
||||
as_role('admin')
|
||||
with app.app_context():
|
||||
from app.models import BackupRecord
|
||||
|
||||
record = BackupRecord(
|
||||
filename='db_backup_restore.dump',
|
||||
file_path=str(backup_file),
|
||||
size_bytes=4,
|
||||
backup_type='manual',
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
record_id = record.id
|
||||
|
||||
def fake_create_backup_record(backup_type='pre_restore', notes=None):
|
||||
from app.models import BackupRecord
|
||||
|
||||
record = BackupRecord(
|
||||
filename='db_backup_safety.dump',
|
||||
file_path=str(safety_file),
|
||||
size_bytes=12,
|
||||
backup_type=backup_type,
|
||||
notes=notes,
|
||||
created_by_id=1,
|
||||
)
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
return record
|
||||
|
||||
monkeypatch.setattr(admin_module, '_create_backup_record', fake_create_backup_record)
|
||||
|
||||
import subprocess
|
||||
|
||||
def fake_subprocess_run(cmd, env=None, capture_output=True, text=True, timeout=None):
|
||||
class Result:
|
||||
returncode = 0
|
||||
stderr = ''
|
||||
stdout = ''
|
||||
|
||||
return Result()
|
||||
|
||||
monkeypatch.setattr(subprocess, 'run', fake_subprocess_run)
|
||||
|
||||
client.post(f'/admin/backup/{record_id}/restore', follow_redirects=False)
|
||||
|
||||
with app.app_context():
|
||||
safety = BackupRecord.query.filter_by(filename='db_backup_safety.dump').first()
|
||||
assert safety is not None
|
||||
assert safety.backup_type == 'pre_restore'
|
||||
Reference in New Issue
Block a user