railiance-platform/tests/test_kv_duplicate_retire.py

"""Offline tests for the duplicate KV retirement helper against a fake ``bao``."""
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import textwrap
import unittest

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / 'scripts/kv_duplicate_retire.py'
spec = importlib.util.spec_from_file_location('kv_duplicate_retire', SCRIPT)
helper = importlib.util.module_from_spec(spec)
sys.path.insert(0, str(ROOT / 'scripts'))
spec.loader.exec_module(helper)

CONFIRM = 'RETIRE activity-core/default/llm-connect-provider-secrets'
DUP = 'workloads/activity-core/default/llm-connect-provider-secrets'
CANON = 'workloads/activity-core/llm-connect/llm-connect-provider-secrets'

FAKE_BAO = textwrap.dedent('''\
    #!/usr/bin/env python3
    import json, os, sys
    sf = os.environ['FAKE_BAO_STATE']
    st = json.load(open(sf))
    st['calls'].append(sys.argv[1:])
    def save(): json.dump(st, open(sf, 'w'))
    def out(text): save(); print(text); sys.exit(0)
    a = [x for x in sys.argv[1:] if x != '-format=json']
    if a[:2] == ['policy', 'list']: out(json.dumps(sorted(st['policies'])))
    if a[:2] == ['policy', 'read']: out(st['policies'][a[2]])
    if a[0] == 'read':
        if a[1] == 'auth/token/lookup-self': out(json.dumps({'data': st['identity']}))
        rest = a[1].split('/', 2)[2]
        if rest not in st['kv']: save(); sys.stderr.write('No value found\\n'); sys.exit(2)
        n = st['kv'][rest]
        out(json.dumps({'data': {'current_version': n, 'versions': {str(i + 1): {} for i in range(n)}}}))
    if a[0] == 'delete':
        st['kv'].pop(a[1].split('/', 2)[2], None); save(); sys.exit(0)
    save(); sys.exit(9)
''')

CLEAN = {
    'platform-admin': 'path "platform/*" { capabilities = ["read", "delete"] }',
    'agent-high-risk-boundary': f'path "platform/data/{DUP}" {{ capabilities = ["deny"] }}',
    'workload-kv-read-activity-core': f'path "platform/data/{CANON}" {{\n  capabilities = ["read"]\n}}',
}


class GlobMatching(unittest.TestCase):
    def test_exact_plus_and_trailing_star(self):
        path = f'platform/data/{DUP}'
        self.assertTrue(helper.glob_matches(path, path))
        self.assertTrue(helper.glob_matches('platform/data/workloads/activity-core/*', path))
        self.assertTrue(helper.glob_matches('platform/data/workloads/+/default/*', path))
        self.assertFalse(helper.glob_matches('platform/data/workloads/activity-core/llm-connect/*', path))
        self.assertFalse(helper.glob_matches('platform/data/workloads/+/llm-connect/*', path))


class Retire(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp())
        (self.tmp / 'bin').mkdir()
        (self.tmp / 'bin/bao').write_text(FAKE_BAO)
        (self.tmp / 'bin/bao').chmod(0o755)
        self.home = self.tmp / '.warden-attended-login' / 'session'
        self.home.mkdir(parents=True)
        self.state = {'calls': [], 'identity': {'policies': ['platform-admin']},
                      'policies': dict(CLEAN), 'kv': {DUP: 1, CANON: 3}}

    def run_helper(self, *extra):
        (self.tmp / 'state.json').write_text(json.dumps(self.state))
        receipt = self.tmp / f'r{len(self.state["calls"])}.json'
        env = {k: v for k, v in os.environ.items() if k not in ('BAO_TOKEN', 'VAULT_TOKEN')}
        env.update(PATH=f'{self.tmp / "bin"}:{env["PATH"]}', HOME=str(self.home),
                   FAKE_BAO_STATE=str(self.tmp / 'state.json'))
        proc = subprocess.run([sys.executable, str(SCRIPT), '--confirm', CONFIRM,
                               '--receipt', str(receipt), *extra], env=env, capture_output=True, text=True)
        self.assertEqual((proc.stdout, proc.stderr), ('', ''))
        self.state = json.loads((self.tmp / 'state.json').read_text())
        return proc.returncode, json.loads(receipt.read_text())

    def test_retires_when_no_policy_grants(self):
        code, receipt = self.run_helper()
        self.assertEqual((code, receipt['outcome']), (0, 'retired'))
        self.assertNotIn(DUP, self.state['kv'])
        self.assertIn(CANON, self.state['kv'])
        self.assertFalse(any(c[0] == 'read' and '/data/' in c[-1] for c in self.state['calls']))

    def test_dry_run_deletes_nothing(self):
        code, receipt = self.run_helper('--dry-run')
        self.assertEqual((code, receipt['outcome']), (0, 'dry_run_would_retire'))
        self.assertIn(DUP, self.state['kv'])

    def test_wildcard_grant_blocks(self):
        self.state['policies']['coding-agent'] = 'path "platform/data/workloads/activity-core/*" { capabilities = ["read"] }'
        code, receipt = self.run_helper()
        self.assertEqual((code, receipt['outcome']), (3, 'live_policy_grants_duplicate'))
        self.assertEqual(list(receipt['granting_policies']), ['coding-agent'])
        self.assertIn(DUP, self.state['kv'])

    def test_mount_wide_metadata_survey_does_not_block(self):
        self.state['policies']['platform-readonly'] = 'path "platform/metadata/*" { capabilities = ["read", "list"] }'
        code, receipt = self.run_helper('--dry-run')
        self.assertEqual((code, receipt['outcome']), (0, 'dry_run_would_retire'))
        self.assertEqual(receipt['metadata_only_policies'], {'platform-readonly': ['platform/metadata/*']})

    def test_unexpected_versions_or_missing_canonical_refuse(self):
        self.state['kv'][DUP] = 2
        code, receipt = self.run_helper()
        self.assertEqual((code, receipt['outcome']), (2, 'duplicate_version_unexpected'))
        self.state['kv'] = {DUP: 1}
        code, receipt = self.run_helper()
        self.assertEqual((code, receipt['outcome']), (2, 'canonical_not_live'))
        self.assertIn(DUP, self.state['kv'])


if __name__ == '__main__':
    unittest.main()