net-kingdom/tools/bounded-execution-readiness/tests/test_ready.py

"""ready.py against synthetic sibling repositories (NK-WP-0058-T03). No network, no cluster, no secret values."""
import hashlib
import importlib.util
import json
import subprocess
from pathlib import Path

import pytest
import yaml

REPO = Path(__file__).resolve().parents[3]
spec = importlib.util.spec_from_file_location('ready', REPO / 'tools/bounded-execution-readiness/ready.py')
ready = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ready)

ADMISSION = REPO / 'admissions/BEA-2026-0003.yaml'
INSIDE = ready.when('2026-11-05T08:25:00Z')
ADMIN = 'uid=admin,ou=people,dc=netkingdom,dc=local'
ROOT_DN = 'uid=platform-root,ou=people,dc=netkingdom,dc=local'
EXECUTOR = 'service:nk-bind-executor'
SA = 'system:serviceaccount:sso:nk-bind-executor'
TYPE = 'security-case-types/oidc-service-client/type.yaml'


def git(repo, *args):
    subprocess.run(['git', '-C', str(repo), *args], check=True, capture_output=True)


def lifecycle(repo, recipient, submitter='nk-bind-executor'):
    d = repo / 'deploy/lifecycle'
    d.mkdir(parents=True, exist_ok=True)
    (d / 'review-groups.json').write_text(json.dumps({'net-kingdom-admins': {'subject': recipient}}))
    (d / 'intake-submitters.json').write_text(json.dumps({SA: {
        'submitter_id': submitter, 'memo_classes': ['bounded-execution'], 'review_groups': ['net-kingdom-admins'],
        'present_existing': {'approval_actor': EXECUTOR, 'actions': ['bounded-execution']}}}))
    git(repo, 'add', '-A')
    git(repo, '-c', 'user.name=t', '-c', 'user.email=t@t', 'commit', '-qm', 'lifecycle')
    return subprocess.run(['git', '-C', str(repo), 'rev-parse', 'HEAD'], capture_output=True, text=True).stdout.strip()


def siblings(tmp_path, *, pin_recipient=ADMIN, main_recipient=None, ccr_status='approved', confirmed=True,
             type_digest=None, profile_submitter='nk-bind-executor'):
    root = tmp_path / 'siblings'
    infd = root / 'informed-decision'
    infd.mkdir(parents=True)
    git(infd, 'init', '-q')
    pin = lifecycle(infd, pin_recipient)
    if main_recipient:
        lifecycle(infd, main_recipient)
    plat = root / 'railiance-platform'
    app = plat / ready.ARGO_APP
    app.parent.mkdir(parents=True)
    app.write_text(yaml.safe_dump({'spec': {'source': {'targetRevision': pin, 'path': 'deploy/lifecycle'}}}))
    (plat / 'scripts').mkdir()
    (plat / 'scripts/credential-change.py').write_text(
        'APPLY_ALLOWED_STATUSES = {"approved"}\nPOST_APPLY_STATUSES = {"applied", "verified", "active"}\n'
        'RUNBOOK_ALLOWED_STATUSES = APPLY_ALLOWED_STATUSES | POST_APPLY_STATUSES\n')
    digest = type_digest or hashlib.sha256((REPO / TYPE).read_bytes()).hexdigest()
    (plat / ready.BROKER).write_text("ACKNOWLEDGED_TYPES = {\n    'oidc-service-client@0.2': (\n"
                                     f"        '{TYPE}',\n        '{digest}'),\n}}\n")
    ccrs = plat / 'credential-change-requests'
    ccrs.mkdir()
    for ccr in ready.CCRS:
        (ccrs / f'{ccr}-synthetic.yaml').write_text(yaml.safe_dump(
            {'id': ccr, 'status': ccr_status, 'openbao': {'auth': {'bound_claims_confirmed': confirmed}}}))
    ae = root / 'access-engine' / ready.POLICY
    ae.parent.mkdir(parents=True)
    ae.write_text('present_existing_profiles := {\n  "service:nk-bind-executor": {\n'
                  f'    "intake_submitter": "{profile_submitter}",\n    "memo_class": "bounded-execution",\n'
                  '    "approval_action": "bounded-execution",\n  },\n}\n')
    rapp = root / 'rapp-bounded-executor' / ready.RAPP_CONFIG
    rapp.parent.mkdir(parents=True)
    rapp.write_text(json.dumps({'executor_identity': EXECUTOR, 'runtime_principal': SA,
                                'approver_subjects': {'admin': ADMIN}}))
    return root


