tegwick
05cf67e582
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d
106 lines
4 KiB
Python
106 lines
4 KiB
Python
"""Recovery must resume the app even when snapshot capture or cleanup fails."""
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import signal
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def helper():
|
|
path = Path(__file__).resolve().parents[1] / 'scripts/snapshot-content-and-db.py'
|
|
spec = importlib.util.spec_from_file_location('csoc_snapshot', path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def setup_fake(helper, monkeypatch, tmp_path, failure=None, execute=True):
|
|
def document(*args):
|
|
kind = args[1]
|
|
if kind == 'namespace':
|
|
return {'metadata': {'uid': helper.CLUSTER_UID}}
|
|
if kind == 'deployment':
|
|
return {'spec': {'replicas': 1, 'template': {'spec': {
|
|
'containers': [{'image': helper.IMAGE}]}}}}
|
|
if kind == 'pvc':
|
|
return {'status': {'phase': 'Bound'}}
|
|
return {'status': {'currentPrimary': 'apps-pg-1'}}
|
|
|
|
calls = []
|
|
|
|
def k(*args, **kwargs):
|
|
calls.append(args)
|
|
if args[0] == 'delete' and failure == 'cleanup':
|
|
raise helper.Refused('command_failed')
|
|
if 'pg_dump' in args:
|
|
return b'invalid' if failure == 'dump' else b'PGDMPfixture'
|
|
return b'fixture-content'
|
|
|
|
dest = tmp_path / 'snapshot'
|
|
argv = ['snapshot', '--destination', str(dest)]
|
|
if execute:
|
|
argv.append('--execute')
|
|
monkeypatch.setattr(sys, 'argv', argv)
|
|
monkeypatch.setattr(helper, 'document', document)
|
|
monkeypatch.setattr(helper, 'k', k)
|
|
monkeypatch.setattr(helper, 'LOCK_PATH', tmp_path / 'snapshot.lock')
|
|
return calls, dest
|
|
|
|
|
|
def test_default_review_does_not_pause_or_capture(helper, monkeypatch, tmp_path):
|
|
calls, dest = setup_fake(helper, monkeypatch, tmp_path, execute=False)
|
|
assert helper.main() == 0
|
|
assert not dest.exists()
|
|
assert not any(c[0] in {'scale', 'exec', 'delete'} for c in calls)
|
|
assert any('--dry-run=server' in c for c in calls)
|
|
|
|
|
|
@pytest.mark.parametrize('failure', ['dump', 'cleanup'])
|
|
def test_failure_always_attempts_app_resume(helper, monkeypatch, tmp_path, failure):
|
|
calls, _ = setup_fake(helper, monkeypatch, tmp_path, failure=failure)
|
|
with pytest.raises(helper.Refused):
|
|
helper.main()
|
|
assert any('--replicas=0' in c for c in calls)
|
|
assert any('--replicas=1' in c for c in calls)
|
|
assert any(c[0] == 'rollout' for c in calls)
|
|
|
|
|
|
def test_success_artifacts_private_and_paired(helper, monkeypatch, tmp_path):
|
|
calls, dest = setup_fake(helper, monkeypatch, tmp_path)
|
|
assert helper.main() == 0
|
|
assert dest.stat().st_mode & 0o777 == 0o700
|
|
for name in ['database.dump', 'content.tar.gz', 'receipt.json']:
|
|
assert (dest / name).stat().st_mode & 0o777 == 0o600
|
|
receipt = json.loads((dest / 'receipt.json').read_text())
|
|
assert receipt['writers_quiesced'] and receipt['app_resumed']
|
|
assert receipt['status'] == 'captured'
|
|
assert receipt['database'] == 'coulomb_social_db'
|
|
assert not any('pg_restore' in c for c in calls)
|
|
|
|
|
|
def test_second_snapshot_refuses_without_pausing(helper, monkeypatch, tmp_path):
|
|
calls, _ = setup_fake(helper, monkeypatch, tmp_path)
|
|
with helper.LOCK_PATH.open('w') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
with pytest.raises(helper.Refused, match='snapshot_already_running'):
|
|
helper.main()
|
|
assert calls == []
|
|
|
|
|
|
def test_termination_during_capture_resumes_app(helper, monkeypatch, tmp_path):
|
|
calls, _ = setup_fake(helper, monkeypatch, tmp_path)
|
|
original = helper.k
|
|
def terminate_during_dump(*args, **kwargs):
|
|
if 'pg_dump' in args:
|
|
handler = signal.getsignal(signal.SIGTERM)
|
|
handler(signal.SIGTERM, None)
|
|
return original(*args, **kwargs)
|
|
monkeypatch.setattr(helper, 'k', terminate_during_dump)
|
|
with pytest.raises(helper.Refused, match='snapshot_interrupted'):
|
|
helper.main()
|
|
assert any('--replicas=1' in call for call in calls)
|
|
assert any(call[0] == 'rollout' for call in calls)
|