445f1361dc
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
43 lines
2.2 KiB
Python
43 lines
2.2 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import pytest
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
|
|
from recovery_evidence import recovery_signals, ROOT
|
|
from service_assurance import evaluate
|
|
|
|
NOW = datetime(2026, 9, 6, 12, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_real_receipts_preserve_completion_and_eventually_expire():
|
|
signals = recovery_signals(NOW)
|
|
assert all(s['result'] == 'pass' for s in signals.values())
|
|
assert signals['apps-pg.restore']['observed_at'] == '2026-09-05T22:30:45.208512+00:00'
|
|
contract = {'cluster_uid': 'test', 'capture_max_age_seconds': 900,
|
|
'signals': {key: {'owner': 'platform', 'max_age_seconds': 2592000} for key in signals}}
|
|
later = NOW + timedelta(days=31)
|
|
result = evaluate(contract, {'schema': 'railiance-platform.observation.v1',
|
|
'cluster_uid': 'test', 'captured_at': later.isoformat(), 'signals': recovery_signals(later)}, later)
|
|
assert all(s['state'] == 'stale' for s in result['signals'].values())
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['hash', 'cleanup', 'provider', 'missing_time', 'future', 'naive'])
|
|
def test_invalid_receipt_is_unavailable(tmp_path, change):
|
|
index = json.loads((ROOT / 'assurance/recovery-evidence.json').read_text())
|
|
index['receipts'] = index['receipts'][:1]
|
|
entry = index['receipts'][0]
|
|
receipt = json.loads((ROOT / entry['path']).read_text())
|
|
if change == 'cleanup': receipt['cleanup'] = False
|
|
if change == 'provider': receipt['primary_destination'] = 's3://other/'
|
|
if change == 'missing_time': del receipt['finished_at']
|
|
if change == 'future': receipt['finished_at'] = '2027-01-01T00:00:00Z'
|
|
if change == 'naive': receipt['finished_at'] = '2026-09-05T23:00:00'
|
|
path = tmp_path / entry['path']
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text(json.dumps(receipt))
|
|
if change != 'hash': entry['sha256'] = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
(tmp_path / 'assurance').mkdir()
|
|
(tmp_path / 'assurance/recovery-evidence.json').write_text(json.dumps(index))
|
|
assert recovery_signals(NOW, tmp_path)['apps-pg.restore']['result'] == 'unavailable'
|