informed-decision/tests/test_lifecycle_http.py

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_route_is_unavailable(api):
    assert call(api, "/v1/decisions", method="POST", body={})[0] == 400
    assert call(api, "/v1/decisions/memo-1/accept", method="POST", body={})[0] == 404


def packet_command(api, ident="new-record"):
    import base64
    import hashlib
    memo = call(api, "/v1/decisions/memo-1")[1]
    content = b"New decision supporting evidence"
    digest = "sha256:" + hashlib.sha256(content).hexdigest()
    memo.update(id=ident, packet=[{"item_id": "doc", "label": "Evidence", "hash": digest}])
    return {"memo": memo, "documents": [{"digest": digest, "media_type": "text/plain",
             "content_base64": base64.b64encode(content).decode()}]}


def test_register_and_revise_packet_through_api_without_operator_load(api):
    command = packet_command(api)
    status, receipt, _ = call(api, "/v1/decisions", method="POST", body=command)
    assert status == 200 and receipt["resource"] == "/v1/decisions/new-record?version=1"
    assert call(api, "/v1/decisions", method="POST", body=command)[1] == receipt
    status, stored, _ = call(api, "/v1/decisions/new-record")
    assert status == 200 and stored == command["memo"]
    stored.update(version=2, brief="Revised assessment")
    revision = {"memo": stored, "expected_version": 1, "summary": "Updated assessment"}
    status, receipt2, _ = call(api, "/v1/decisions/new-record/revisions", method="POST", body=revision)
    assert status == 200 and receipt2["version"] == 2
    assert call(api, "/v1/decisions/new-record/revisions", method="POST", body=revision)[1] == receipt2
    assert call(api, "/v1/decisions/new-record", query="version=1")[1]["brief"] != "Revised assessment"
    assert call(api, "/v1/decisions/new-record/revisions", method="POST", body=revision,
                HTTP_IDEMPOTENCY_KEY="different-writer")[0] == 409


@pytest.mark.parametrize("mutate", [
    lambda c: c["memo"].update(version=True),
    lambda c: c["memo"].update(sealed="false"),
    lambda c: c["memo"].update(unknown="not allowed"),
    lambda c: c["memo"]["binding"]["target"].update(requires_new_bind="false"),
    lambda c: c["memo"]["binding"]["principal"].update(identity_verified=True),
    lambda c: c["memo"]["packet"].append(dict(c["memo"]["packet"][0])),
    lambda c: c["documents"][0].update(content_base64="not base64"),
    lambda c: c["documents"][0].update(content_base64="bm90IHRoZSBzYW1l"),
    lambda c: c["memo"].update(approval_id="invented"),
])
def test_closed_memo_and_content_validation(api, mutate):
    command = packet_command(api)
    mutate(command)
    assert call(api, "/v1/decisions", method="POST", body=command)[0] == 400
    assert api.service.repository.store.lifecycle_counts()["records"] == 1


def test_known_digest_alone_does_not_grant_custody_of_other_record(api):
    command = packet_command(api)
    assert call(api, "/v1/decisions", method="POST", body=command)[0] == 200
    command["memo"]["id"] = "another-record"
    command.pop("documents")
    assert call(api, "/v1/decisions", method="POST", body=command,
                HTTP_IDEMPOTENCY_KEY="another")[0] == 404


def test_packet_write_rolls_back_when_event_queue_fails(api):
    command = packet_command(api)
    store = api.service.repository.store
    with store._connection() as db:
        db.execute("CREATE TRIGGER fail_packet_event BEFORE INSERT ON lifecycle_pending "
                   "BEGIN SELECT RAISE(ABORT,'injected'); END")
    assert call(api, "/v1/decisions", method="POST", body=command)[0] == 503
    with store._connection() as db:
        assert db.execute("SELECT COUNT(*) FROM documents").fetchone()[0] == 0
    assert store.lifecycle_counts()["records"] == 1


@pytest.mark.parametrize("act,step,level", [("login", "acknowledge", "acknowledgment"),
                                          ("approve", "approve", "organizational")])
def test_shared_command_decoder_preserves_l0_and_l3_binding_fields(api, act, step, level):
    command = packet_command(api)
    command["memo"].update(requested_act=act, step_kind=step, binding_level=level)
    command["memo"]["binding"]["target"]["requires_new_bind"] = True
    assert call(api, "/v1/decisions", method="POST", body=command)[0] == 200
    result = call(api, "/v1/decisions/new-record")[1]
    assert result["binding"]["target"]["requires_new_bind"] is True
    assert result["requested_act"] == act and result["binding_level"] == level