tegwick
9bc581a547
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
182 lines
10 KiB
Python
182 lines
10 KiB
Python
import re
|
|
import uuid
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
|
|
from informed_decision.lifecycle_service import Caller
|
|
from informed_decision.oidc import ORIGIN
|
|
from informed_decision.web import App, SESSION_COOKIE
|
|
from test_browser_auth import login, signing_key, finish, call
|
|
from test_lifecycle import workspace
|
|
from test_lifecycle_service import service
|
|
|
|
|
|
@pytest.fixture
|
|
def browser(workspace, login):
|
|
lifecycle, _ = service(workspace)
|
|
sid = finish(login)
|
|
session = login[0].session(sid)
|
|
return App(login[0], lifecycle=lifecycle), lifecycle, session, f"{SESSION_COOKIE}={sid}"
|
|
|
|
|
|
def post(browser, path, **values):
|
|
app, _, session, cookie = browser
|
|
values.setdefault("csrf", session.csrf)
|
|
values.setdefault("key", str(uuid.uuid4()))
|
|
return call(app, path, "POST", cookie=cookie, origin=ORIGIN, body=urlencode(values).encode())
|
|
|
|
|
|
def create(browser):
|
|
response = post(browser, "/decisions/new/prepare", question="Which storage option should we adopt?",
|
|
brief="Compare cost, recovery and maintenance.", scope_id="storage", scope_label="Storage strategy")
|
|
assert response["status"] == 303
|
|
return dict(response["headers"])["Location"]
|
|
|
|
|
|
def test_human_workspace_create_refine_publish_discuss_and_retrieve(browser):
|
|
app, lifecycle, session, cookie = browser
|
|
assert 'Open the decision workspace' in call(app, cookie=cookie)["body"]
|
|
path = create(browser)
|
|
page = call(app, path, cookie=cookie)
|
|
assert page["status"] == 200 and "Storage strategy" in page["body"]
|
|
assert "Human acceptance is not available" in page["body"]
|
|
assert post(browser, path + "/comment", version="1", text="Please include restore time.")["status"] == 303
|
|
assert post(browser, path + "/publish", version="1", subjects=session.subject, groups="")["status"] == 303
|
|
assert "published" in call(app, path, cookie=cookie)["body"]
|
|
assert post(browser, path + "/revise", version="1", question="Which storage option should we adopt?",
|
|
brief="Option A restores in one hour.", summary="Added recovery estimate")["status"] == 303
|
|
assert "Revision 2" in call(app, path, cookie=cookie)["body"]
|
|
assert "Compare cost" in call(app, path, cookie=cookie, query="version=1")["body"]
|
|
listing = call(app, "/decisions", cookie=cookie, query="q=storage")
|
|
assert listing["status"] == 200 and path in listing["body"]
|
|
memo_id = path.rsplit("/", 1)[1]
|
|
assert lifecycle.repository.policy_context(session.tenant.value, memo_id)["published_audience"] is None
|
|
assert lifecycle.repository.store.lifecycle_counts()["pending"] == 4
|
|
|
|
|
|
def test_same_create_form_replay_reuses_record(browser):
|
|
fields = {"question": "Which option?", "brief": "Consider alternatives", "scope_id": "target",
|
|
"scope_label": "Target", "key": str(uuid.uuid4())}
|
|
first = post(browser, "/decisions/new/prepare", **fields)
|
|
second = post(browser, "/decisions/new/prepare", **fields)
|
|
assert dict(first["headers"])["Location"] == dict(second["headers"])["Location"]
|
|
assert browser[1].repository.store.lifecycle_counts()["records"] == 1
|
|
|
|
|
|
def test_csrf_and_origin_are_enforced_before_any_mutation(browser):
|
|
app, lifecycle, session, cookie = browser
|
|
values = {"question": "Which option?", "brief": "Consider alternatives", "scope_id": "target",
|
|
"scope_label": "Target", "key": str(uuid.uuid4()), "csrf": session.csrf}
|
|
for origin, csrf in [(None, session.csrf), ("https://other.test", session.csrf), (ORIGIN, "wrong")]:
|
|
response = call(app, "/decisions/new/prepare", "POST", cookie=cookie, origin=origin,
|
|
body=urlencode({**values, "csrf": csrf}).encode())
|
|
assert response["status"] == 403
|
|
assert lifecycle.repository.store.lifecycle_counts()["records"] == 0
|
|
|
|
|
|
def test_workspace_is_opt_in_and_requires_human_session(browser):
|
|
app, _, _, cookie = browser
|
|
assert call(app, "/decisions")["status"] == 401
|
|
assert call(App(app.login), "/decisions", cookie=cookie)["status"] == 404
|
|
|
|
|
|
def test_workspace_escapes_content_and_policy_failure_returns_no_memo(browser):
|
|
app, lifecycle, _, cookie = browser
|
|
path = create(browser)
|
|
post(browser, path + "/comment", version="1", text='<script>alert("x")</script>')
|
|
page = call(app, path, cookie=cookie)["body"]
|
|
assert "<script>" not in page and "<script>" in page
|
|
lifecycle.policy.transport.change = lambda data: data.update(effect="deny")
|
|
response = call(app, path, cookie=cookie)
|
|
assert response["status"] == 403 and "Compare cost" not in response["body"]
|
|
|
|
|
|
def approval_draft(browser):
|
|
response = post(browser, '/decisions/new/prepare', question='Approve the storage proposal?',
|
|
brief='Compare recovery time and costs.', scope_id='storage', scope_label='Storage strategy',
|
|
review_profile='approval')
|
|
assert response['status'] == 303
|
|
return dict(response['headers'])['Location']
|
|
|
|
|
|
def test_browser_authors_packet_and_exports_exact_durable_review_handoff(browser):
|
|
import json
|
|
from informed_decision.lifecycle import LifecycleRepository
|
|
from informed_decision.lifecycle_service import LifecycleService
|
|
from informed_decision.store import Store
|
|
app, lifecycle, session, cookie = browser
|
|
path = approval_draft(browser)
|
|
fields = dict(version='1', label='Storage proposal', content='Recover in one hour. Cost: 50 EUR.',
|
|
highlight='This is a production change.', summary='Added the proposal', key=str(uuid.uuid4()))
|
|
assert post(browser, path+'/document', **fields)['status'] == 303
|
|
assert post(browser, path+'/document', **fields)['status'] == 303
|
|
memo_id = path.split('/')[-1]
|
|
assert lifecycle.repository.store.memo(memo_id).version == 2
|
|
assert lifecycle.repository.store.memo(memo_id, 1).packet == ()
|
|
packet = call(app, path+'/packet/0', cookie=cookie, query='version=2')
|
|
assert packet['status'] == 200 and packet['body'] == fields['content']
|
|
comparison = call(app, path, cookie=cookie)['body']
|
|
assert 'Supporting documents changed' in comparison and 'Review acknowledgments changed' in comparison
|
|
assert fields['highlight'] in comparison
|
|
assert post(browser, path+'/publish', version='2', subjects='', groups='net-kingdom-admins')['status'] == 303
|
|
review = dict(version='2', principal='storage-owner', purpose='Approve the recovery proposal', key=str(uuid.uuid4()))
|
|
assert post(browser, path+'/request-review', **review)['status'] == 303
|
|
assert post(browser, path+'/request-review', **review)['status'] == 303
|
|
page = call(app, path, cookie=cookie)
|
|
assert 'Download review request' in page['body'] and 'has not created an approval' in page['body']
|
|
exported = call(app, path+'/review-request', cookie=cookie, query='version=2')
|
|
assert exported['status'] == 200 and 'attachment;' in dict(exported['headers'])['Content-Disposition']
|
|
envelope = json.loads(exported['body'])
|
|
assert envelope['packet'][0]['content'] == fields['content']
|
|
assert envelope['memo']['highlights'][0]['required_ack'] is True
|
|
assert envelope['memo']['binding']['principal']['id'] == 'pending-human-session'
|
|
assert envelope['approval_intent']['principal'] == 'storage-owner'
|
|
assert envelope['memo_class'] == 'decision-record' and envelope['review_group'] == 'net-kingdom-admins'
|
|
reopened = LifecycleService(LifecycleRepository(Store(lifecycle.repository.store.path)), lifecycle.policy)
|
|
retained = reopened.review_handoff(Caller.from_human_session(session), memo_id, 2)
|
|
assert retained['envelope'] == envelope
|
|
with reopened.repository.store._connection() as db:
|
|
assert db.execute('select count(*) from intake_ledger').fetchone()[0] == 0
|
|
assert db.execute('select count(*) from submissions').fetchone()[0] == 0
|
|
assert post(browser, path+'/revise', version='2', question='Approve the revised storage proposal?',
|
|
brief='A different proposal.', summary='Changed the proposal')['status'] == 303
|
|
current = call(app, path, cookie=cookie)['body']
|
|
assert 'Download review request' not in current
|
|
assert 'earlier revision' in call(app, path, cookie=cookie, query='version=2')['body']
|
|
|
|
|
|
def test_packet_conflict_and_rejected_content_leave_no_partial_revision(browser):
|
|
app, lifecycle, session, cookie = browser
|
|
path = approval_draft(browser)
|
|
fields = dict(version='1', label='Document', content='Supporting text', highlight='', summary='Add document')
|
|
assert post(browser, path+'/document', **fields)['status'] == 303
|
|
assert post(browser, path+'/document', **{**fields, 'content':'Concurrent replacement'})['status'] == 409
|
|
assert post(browser, path+'/document', **{**fields, 'version':'2', 'content':'-----BEGIN OPENSSH PRIVATE KEY-----'})['status'] == 400
|
|
assert lifecycle.repository.store.memo(path.split('/')[-1]).version == 2
|
|
assert len(lifecycle.repository.store.memo(path.split('/')[-1]).packet) == 1
|
|
assert call(app, path+'/review-request', cookie=cookie, query='version=2')['status'] == 404
|
|
|
|
|
|
def test_review_and_document_mutations_require_policy_origin_and_csrf(browser):
|
|
app, lifecycle, session, cookie = browser
|
|
path = approval_draft(browser)
|
|
fields = dict(version='1', label='Document', content='Supporting text', highlight='', summary='Add document')
|
|
assert post(browser, path+'/document', **fields, csrf='wrong')['status'] == 403
|
|
assert post(browser, path+'/request-review', version='1', principal='owner', purpose='Review', csrf='wrong')['status'] == 403
|
|
lifecycle.policy.transport.change = lambda data: data.update(effect='deny')
|
|
assert post(browser, path+'/document', **fields)['status'] == 403
|
|
assert call(app, path+'/review-request', cookie=cookie, query='version=1')['status'] == 403
|
|
assert lifecycle.repository.store.memo(path.split('/')[-1]).version == 1
|
|
|
|
|
|
def test_review_profiles_are_explicit_and_no_browser_route_binds(browser):
|
|
app, lifecycle, session, cookie = browser
|
|
form = call(app, '/decisions/new/prepare', cookie=cookie)['body']
|
|
assert 'Human approval (organizational)' in form and 'Discussion only' in form
|
|
response = post(browser, '/decisions/new/prepare', question='Discuss the storage proposal?', brief='Discuss alternatives.',
|
|
scope_id='storage', scope_label='Storage', review_profile='discussion')
|
|
path = dict(response['headers'])['Location']
|
|
assert 'not supported by the current human approval flow' in call(app, path, cookie=cookie)['body']
|
|
assert post(browser, path+'/request-review', version='1', principal='owner', purpose='Review')['status'] == 400
|
|
assert post(browser, path+'/accept', version='1')['status'] == 404
|