tegwick
815fbecbf0
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e387-534d-70e3-ad53-4ea05676db8c
96 lines
4.7 KiB
Python
96 lines
4.7 KiB
Python
import copy
|
|
import json
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from informed_decision.approval_http import ApprovalHTTPClient, ordinary_review_ids
|
|
from informed_decision.approval_client import ApprovalEngineError
|
|
from informed_decision.provenance import Claim, Route, HumanControlNotDischargeable
|
|
from test_approval_http import APPROVAL, ORIGIN, Responses, session
|
|
from test_deployment_candidate import inputs, module
|
|
from test_review_runtime import config
|
|
from informed_decision.runtime import Runtime
|
|
|
|
|
|
def test_human_can_review_only_the_explicit_ordinary_record():
|
|
ordinary = copy.deepcopy(APPROVAL)
|
|
ordinary['binding']['human_control'] = False
|
|
transport = Responses((200, ordinary), (200, ordinary))
|
|
result = ApprovalHTTPClient(ORIGIN, session(), transport=transport,
|
|
ordinary_approval_ids=['fixture']).add_entry('fixture')
|
|
assert result.correlation == ('fixture', 'human-fixture', '2026-09-10T00:00:00Z')
|
|
assert transport.calls[-1][2]['body'] == b'{}'
|
|
transport = Responses((200, ordinary))
|
|
with pytest.raises(ApprovalEngineError, match='invalid_approval_response'):
|
|
ApprovalHTTPClient(ORIGIN, session(), transport=transport,
|
|
ordinary_approval_ids=['different-id']).add_entry('fixture')
|
|
assert [c[0] for c in transport.calls] == ['GET']
|
|
|
|
|
|
def test_ordinary_opt_in_cannot_make_a_service_a_human():
|
|
service = replace(session(), principal_type=Claim('service', Route.AUTHENTICATION))
|
|
with pytest.raises(HumanControlNotDischargeable):
|
|
ApprovalHTTPClient(ORIGIN, service, ordinary_approval_ids=['fixture'])
|
|
|
|
|
|
@pytest.mark.parametrize('value', [None, True, '*', ['*'], ['a', 'a'], ['x/y'], [1], list(map(str, range(33)))])
|
|
def test_invalid_ordinary_admission_refused(value):
|
|
with pytest.raises(ValueError):
|
|
ordinary_review_ids(value)
|
|
|
|
|
|
def test_renderer_pins_exact_ordinary_ids_in_immutable_configuration():
|
|
base = module.render(inputs())
|
|
data = inputs(); data['ordinary_approval_ids'] = ['worker-verify', 'worker-apply']
|
|
rendered = module.render(data)
|
|
old = next(o for o in base['items'] if o['kind'] == 'ConfigMap')
|
|
new = next(o for o in rendered['items'] if o['kind'] == 'ConfigMap')
|
|
assert new['metadata']['name'] != old['metadata']['name']
|
|
assert new['immutable'] is True
|
|
assert json.loads(new['data']['runtime.json'])['ordinary_approval_ids'] == ['worker-apply', 'worker-verify']
|
|
assert 'ordinary_approval_ids' not in json.loads(old['data']['runtime.json'])
|
|
data['ordinary_approval_ids'] = ['*']
|
|
with pytest.raises(ValueError): module.render(data)
|
|
|
|
|
|
def test_runtime_refuses_invalid_ordinary_ids_before_credentials(tmp_path):
|
|
path, data = config(tmp_path)
|
|
data['ordinary_approval_ids'] = ['*']; path.write_text(json.dumps(data))
|
|
with pytest.raises(ValueError, match='exact ordinary approval ids'):
|
|
Runtime.from_file(path)
|
|
|
|
|
|
def test_runtime_passes_only_configured_ids_to_human_client(tmp_path):
|
|
path, data = config(tmp_path)
|
|
data['ordinary_approval_ids'] = ['worker-apply']; path.write_text(json.dumps(data))
|
|
runtime = Runtime.from_file(path)
|
|
assert runtime.controller.approval_factory(session()).ordinary_approval_ids == frozenset({'worker-apply'})
|
|
|
|
|
|
from test_approval_component import component
|
|
from test_browser_auth import signing_key
|
|
|
|
|
|
def test_real_engine_retains_ordinary_control_and_records_actual_human(component):
|
|
from datetime import datetime, timedelta, timezone
|
|
_, engine, transport, human, _ = component
|
|
now = datetime.now(timezone.utc)
|
|
engine.create({'actor':'synthetic-requester','principal':'factory-fixture',
|
|
'action':'deliver','purpose':'ordinary-component-proof','target':{'resource':'fixture'}},
|
|
{'not_before':(now-timedelta(seconds=1)).isoformat(),
|
|
'expires_at':(now+timedelta(minutes=10)).isoformat()},
|
|
human_control=False, approval_id='ordinary-fixture')
|
|
strict = ApprovalHTTPClient('https://approval.test', human, transport=transport)
|
|
with pytest.raises(ApprovalEngineError, match='invalid_approval_response'):
|
|
strict.add_entry('ordinary-fixture')
|
|
assert not engine.get('ordinary-fixture').entries
|
|
admitted = ApprovalHTTPClient('https://approval.test', human, transport=transport,
|
|
ordinary_approval_ids=['ordinary-fixture'])
|
|
result = admitted.add_entry('ordinary-fixture')
|
|
record = engine.get('ordinary-fixture').as_dict()
|
|
assert result.status == 'approved' and record['binding']['human_control'] is False
|
|
assert len(record['entries']) == 1
|
|
assert record['entries'][0]['principal_type'] == 'human'
|
|
assert record['entries'][0]['subject_id'] == human.subject
|
|
assert not any(url.endswith('/consume') for _, url in transport.calls)
|