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) self.assertEqual(result['platform-pg-2']['consumers'], ['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()