def results(r):
    return {c['check']: c['result'] for c in r['checks']}


def test_clean_chain_passes_and_names_the_founder_checks(tmp_path):
    r = ready.run(ADMISSION, 'apr-1', siblings(tmp_path), INSIDE)
    assert r['agent_checks_pass'] and r['failures'] == [] and r['warnings'] == []
    assert [f['id'] for f in r['founder_checks']] == ['write_role_and_fence', 'preflight_job', 'custody_delivery']
    assert 'APPROVAL_ID=apr-1' in r['founder_checks'][1]['command']
    assert r['values_recorded'] is False


def test_replay_2026_11_04_routing_to_platform_root_on_main_is_caught(tmp_path):
    """The Argo pin lagged a founder-run release that routed net-kingdom-admins to platform-root."""
    r = ready.run(ADMISSION, 'apr-1', siblings(tmp_path, main_recipient=ROOT_DN), INSIDE)
    assert not r['agent_checks_pass']
    assert 'recipient_is_approver:main' in r['failures']
    assert set(r['warnings']) == {'argo_pin_matches_source:review-groups.json'}
    assert results(r)['recipient_is_approver:pinned'] == 'pass'


def test_routing_elsewhere_at_the_pin(tmp_path):
    r = ready.run(ADMISSION, 'apr-1', siblings(tmp_path, pin_recipient=ROOT_DN), INSIDE)
    assert {'recipient_is_approver:pinned', 'recipient_is_approver:main'} <= set(r['failures'])


@pytest.mark.parametrize('kwargs, failed', [
    ({'ccr_status': 'proposed'}, 'CCR-2026-0058:approved'),
    ({'ccr_status': 'verified'}, None),
    ({'confirmed': False}, 'CCR-2026-0061:bindings_confirmed'),
    ({'type_digest': '0' * 64}, 'case_type_bytes_match_broker'),
    ({'profile_submitter': 'someone-else'}, 'profile_submitter_matches'),
])
def test_each_owner_contract_is_checked(tmp_path, kwargs, failed):
    r = ready.run(ADMISSION, 'apr-1', siblings(tmp_path, **kwargs), INSIDE)
    if failed is None:
        assert r['agent_checks_pass'], r['failures']
    else:
        assert failed in r['failures']


def test_closed_window_and_receipt_handling(tmp_path):
    root = siblings(tmp_path)
    r = ready.run(ADMISSION, 'apr-1', root, ready.when('2026-11-06T00:00:00Z'))
    assert 'window_not_closed' in r['failures']
    receipt = tmp_path / 'r.json'
    assert ready.main(['--admission', str(ADMISSION), '--approval-id', 'apr-1', '--receipt', str(receipt),
                       '--siblings', str(root), '--now', '2026-11-05T08:25:00Z']) == 0
    assert json.loads(receipt.read_text())['agent_checks_pass'] is True
    with pytest.raises(FileExistsError):
        ready.main(['--admission', str(ADMISSION), '--approval-id', 'apr-1', '--receipt', str(receipt),
                    '--siblings', str(root)])


def test_a_missing_sibling_fails_by_name(tmp_path):
    receipt = tmp_path / 'r.json'
    assert ready.main(['--admission', str(ADMISSION), '--approval-id', 'a', '--receipt', str(receipt),
                       '--siblings', str(tmp_path / 'none')]) == 1
    r = json.loads(receipt.read_text())
    assert r['failures'] == ['error:FileNotFoundError'] and r['agent_checks_pass'] is False