net-kingdom/tools/bounded-execution-run/tests/test_bx.py

"""bx.py on stubs (NK-WP-0058-T04): every outside command is faked; no cluster, no OpenBao, no warden."""
import contextlib
import importlib.util
import io
import json
from pathlib import Path

import pytest

REPO = Path(__file__).resolve().parents[3]
spec = importlib.util.spec_from_file_location('bx', REPO / 'tools/bounded-execution-run/bx.py')
bx = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bx)
ADMISSION = REPO / 'admissions/BEA-2026-0003.yaml'


def arg(argv, name):
    return argv[argv.index(name) + 1]


class Fake:
    def __init__(self, tool, preflight_ready=True, window_opens=True, result='completed', logs_raise=False):
        self.tool, self.calls, self.child_output = tool, [], ''
        self.preflight_ready, self.window_opens, self.result, self.logs_raise = (
            preflight_ready, window_opens, result, logs_raise)

    def write(self, argv, data):
        Path(arg(argv, '--receipt')).write_text(json.dumps(data))

    def run(self, argv, env=None, timeout=900):
        argv = [str(a) for a in argv]
        joined = ' '.join(argv)
        if argv[:3] == ['warden', 'access', 'openbao-platform-admin-login']:
            self.calls.append('lane')
            assert env['BX_STATE_DIR'] and env['BX_KUBECONFIG']  # absolute paths: the child's HOME is isolated
            child = argv[argv.index('--') + 1:]
            if '_run-inner' in child:
                with contextlib.redirect_stdout(io.StringIO()) as out, contextlib.redirect_stderr(out):
                    self.tool.main(['_run-inner', '--state', arg(child, '--state')], bx=self.tool_bx)
                self.child_output += out.getvalue()
                return 0, ''
            return self.run(child)
        if 'ready.py' in joined:
            self.calls.append('readiness')
            self.write(argv, {'agent_checks_pass': True, 'failures': []})
            return 0, ''
        if 'externalsecret' in joined:
            self.calls.append('custody')
            return 0, 'True'
        if 'executor-lanes.py' in joined:
            action = argv[2]
            self.calls.append(action)
            assert '--quiet' in argv
            if action == 'status':
                self.write(argv, {'status': 'status_read', 'ready_for_window': True})
            elif action == 'window-open':
                self.write(argv, {'status': 'window_opened' if self.window_opens else 'failed'})
            elif action == 'window-close':
                self.write(argv, {'status': 'window_closed'})
            return 0, ''
        if argv[:1] == ['make']:
            target = argv[3]
            self.calls.append('make ' + target)
            if target == 'request':
                return 0, 'kubectl ...\n{\n  "approval_id": "apr-9",\n  "outcome": "presented"\n}\n'
            if target == 'preflight':
                report = {'mode': 'preflight-only', 'ready': self.preflight_ready, 'failures':
                          [] if self.preflight_ready else ['preflight:approval']}
                return (0 if self.preflight_ready else 2), json.dumps(report, indent=2) + '\n'
            if target == 'run':
                return 0, 'kubectl create -f - < .rendered/run.yaml\n'
        if 'get' in argv and any(a.startswith('job/') for a in argv):
            self.calls.append('poll')
            return 0, '1'
        if 'logs' in argv:
            self.calls.append('logs')
            if self.logs_raise:
                raise TimeoutError
            return 0, json.dumps({'outcome': self.result, 'run_id': 'r-1', 'effects': True}, indent=2) + '\n'
        raise AssertionError(argv)

    def sleep(self, seconds):
        pass


@pytest.fixture
def env(tmp_path, monkeypatch):
    monkeypatch.setattr(bx, 'STATE_DIR', tmp_path / 'state')
    monkeypatch.setattr(bx, 'REPO', tmp_path / 'repo')
    rapp = tmp_path / 'rapp'
    (rapp / 'examples').mkdir(parents=True)
    (rapp / 'examples/config.pilot.json').write_text(json.dumps({'executor_identity': 'service:nk-bind-executor'}))
    monkeypatch.setattr(bx, 'RAPP', rapp)
    ccrs = tmp_path / 'platform/credential-change-requests'
    ccrs.mkdir(parents=True)
    for name in ('CCR-2026-0058-w.yaml', 'CCR-2026-0059-j.yaml'):
        (ccrs / name).write_text('id: x\n')
    monkeypatch.setattr(bx, 'PLATFORM', tmp_path / 'platform')
    return tmp_path


