tegwick
83849b75d4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
254 lines
13 KiB
Python
254 lines
13 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import replace
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
from urllib.parse import urlencode
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from informed_decision.approval_http import ApprovalHTTPClient
|
|
from informed_decision.disposition import DispositionRefused, Verb
|
|
from informed_decision.http_transport import TransportError
|
|
from informed_decision.memo import BindingLevel, Highlight, PacketItem, Principal, Scope, StepKind
|
|
from informed_decision.policy import PolicyError
|
|
from informed_decision.review import ReviewController, ReviewError
|
|
from informed_decision.store import Conflict, Store
|
|
from informed_decision.web import App, SESSION_COOKIE
|
|
from informed_decision.oidc import ORIGIN
|
|
from test_approval_component import component, signing_key
|
|
from test_browser_auth import call
|
|
from test_policy_client import policy
|
|
from test_skeleton import make_memo
|
|
|
|
|
|
@pytest.fixture
|
|
def review(component, tmp_path):
|
|
client,engine,transport,session,_=component
|
|
private=tmp_path/'private';private.mkdir(mode=0o700)
|
|
store=Store(private/'review.sqlite')
|
|
packet=store.put_document(b'<script>private-packet-sentinel</script>\nFull fixture terms.')
|
|
memo=make_memo(approval_id='fixture',approval_binding_digest=client.get_approval('fixture')['binding']['digest'],
|
|
question='Approve the synthetic factory delivery?', requested_act='deliver',
|
|
highlights=(Highlight('h-1','doc-1','Disposable test target; no production action',required_ack=True),),
|
|
packet=(PacketItem('doc-1','Whole fixture document',packet),),
|
|
brief='Disposable browser/component exercise. private-brief-sentinel <img src=x onerror=alert(1)>')
|
|
memo=replace(memo,binding=replace(memo.binding,principal=Principal(session.subject,'person','Fixture reviewer'),
|
|
target=Scope('resource','fixture','Synthetic factory target','test')))
|
|
store.save_memo(memo)
|
|
controller=ReviewController(store,policy(),lambda s:ApprovalHTTPClient('https://approval.test',s,transport=transport))
|
|
class Login:
|
|
def session(self,sid):return session if sid=='fixture-session' else None
|
|
app=App(Login(),controller,readiness=lambda:True)
|
|
return controller,session,app,memo,engine,transport
|
|
|
|
|
|
def opened(review):
|
|
controller,session,*_=review
|
|
return controller.open(session,'memo-1')
|
|
|
|
|
|
def post(review,p,route,**params):
|
|
_,session,app,*_=review
|
|
return call(app,'/presentations/'+p.id+'/'+route,'POST',cookie=SESSION_COOKIE+'=fixture-session',
|
|
origin=ORIGIN,body=urlencode({'csrf':session.csrf,**params}).encode())
|
|
|
|
|
|
def test_read_denial_precedes_disclosure_engine_read_and_presentation(review):
|
|
controller,session,app,memo,engine,transport=review
|
|
controller.policy.transport.change=lambda d:d.update(effect='deny')
|
|
before=len(transport.calls)
|
|
r=call(app,'/review',query='memo_id=memo-1',cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert r['status']==403 and 'private-brief-sentinel' not in r['body']
|
|
assert len(transport.calls)==before and controller.store.evidence()==[]
|
|
assert controller.store.policy_observations()[0]['outcome']=='policy_denied'
|
|
|
|
|
|
def test_no_session_or_wrong_named_recipient_cannot_check_or_render(review):
|
|
controller,session,app,*_=review
|
|
assert call(app,'/review',query='memo_id=memo-1')['status']==401
|
|
with pytest.raises(ReviewError):controller.open(replace(session,subject='other'),'memo-1')
|
|
assert controller.policy.transport.calls==[] and controller.store.policy_observations()==[]
|
|
|
|
|
|
@pytest.mark.parametrize('change', [{'locale':'de'}, {'binding_level':BindingLevel.AES},
|
|
{'binding_level':BindingLevel.QES}, {'step_kind':StepKind.REVIEW}])
|
|
def test_unsupported_profile_cannot_misstate_locale_or_binding_level(review,change):
|
|
controller,session,_,memo,*_=review
|
|
controller.store.save_memo(memo.next_version(**change))
|
|
with pytest.raises(ReviewError,match='unsupported_review_profile'):
|
|
controller.open(session,memo.id)
|
|
assert controller.policy.transport.calls==[] and controller.store.evidence()==[]
|
|
|
|
|
|
def test_open_persists_exact_policy_and_binding_but_not_browser_credential(review):
|
|
controller,session,app,*_=review
|
|
page=opened(review);content=controller.store.presentation_content(page.presentation.id)
|
|
observed=controller.store.policy_observations()
|
|
assert observed[0]['decision_attributable']==0
|
|
assert content['policy_observation_id']==observed[0]['id']
|
|
assert content['approval_binding']['digest']==page.memo.approval_binding_digest
|
|
assert session.access_token not in json.dumps(observed)
|
|
assert all('policy_observation_id' not in r['envelope'] for r in controller.store.evidence())
|
|
response=call(app,'/presentations/'+page.presentation.id,cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert response['status']==200 and '<img' in response['body'] and '<img src=x' not in response['body']
|
|
assert '<script>private-packet-sentinel' in response['body']
|
|
|
|
|
|
@pytest.mark.parametrize('fault',['origin','csrf','duplicate','content-type','field'])
|
|
def test_post_form_refusals_precede_policy_and_side_effects(review,fault):
|
|
controller,session,app,*_=review;p=opened(review).presentation
|
|
body=urlencode({'csrf':session.csrf,'h0':'h-1'}).encode();origin=ORIGIN;ctype='application/x-www-form-urlencoded'
|
|
if fault=='origin':origin='https://other.test'
|
|
if fault=='csrf':body=b'csrf=wrong&h0=h-1'
|
|
if fault=='duplicate':body+=b'&h0=other'
|
|
if fault=='content-type':ctype='application/json'
|
|
if fault=='field':body+=b'&subject=other'
|
|
before=len(controller.policy.transport.calls)
|
|
r=call(app,'/presentations/'+p.id+'/ack','POST',cookie=SESSION_COOKIE+'=fixture-session',body=body,origin=origin,content_type=ctype)
|
|
assert r['status'] in (400,403) and len(controller.policy.transport.calls)==before
|
|
assert not controller.store.presentation(p.id).acked_highlight_ids
|
|
|
|
|
|
def test_highlights_gate_accept_then_one_real_entry_and_original_view(review):
|
|
controller,session,app,memo,engine,transport=review;p=opened(review).presentation
|
|
op=str(uuid.uuid4())
|
|
assert post(review,p,'act',verb='accept',operation_id=op)['status']==409
|
|
assert not engine.get('fixture').entries
|
|
assert post(review,p,'ack',h0='h-1')['status']==303
|
|
assert post(review,p,'act',verb='accept',operation_id=op)['status']==303
|
|
assert len(engine.get('fixture').entries)==1
|
|
posts=sum(method=='POST' for method,_ in transport.calls)
|
|
assert post(review,p,'act',verb='accept',operation_id=op)['status']==303
|
|
assert sum(method=='POST' for method,_ in transport.calls)==posts==1
|
|
page=controller.open(session,memo.id)
|
|
assert page.presentation.id==p.id and page.intent['state']=='confirmed'
|
|
response=call(app,'/presentations/'+p.id,cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert 'Approval entry recorded.' in response['body']
|
|
assert 'Accept the complete request</button>' not in response['body']
|
|
assert not any(url.endswith('/consume') for _,url in transport.calls)
|
|
|
|
|
|
@pytest.mark.parametrize('verb',[Verb.RETURN,Verb.DISCUSS,Verb.DECLINE])
|
|
def test_nonaccept_actions_preserve_vocabulary_without_engine_mutation(review,verb):
|
|
controller,session,app,memo,engine,transport=review;p=opened(review).presentation
|
|
controller.acknowledge(session,p.id,['h-1'])
|
|
result=post(review,p,'act',verb=verb.value,operation_id=str(uuid.uuid4()),reason='clarification_needed' if verb is Verb.RETURN else '',note='Private response')
|
|
assert result['status']==303 and not engine.get('fixture').entries
|
|
assert controller.store.dispositions_for(p.id)[0].verb is verb
|
|
response=call(app,'/presentations/'+p.id,cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert 'Private response' in response['body']
|
|
assert all('Private response' not in r['envelope'] for r in controller.store.evidence())
|
|
|
|
|
|
def test_fresh_act_refusal_after_allowed_render_cannot_submit(review):
|
|
controller,session,app,memo,engine,transport=review;p=opened(review).presentation
|
|
controller.acknowledge(session,p.id,['h-1'])
|
|
controller.policy.transport.change=lambda d:d.update(effect='deny')
|
|
assert post(review,p,'act',verb='accept',operation_id=str(uuid.uuid4()))['status']==403
|
|
assert not engine.get('fixture').entries and controller.store.dispositions_for(p.id)==[]
|
|
|
|
|
|
def test_policy_storage_failure_prevents_presentation(review):
|
|
controller,session,*_=review
|
|
with sqlite3.connect(controller.store.path) as db:
|
|
db.execute("CREATE TRIGGER fail_observation BEFORE INSERT ON policy_observations BEGIN SELECT RAISE(ABORT,'disk failure'); END")
|
|
with pytest.raises(sqlite3.IntegrityError):opened(review)
|
|
assert controller.store.evidence()==[]
|
|
|
|
|
|
def test_revision_between_policy_and_present_cannot_render_unchecked_version(review):
|
|
controller,session,_,memo,*_=review
|
|
controller.policy.transport.change=lambda d:controller.store.save_memo(memo.next_version(brief='changed'))
|
|
with pytest.raises(Conflict):opened(review)
|
|
assert controller.store.evidence()==[]
|
|
|
|
|
|
def test_stale_presentation_and_wrong_actor_cannot_ack_or_act(review):
|
|
controller,session,_,memo,engine,_=review;p=opened(review).presentation
|
|
with pytest.raises(ReviewError):controller.acknowledge(replace(session,subject='other'),p.id,['h-1'])
|
|
controller.store.save_memo(memo.next_version(brief='new version'))
|
|
with pytest.raises(ReviewError):controller.acknowledge(session,p.id,['h-1'])
|
|
with pytest.raises(ReviewError):controller.act(session,p.id,Verb.ACCEPT,operation_id='stale')
|
|
assert not engine.get('fixture').entries
|
|
|
|
|
|
def test_expiry_after_last_engine_read_prevents_reservation_and_post(review):
|
|
controller,session,_,memo,engine,transport=review;p=opened(review).presentation
|
|
controller.acknowledge(session,p.id,['h-1'])
|
|
original=transport.request;reads=0
|
|
def request(method,url,**kwargs):
|
|
nonlocal reads
|
|
result=original(method,url,**kwargs)
|
|
if method=='GET':
|
|
reads+=1
|
|
if reads==2:controller.clock=lambda:time.time()+40
|
|
return result
|
|
transport.request=request
|
|
with pytest.raises(PolicyError):controller.act(session,p.id,Verb.ACCEPT,operation_id='late')
|
|
assert not engine.get('fixture').entries
|
|
assert controller.store.intent_for('fixture',session.subject)['state']=='prepared'
|
|
|
|
|
|
def test_parallel_clicks_submit_once(review):
|
|
controller,session,_,memo,engine,transport=review;p=opened(review).presentation
|
|
controller.acknowledge(session,p.id,['h-1'])
|
|
def click(_):return controller.act(session,p.id,Verb.ACCEPT,operation_id='one-click')
|
|
with ThreadPoolExecutor(max_workers=2) as pool:assert list(pool.map(click,range(2)))==[p.id,p.id]
|
|
assert len(engine.get('fixture').entries)==1
|
|
assert sum(method=='POST' for method,_ in transport.calls)==1
|
|
|
|
|
|
def test_lost_entry_reply_is_visible_and_never_retried(review):
|
|
controller,session,app,memo,engine,transport=review;p=opened(review).presentation
|
|
controller.acknowledge(session,p.id,['h-1'])
|
|
original=transport.request
|
|
def lose_reply(method,url,**kwargs):
|
|
result=original(method,url,**kwargs)
|
|
if method=='POST':raise TransportError('lost response')
|
|
return result
|
|
transport.request=lose_reply
|
|
controller.act(session,p.id,Verb.ACCEPT,operation_id='lost')
|
|
assert controller.store.intent_for('fixture',session.subject)['state']=='unresolved'
|
|
response=call(app,'/presentations/'+p.id,cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert 'Submission outcome is not confirmed.' in response['body']
|
|
assert 'Accept the complete request</button>' not in response['body']
|
|
controller.act(session,p.id,Verb.ACCEPT,operation_id='new-click')
|
|
assert len(engine.get('fixture').entries)==1 and sum(m=='POST' for m,_ in transport.calls)==1
|
|
|
|
|
|
def test_unknown_existing_entry_never_acquires_a_new_presentation_link(review):
|
|
controller,session,_,memo,engine,transport=review;p=opened(review).presentation
|
|
ApprovalHTTPClient('https://approval.test',session,transport=transport).add_entry('fixture')
|
|
with pytest.raises(ReviewError,match='existing_entry_unlinked'):
|
|
controller.act(session,p.id,Verb.ACCEPT,operation_id='unlinked')
|
|
assert controller.store.intent_for('fixture',session.subject) is None
|
|
assert sum(m=='POST' for m,_ in transport.calls)==1
|
|
|
|
|
|
def test_packet_download_rechecks_entitlement_and_forces_attachment(review):
|
|
controller,session,app,*_=review;p=opened(review).presentation
|
|
path='/presentations/'+p.id+'/packet/0'
|
|
response=call(app,path,cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert response['status']==200
|
|
assert dict(response['headers'])['Content-Type']=='application/octet-stream'
|
|
assert dict(response['headers'])['Content-Disposition'].startswith('attachment;')
|
|
controller.policy.transport.change=lambda d:d.update(effect='deny')
|
|
denied=call(app,path,cookie=SESSION_COOKIE+'=fixture-session')
|
|
assert denied['status']==403 and 'private-packet-sentinel' not in denied['body']
|
|
|
|
|
|
def test_audit_not_ready_blocks_accept_before_policy_or_post(review):
|
|
controller,session,app,memo,engine,_=review;p=opened(review).presentation
|
|
app.readiness=lambda:False
|
|
assert post(review,p,'act',verb='accept',operation_id=str(uuid.uuid4()))['status']==503
|
|
assert not engine.get('fixture').entries
|
|
assert call(app,'/readyz')['status']==503
|
|
|
|
|
|
def test_renderer_release_cannot_silently_change_under_a_memo(review):
|
|
controller,session,_,memo,*_=review
|
|
controller.store.save_memo(memo.next_version(ui_release='informed-decision@0.1.0'))
|
|
with pytest.raises(ReviewError,match='renderer_changed'):opened(review)
|
|
assert controller.store.evidence()==[]
|