tegwick
0ef57537ac
Candidate files: review policy pin v14 (sha256:8911458f...), the nk-bind-executor present_existing registry row (daily limit 10), and an 8081 ingress peer for exactly sso + app.kubernetes.io/name=bounded-executor. The state-hub row is unchanged and outcome_read stays off. The image pin stays at the live f5588f49 because no owner-CI full-service image exists yet; the founder runbook (deploy/lifecycle/RELEASE-INFD-WP-0010-T02.md) gates the release on it. Adds a side-effect-free present_existing probe and release tests. Nothing applied. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 5502@bnt-lap001 Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196
89 lines
4.5 KiB
Python
89 lines
4.5 KiB
Python
"""INFD-WP-0010-T02 release candidate: registry row, network admission, policy pin and probe."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import uuid
|
|
|
|
from informed_decision.intake_service import IntakeError, configuration
|
|
from test_intake_present_existing import ACTOR, approval, setup # noqa: F401 (fixture)
|
|
from test_intake_service import request
|
|
|
|
ROOT = Path(__file__).parents[1]
|
|
LIFECYCLE = ROOT / 'deploy' / 'lifecycle'
|
|
|
|
|
|
def _tool(name):
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / 'tools' / f'{name}.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_registry_rows_and_projection():
|
|
submitters, _ = configuration(LIFECYCLE / 'intake-submitters.json', LIFECYCLE / 'review-groups.json')
|
|
assert submitters['system:serviceaccount:state-hub:state-hub'] == {
|
|
'submitter_id': 'state-hub', 'memo_classes': ['decision-record'],
|
|
'review_groups': ['net-kingdom-admins'], 'daily_limit': 50} # unchanged; outcome_read waits on T05
|
|
assert submitters['system:serviceaccount:sso:nk-bind-executor'] == {
|
|
'submitter_id': 'nk-bind-executor', 'memo_classes': ['bounded-execution'],
|
|
'review_groups': ['net-kingdom-admins'], 'daily_limit': 10,
|
|
'present_existing': {'approval_actor': 'service:nk-bind-executor', 'actions': ['bounded-execution']}}
|
|
assert len(submitters) == 2
|
|
projection = json.loads((LIFECYCLE / 'registries.candidate.json').read_text())['data']
|
|
for name, raw in projection.items():
|
|
assert (LIFECYCLE / name).read_text() == raw
|
|
|
|
|
|
def test_candidate_render_pins_v14_and_admits_only_the_executor_on_8081():
|
|
render = _tool('render_deployment').render
|
|
admission = json.loads((LIFECYCLE / 'admission.candidate.json').read_text())
|
|
rendered = render(admission)
|
|
assert rendered == json.loads((LIFECYCLE / 'service.candidate.json').read_text())
|
|
config = json.loads(next(i for i in rendered['items'] if i['kind'] == 'ConfigMap')['data']['runtime.json'])
|
|
assert config['policy']['version'] == 'v14'
|
|
assert config['policy']['package_digest'] == (
|
|
'sha256:8911458f36f1a60598aa19592a2d752a74dffe2617fd86238407ac87bc300fde')
|
|
netpol = next(i for i in rendered['items'] if i['metadata']['name'] == 'informed-decision-review')
|
|
private = [r for r in netpol['spec']['ingress'] if r['ports'] == [{'protocol': 'TCP', 'port': 8081}]]
|
|
assert len(private) == 1
|
|
assert private[0]['from'] == [
|
|
{'namespaceSelector': {'matchLabels': {'kubernetes.io/metadata.name': ns}},
|
|
'podSelector': {'matchLabels': {'app.kubernetes.io/name': app}}}
|
|
for ns, app in (('state-hub', 'state-hub'), ('sso', 'bounded-executor'))]
|
|
|
|
|
|
def _ledger_rows(store):
|
|
with store._transaction() as db:
|
|
return db.execute('SELECT COUNT(*) FROM intake_ledger').fetchone()[0]
|
|
|
|
|
|
def test_probe_envelope_is_refused_after_a_read_and_leaves_nothing(setup): # noqa: F811
|
|
service, store, created, reads, current = setup
|
|
probe = _tool('present_existing_probe')
|
|
approval_id = str(uuid.uuid4())
|
|
current['body'] = approval(id=approval_id, **{'binding.actor': 'informed-decision'})
|
|
status, body = request(service, probe.build(approval_id, '20261101000000'), token='executor-token')
|
|
assert (status, body.get('error')) == (409, 'approval_actor_mismatch')
|
|
assert reads == [approval_id] and created == [] and _ledger_rows(store) == 0
|
|
|
|
|
|
def test_probe_envelope_with_unreadable_approval(setup): # noqa: F811
|
|
service, store, created, reads, current = setup
|
|
probe = _tool('present_existing_probe')
|
|
service.reader = lambda approval_id: (_ for _ in ()).throw(IntakeError(503, 'approval_unreadable'))
|
|
status, body = request(service, probe.build(str(uuid.uuid4()), '20261101000001'), token='executor-token')
|
|
assert (status, body.get('error')) == (503, 'approval_unreadable')
|
|
assert created == [] and _ledger_rows(store) == 0
|
|
|
|
|
|
def test_probe_program_is_self_contained_python(capsys, monkeypatch):
|
|
probe = _tool('present_existing_probe')
|
|
monkeypatch.setattr(sys, 'argv', ['probe', '--approval-id', str(uuid.uuid4()), '--program', '--stamp', '20261101000002'])
|
|
probe.main()
|
|
program = capsys.readouterr().out
|
|
compile(program, 'probe', 'exec')
|
|
namespace = {}
|
|
exec(program.split('\n', 1)[0], namespace) # first line only: the envelope literal
|
|
assert json.loads(namespace['ENVELOPE'])['memo_class'] == 'bounded-execution'
|
|
assert probe.WORKLOAD.endswith(':8081/v1/memos') and 'Bearer' in program
|