tegwick
3166da1d2e
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
92 lines
4.7 KiB
Python
92 lines
4.7 KiB
Python
from contextlib import redirect_stdout
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_alert_ack import NOW, START
|
|
from alert_ack import Actor, Store
|
|
from alert_admin import main, snapshot
|
|
|
|
|
|
class MaintenanceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.path = self.root / 'source.db'
|
|
self.store = Store(self.path)
|
|
self.actor = Actor('https://fixture.example', 'private-human', 'tenant:platform',
|
|
('railiance-admin',), NOW+900, 'c'*32)
|
|
self.ids = []
|
|
self.events = []
|
|
for i in range(3):
|
|
identity = self.store.receive({'version':'4', 'receiver':'railiance-admin-email', 'alerts':[
|
|
{'status':'firing','fingerprint':f'{i:016x}','startsAt':START,
|
|
'labels':{'alertname':'Fixture','owner':'railiance-telemetry'}}]},NOW)[0]
|
|
self.ids.append(identity)
|
|
self.events.append(self.store.acknowledge(identity,self.actor,'decision',NOW))
|
|
def send(event):
|
|
if event['id'] == self.events[0]: return 202, {'status':'accepted','reference':'audit:'+event['id']}
|
|
if event['id'] == self.events[1]: return 403, {}
|
|
raise TimeoutError()
|
|
self.store.drain(send)
|
|
|
|
def cli(self,*args):
|
|
output=io.StringIO()
|
|
with redirect_stdout(output): code=main(['--database',str(self.path),*args])
|
|
return code,json.loads(output.getvalue())
|
|
|
|
def test_status_and_explicit_requeue_preserve_evidence(self):
|
|
self.assertEqual(self.cli('status'),(0,{'outbox':{'pending':1,'blocked':1,'delivered':1}}))
|
|
self.assertEqual(self.cli('list-blocked')[1]['event_ids'],[self.events[1]])
|
|
with self.store.connect() as db: before=db.execute('SELECT event_id,body FROM outbox ORDER BY event_id').fetchall()
|
|
self.assertTrue(self.cli('requeue','--event-id',self.events[1])[1]['requeued'])
|
|
self.assertEqual(self.cli('requeue','--event-id',self.events[0])[0],2)
|
|
with self.store.connect() as db: after=db.execute('SELECT event_id,body FROM outbox ORDER BY event_id').fetchall()
|
|
self.assertEqual([tuple(r) for r in before],[tuple(r) for r in after])
|
|
|
|
def test_backup_restore_preserves_all_states_identity_and_immutability(self):
|
|
backup=self.root/'snapshot.db'; restored=self.root/'restored.db'
|
|
snapshot(self.path,backup)
|
|
snapshot(backup,restored)
|
|
store=Store(restored)
|
|
for identity,event in zip(self.ids,self.events):
|
|
self.assertEqual(store.acknowledge(identity,self.actor,'new-decision',NOW+5),event)
|
|
self.assertEqual([store.get(i)['audit_status'] for i in self.ids],['delivered','blocked','pending'])
|
|
with store.connect() as db:
|
|
with self.assertRaises(sqlite3.IntegrityError):db.execute('DELETE FROM acknowledgments')
|
|
with self.assertRaises(sqlite3.IntegrityError):db.execute("UPDATE outbox SET body='{}'")
|
|
self.assertEqual(restored.stat().st_mode & 0o777,0o600)
|
|
self.assertTrue(store.requeue(self.events[1]))
|
|
sent=[]
|
|
def duplicate(event):
|
|
sent.append(event['id']);return 200,{'status':'duplicate','reference':'audit:'+event['id']}
|
|
store.drain(duplicate)
|
|
self.assertEqual(set(sent),set(self.events[1:]))
|
|
self.assertFalse(store.audit_debt())
|
|
self.assertEqual(self.store.audit_debt(),{'blocked':1,'pending':1})
|
|
|
|
def test_refuse_overwrite_missing_input_bad_schema_and_symlink(self):
|
|
with self.assertRaises(FileExistsError):snapshot(self.path,self.path)
|
|
missing=self.root/'missing.db'
|
|
with redirect_stdout(io.StringIO()):self.assertEqual(main(['--database',str(missing),'status']),1)
|
|
self.assertFalse(missing.exists())
|
|
invalid=self.root/'invalid.db';invalid.write_bytes(b'bad')
|
|
invalid.chmod(0o600)
|
|
with self.assertRaises(sqlite3.DatabaseError):snapshot(invalid,self.root/'bad-backup.db')
|
|
self.assertFalse((self.root/'bad-backup.db').exists())
|
|
link=self.root/'link.db';link.symlink_to(self.path)
|
|
with self.assertRaises(ValueError):snapshot(link,self.root/'link-backup.db')
|
|
|
|
def test_metrics_are_aggregate_and_show_empty_states(self):
|
|
metrics=self.store.audit_metrics(123)
|
|
for state in ('pending','blocked','delivered'):
|
|
self.assertIn(f'{{status="{state}"}} 1',metrics)
|
|
self.assertIn('timestamp_seconds 123.0',metrics)
|
|
self.assertNotIn(self.actor.subject,metrics)
|
|
for event in self.events:self.assertNotIn(event,metrics)
|
|
empty=Store(self.root/'empty.db').audit_metrics(0)
|
|
self.assertIn('{status="pending"} 0',empty)
|