railiance-platform/tests/test_service_assurance.py

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()