04f66f81b6
Updated by fix-consistency on 2026-10-04: - update .custodian-brief.md for informed-decision Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
92 lines
4.4 KiB
Python
92 lines
4.4 KiB
Python
import io
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from informed_decision.lifecycle_http import LifecycleAPI, MAX_BODY
|
|
from test_lifecycle import workspace
|
|
from test_lifecycle_service import caller, service
|
|
|
|
|
|
def call(api, path, *, method="GET", body=None, token="fixture", query="", raw=None, **extra):
|
|
raw = raw if raw is not None else json.dumps(body).encode() if body is not None else b""
|
|
environ = {"PATH_INFO": path, "REQUEST_METHOD": method, "QUERY_STRING": query,
|
|
"HTTP_AUTHORIZATION": "Bearer " + token if token is not None else "",
|
|
"CONTENT_TYPE": "application/json", "CONTENT_LENGTH": str(len(raw)),
|
|
"HTTP_IDEMPOTENCY_KEY": "operation-1", "wsgi.input": io.BytesIO(raw), **extra}
|
|
response = {}
|
|
payload = b"".join(api(environ, lambda status, headers: response.update(status=int(status.split()[0]), headers=dict(headers))))
|
|
return response["status"], json.loads(payload), response["headers"]
|
|
|
|
|
|
@pytest.fixture
|
|
def api(workspace):
|
|
app, memo = service(workspace)
|
|
app.register(caller(), memo, key="initial")
|
|
def authenticate(token):
|
|
if token != "fixture":
|
|
raise ValueError("secret diagnostic must not escape")
|
|
return caller()
|
|
return LifecycleAPI(app, authenticate)
|
|
|
|
|
|
def test_authenticated_retrieval_comment_replay_and_history(api):
|
|
status, memo, headers = call(api, "/v1/decisions/memo-1")
|
|
assert status == 200 and memo["version"] == 1
|
|
assert headers["Cache-Control"] == "no-store"
|
|
command = {"version": 1, "body": {"text": "Compare alternatives"}}
|
|
status, receipt, _ = call(api, "/v1/decisions/memo-1/comments", method="POST", body=command)
|
|
assert status == 200
|
|
assert call(api, "/v1/decisions/memo-1/comments", method="POST", body=command)[1] == receipt
|
|
status, result, _ = call(api, "/v1/decisions/memo-1/events", query="limit=1")
|
|
cursor = result["events"][0]["sequence"]
|
|
status, result, _ = call(api, "/v1/decisions/memo-1/events", query=f"after={cursor}")
|
|
assert result["events"][0]["id"] == receipt["event_id"]
|
|
assert len(result["events"]) == 1
|
|
|
|
|
|
def test_authentication_runs_before_body_read_and_ignores_cookie(api):
|
|
class Unreadable:
|
|
def read(self, size):
|
|
pytest.fail("unauthenticated body read")
|
|
status, result, headers = call(api, "/v1/decisions/memo-1/comments", method="POST",
|
|
token=None, HTTP_COOKIE="session=forged", **{"wsgi.input": Unreadable()})
|
|
assert status == 401 and headers["WWW-Authenticate"] == "Bearer"
|
|
assert call(api, "/v1/decisions/memo-1", token="bad")[1] == {"error": "authentication_failed"}
|
|
|
|
|
|
def test_policy_denial_does_not_disclose_content(api):
|
|
api.service.policy.transport.change = lambda data: data.update(effect="deny")
|
|
for path in ["/v1/decisions/memo-1", "/v1/decisions/absent", "/v1/decisions/memo-1/events"]:
|
|
status, result, _ = call(api, path)
|
|
assert status == 403 and result == {"error": "policy_denied"}
|
|
|
|
|
|
@pytest.mark.parametrize("query", ["version=1&version=2", "version=0", "tenant=other", "version=bad"])
|
|
def test_closed_query_contract(api, query):
|
|
assert call(api, "/v1/decisions/memo-1", query=query)[0] == 400
|
|
|
|
|
|
@pytest.mark.parametrize("raw", [
|
|
b'{"version":1,"version":2,"body":{"text":"duplicate"}}',
|
|
b'{"version":NaN,"body":{"text":"bad"}}',
|
|
b'{"version":1,"body":{"text":"hello"},"actor":"forged"}',
|
|
b'{"version":true,"body":{"text":"bad"}}', b'[]',
|
|
])
|
|
def test_malformed_commands_refused_without_event(api, raw):
|
|
assert call(api, "/v1/decisions/memo-1/comments", method="POST", raw=raw)[0] == 400
|
|
assert api.service.repository.store.lifecycle_counts()["events"] == 1
|
|
|
|
|
|
def test_body_limits_and_idempotency_conflict(api):
|
|
path = "/v1/decisions/memo-1/comments"
|
|
assert call(api, path, method="POST", CONTENT_LENGTH=str(MAX_BODY + 1))[0] == 413
|
|
assert call(api, path, method="POST", CONTENT_TYPE="text/plain")[0] == 415
|
|
assert call(api, path, method="POST", raw=b'{}', CONTENT_LENGTH="20")[0] == 400
|
|
assert call(api, path, method="POST", body={"version": 1, "body": {"text": "first"}})[0] == 200
|
|
assert call(api, path, method="POST", body={"version": 1, "body": {"text": "second"}})[0] == 409
|
|
|
|
|
|
def test_unimplemented_binding_and_creation_routes_are_unavailable(api):
|
|
assert call(api, "/v1/decisions", method="POST", body={})[0] == 404
|
|
assert call(api, "/v1/decisions/memo-1/accept", method="POST", body={})[0] == 404
|