52224b84f8
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 150322@bnt-lap001 Assistant-Session: 16a7b788-374e-4915-a1df-fc87ffd9a5e4
186 lines
9.2 KiB
Python
186 lines
9.2 KiB
Python
import copy
|
|
from datetime import datetime, timezone
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location('service_assurance', ROOT / 'scripts/service_assurance.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
class EvidenceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = datetime(2026, 9, 5, 12, tzinfo=timezone.utc)
|
|
self.contract = {'cluster_uid': 'expected', 'capture_max_age_seconds': 60,
|
|
'signals': {'db.backup': {'owner': 'platform', 'max_age_seconds': 3600}}}
|
|
self.observation = {'schema': 'railiance-platform.observation.v1',
|
|
'cluster_uid': 'expected', 'captured_at': '2026-09-05T12:00:00Z',
|
|
'signals': {'db.backup': {'observed_at': '2026-09-05T11:30:00Z', 'result': 'pass'}}}
|
|
|
|
def state(self):
|
|
return m.evaluate(self.contract, self.observation, self.now)['signals']['db.backup']['state']
|
|
|
|
def test_current_old_missing_failure_and_unavailable_are_distinct(self):
|
|
self.assertEqual(self.state(), 'healthy')
|
|
self.observation['signals']['db.backup']['observed_at'] = '2026-09-05T10:00:00Z'
|
|
self.assertEqual(self.state(), 'stale')
|
|
self.observation['signals']['db.backup']['result'] = 'fail'
|
|
self.assertEqual(self.state(), 'failed')
|
|
self.observation['signals']['db.backup']['result'] = 'unavailable'
|
|
self.assertEqual(self.state(), 'unavailable')
|
|
self.observation['signals'].clear()
|
|
self.assertEqual(self.state(), 'missing')
|
|
|
|
def test_replayed_capture_cannot_stay_healthy(self):
|
|
self.observation['captured_at'] = '2026-09-05T11:59:00Z'
|
|
self.assertEqual(self.state(), 'healthy')
|
|
self.observation['captured_at'] = '2026-09-05T11:58:59Z'
|
|
self.assertEqual(self.state(), 'stale')
|
|
|
|
def test_wrong_cluster_future_and_naive_times_refused(self):
|
|
for field, value in [('cluster_uid', 'other'), ('captured_at', '2026-09-05T12:01:00Z'),
|
|
('captured_at', '2026-09-05T12:00:00')]:
|
|
d = copy.deepcopy(self.observation); d[field] = value
|
|
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
|
|
m.evaluate(self.contract, d, self.now)
|
|
self.observation['signals']['db.backup']['observed_at'] = '2026-09-06T00:00:00Z'
|
|
with self.assertRaises(ValueError):
|
|
self.state()
|
|
|
|
def test_no_unrecognized_payload_can_escape_in_error(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
p = Path(directory) / 'input.json'
|
|
self.observation['token'] = 'CANARY_MUST_NOT_ESCAPE'
|
|
p.write_text(json.dumps(self.observation))
|
|
result = subprocess.run([sys.executable, str(ROOT / 'scripts/service_assurance.py'),
|
|
'evaluate', str(p)], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertNotIn('CANARY', result.stdout + result.stderr)
|
|
|
|
def test_unknown_signal_and_extra_sample_fields_fail(self):
|
|
for signals in [{'arbitrary': {}}, {'db.backup': {'observed_at': '2026-09-05T12:00:00Z',
|
|
'result': 'pass', 'value': 'secret'}}]:
|
|
self.observation['signals'] = signals
|
|
with self.assertRaises(ValueError):
|
|
self.state()
|
|
|
|
def test_health_does_not_imply_monitoring_or_guarantees(self):
|
|
result = m.evaluate(self.contract, self.observation, self.now)
|
|
self.assertTrue(result['healthy'])
|
|
self.assertEqual(result['transport'], 'unmonitored')
|
|
self.assertEqual(result['guarantees'], 'unsupported')
|
|
|
|
|
|
@unittest.skipUnless((ROOT.parent / 'rapp-postgres/scripts/render_consumers.py').exists(),
|
|
'owner package checkout required for integration checks')
|
|
class AdmissionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name) / 'railiance-platform'
|
|
self.package = self.root.parent / 'rapp-postgres'
|
|
for folder in ['tools', 'helm', 'assurance']:
|
|
shutil.copytree(ROOT / folder, self.root / folder)
|
|
for folder in ['scripts', 'helm', 'consumers']:
|
|
shutil.copytree(ROOT.parent / 'rapp-postgres' / folder, self.package / folder)
|
|
|
|
def check(self):
|
|
return m.admission(self.root, self.package)
|
|
|
|
def test_current_occupancy_and_real_overflow(self):
|
|
result = self.check()['cells']
|
|
self.assertEqual(len(result['platform-pg']['consumers']), 4)
|
|
# platform-pg-2 took canned-prompts on 2026-09-08 (RAPP-POSTGRES-WP-0006-T03),
|
|
# so the overflow cell is 2/4 rather than 1/4.
|
|
self.assertEqual(result['platform-pg-2']['consumers'], ['canned-prompts', 'sbom-nexus'])
|
|
self.assertEqual(result['apps-pg-2']['consumers'], [])
|
|
|
|
def test_unowned_consumer_fails(self):
|
|
p = self.root / 'assurance/placement-owners.json'
|
|
data = json.loads(p.read_text()); del data['audit-core']; p.write_text(json.dumps(data))
|
|
with self.assertRaises(ValueError): self.check()
|
|
|
|
def test_native_validator_rejects_fifth_consumer(self):
|
|
p = self.package / 'consumers/sbom-nexus.yaml'
|
|
d = yaml.safe_load(p.read_text()); d['spec']['cluster'] = 'platform-pg'; p.write_text(yaml.safe_dump(d))
|
|
with self.assertRaisesRegex(ValueError, 'ceiling'): self.check()
|
|
|
|
def test_native_retention_and_effective_horizon_both_enforced(self):
|
|
p = self.package / 'consumers/audit-core.yaml'
|
|
for days in [1, 7, 60]:
|
|
d = yaml.safe_load(p.read_text()); d['spec']['retention']['backupRetentionDays'] = days
|
|
p.write_text(yaml.safe_dump(d))
|
|
with self.subTest(days=days), self.assertRaises(ValueError): self.check()
|
|
|
|
def test_missing_overflow_and_instance_policy_drift(self):
|
|
(self.package / 'helm/platform-pg-2-cluster.yaml').unlink()
|
|
with self.assertRaises(OSError): self.check()
|
|
|
|
|
|
class CollectorTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
sys.path.insert(0, str(ROOT / 'scripts'))
|
|
cls.collector = __import__('capture_service_observation')
|
|
|
|
def test_wrong_cluster_stops_before_other_reads(self):
|
|
with patch.object(self.collector, 'query', return_value='wrong') as query:
|
|
with self.assertRaises(ValueError): self.collector.capture()
|
|
self.assertEqual(query.call_count, 1)
|
|
|
|
def test_failed_native_reads_are_unavailable_not_pass(self):
|
|
contract = m.read_contract(ROOT / 'assurance/service-contract.json')
|
|
calls = []
|
|
def query(args, **kwargs):
|
|
calls.append(args)
|
|
if args[:3] == ['get', 'namespace', 'kube-system']:
|
|
return contract['cluster_uid']
|
|
raise ValueError('unavailable')
|
|
baseline = json.loads((ROOT / 'assurance/admission-baseline.json').read_text())
|
|
with patch.object(self.collector, 'query', side_effect=query), patch.object(self.collector, 'admission', return_value=baseline):
|
|
observation = self.collector.capture()
|
|
for name, sample in observation['signals'].items():
|
|
if name not in ('apps-pg.restore', 'forgejo-db.restore', 'openbao.snapshot'):
|
|
self.assertEqual(sample['result'], 'unavailable')
|
|
# Recorded recovery evidence is independent of failed live status reads.
|
|
self.assertEqual(observation['signals']['apps-pg.restore']['result'], 'pass')
|
|
self.assertNotIn('secret', [str(a).lower() for call in calls for a in call])
|
|
result = m.evaluate(contract, observation, datetime.now(timezone.utc))
|
|
self.assertFalse(result['healthy'])
|
|
|
|
def test_quantity_conversion_and_invalid_units(self):
|
|
self.assertEqual(self.collector.memory_bytes('1024Mi'), 1024**3)
|
|
for value in ['NaN', '-1Gi', '1password', '']:
|
|
with self.subTest(value=value), self.assertRaises(ValueError): self.collector.memory_bytes(value)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|
|
|
|
def test_eso_token_renewal_signal_classification():
|
|
import importlib.util
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
spec = importlib.util.spec_from_file_location(
|
|
'capture', Path(__file__).resolve().parents[1] / 'scripts/capture_service_observation.py')
|
|
capture = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(capture)
|
|
now = datetime(2026, 9, 25, 12, 0, tzinfo=timezone.utc)
|
|
assert capture.renewal_signal({}, now)[0] == 'unavailable'
|
|
ok = {'lastSuccessfulTime': '2026-09-25T02:40:09Z', 'lastScheduleTime': '2026-09-25T02:40:00Z'}
|
|
assert capture.renewal_signal(ok, now) == ('pass', '2026-09-25T02:40:09Z')
|
|
failed = {'lastSuccessfulTime': '2026-09-24T02:40:09Z', 'lastScheduleTime': '2026-09-25T02:40:00Z'}
|
|
assert capture.renewal_signal(failed, now) == ('fail', '2026-09-25T02:40:00Z')
|
|
running = {'lastSuccessfulTime': '2026-09-24T02:40:09Z', 'lastScheduleTime': '2026-09-25T11:30:00Z'}
|
|
assert capture.renewal_signal(running, now)[0] == 'pass'
|