def tool(**kw):
    fake = Fake(bx, **kw)
    fake.tool_bx = bx.Bx(fake)
    return fake, fake.tool_bx


def cli(b, *argv):
    with contextlib.redirect_stdout(io.StringIO()) as out:
        assert bx.main(list(argv), bx=b) == 0
    lines = out.getvalue().splitlines()
    assert len(lines) == 1  # one value-free line per command
    return json.loads(lines[0])


def test_full_run_is_four_commands_plus_the_approval(env):
    fake, b = tool()
    assert cli(b, 'ready', str(ADMISSION))['next'] == 'bx request'
    assert cli(b, 'request')['approval_id'] == 'apr-9'
    done = cli(b, 'run')
    assert (done['outcome'], done['window_closed'], done['receipts']) == ('completed', True, 4)
    assert cli(b, 'status')['outcome'] == 'completed'
    assert fake.calls == ['readiness', 'custody', 'custody', 'lane', 'status', 'make request', 'lane',
                          'make preflight', 'window-open', 'make run', 'poll', 'logs', 'window-close']
    assert fake.child_output == ''  # the lane child is silent
    evidence = sorted(p.name for p in (env / 'repo/docs/evidence').glob('*/*'))
    assert [n[:2] for n in evidence] == [f'{i:02d}' for i in range(1, 9)]
    assert evidence[4:] == ['05-preflight.json', '06-window-open.json', '07-result.json', '08-window-close.json']
    state = json.loads((env / 'state/BEA-2026-0003.json').read_text())
    assert state['attest'] == ['runtime_secret_trust', 'runtime_identity_trust']
    cfg = json.loads((env / 'state/BEA-2026-0003.config.json').read_text())
    assert cfg['trust_satisfied'] == state['attest']


def test_commands_are_safe_to_repeat(env):
    fake, b = tool()
    cli(b, 'ready', str(ADMISSION))
    cli(b, 'request')
    assert cli(b, 'request')['status'] == 'already_requested'
    cli(b, 'run')
    calls = len(fake.calls)
    assert cli(b, 'run')['status'] == 'already_run'
    assert len(fake.calls) == calls  # no second run, no second login


def test_preflight_not_ready_never_opens_the_window(env):
    fake, b = tool(preflight_ready=False)
    cli(b, 'ready', str(ADMISSION))
    cli(b, 'request')
    done = cli(b, 'run')
    assert (done['outcome'], done['reason']) == ('not_started', 'preflight_not_ready')
    assert 'window-open' not in fake.calls and 'make run' not in fake.calls
    fake.preflight_ready = True
    assert cli(b, 'run')['outcome'] == 'completed'  # a run that never started may be retried


@pytest.mark.parametrize('kw, outcome', [({'window_opens': False}, 'not_started'), ({'result': 'reverted'}, 'reverted')])
def test_the_window_is_always_closed(env, kw, outcome):
    fake, b = tool(**kw)
    cli(b, 'ready', str(ADMISSION))
    cli(b, 'request')
    done = cli(b, 'run')
    assert done['outcome'] == outcome and fake.calls[-1] == 'window-close' and done['window_closed']


def test_an_error_in_the_child_still_closes_and_stays_silent(env):
    fake, b = tool(logs_raise=True)
    cli(b, 'ready', str(ADMISSION))
    cli(b, 'request')
    done = cli(b, 'run')
    assert done['error'] == 'TimeoutError' and done['window_closed'] is True
    assert fake.calls[-1] == 'window-close' and fake.child_output == ''


def test_close_is_one_lane_and_idempotent(env):
    fake, b = tool()
    cli(b, 'ready', str(ADMISSION))
    assert cli(b, 'close')['status'] == 'window_closed'
    assert cli(b, 'close')['status'] == 'window_closed'


def test_instructions_fit_120_characters():
    assert all(len(line) <= 120 for line in bx.__doc__.splitlines())