tegwick
5fec7de4cb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
92 lines
4.5 KiB
Python
92 lines
4.5 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"]
|