tegwick
5fec7de4cb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
111 lines
5.5 KiB
Python
111 lines
5.5 KiB
Python
from dataclasses import replace
|
|
import json
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from informed_decision.lifecycle_service import LifecycleError, LifecycleService
|
|
from informed_decision.store import Conflict, EvidenceUnavailable, Store
|
|
from test_lifecycle import AGENT, workspace
|
|
from test_lifecycle_service import caller, service
|
|
from test_policy_client import policy
|
|
|
|
|
|
AUDIENCE = {"subjects": ["reviewer"], "groups": []}
|
|
|
|
|
|
def test_publication_is_revision_pinned_durable_and_replayable(workspace):
|
|
app, memo = service(workspace)
|
|
actor = caller()
|
|
app.register(actor, memo, key="register")
|
|
receipt = app.publish(actor, memo.id, 1, audience=AUDIENCE, key="publish")
|
|
assert app.publish(actor, memo.id, 1, audience=AUDIENCE, key="publish") == receipt
|
|
state = app.repository.policy_context(AGENT.tenant, memo.id)
|
|
assert state["published_audience"] == AUDIENCE
|
|
reopened = type(app.repository)(Store(app.repository.store.path))
|
|
assert reopened.policy_context(AGENT.tenant, memo.id) == state
|
|
assert reopened.events(AGENT.tenant, memo.id)[-1]["kind"] == "published"
|
|
assert app.policy.transport.calls[1][3]["context"]["publication_audience"] == AUDIENCE
|
|
app.revise(actor, memo.next_version(brief="revised"), expected_version=1, summary="Changed", key="revise")
|
|
assert app.repository.policy_context(AGENT.tenant, memo.id)["published_audience"] is None
|
|
assert app.repository.policy_context(AGENT.tenant, memo.id, 1)["published_audience"] == AUDIENCE
|
|
with pytest.raises(Conflict):
|
|
app.publish(actor, memo.id, 1, audience=AUDIENCE, key="new-publish")
|
|
with app.repository.store._connection() as db:
|
|
assert db.execute("SELECT count(*) FROM dispositions").fetchone()[0] == 0
|
|
assert db.execute("SELECT count(*) FROM submissions").fetchone()[0] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("audience", [{}, {"public": True}, {"subjects": [], "groups": []},
|
|
{"subjects": ["*"], "groups": []},
|
|
{"subjects": "reviewer", "groups": []}])
|
|
def test_publication_requires_explicit_bounded_audience(workspace, audience):
|
|
app, memo = service(workspace)
|
|
app.register(caller(), memo, key="register")
|
|
with pytest.raises(ValueError):
|
|
app.publish(caller(), memo.id, 1, audience=audience, key="publish")
|
|
assert len(app.repository.events(AGENT.tenant, memo.id)) == 1
|
|
|
|
|
|
def test_audience_is_not_local_permission_and_denial_prevents_publication(workspace):
|
|
app, memo = service(workspace)
|
|
app.register(caller(), memo, key="register")
|
|
app.policy.transport.change = lambda data: data.update(effect="deny")
|
|
with pytest.raises(LifecycleError) as error:
|
|
app.publish(caller(), memo.id, 1, audience=AUDIENCE, key="publish")
|
|
assert error.value.status == 403
|
|
assert app.repository.policy_context(AGENT.tenant, memo.id)["published_audience"] is None
|
|
|
|
|
|
def test_audience_change_during_authorization_blocks_read_and_write(workspace, monkeypatch):
|
|
app, memo = service(workspace)
|
|
app.register(caller(), memo, key="register")
|
|
original = app.repository.annotate
|
|
def changed(*args, **kwargs):
|
|
app.repository.publish(AGENT, memo.id, 1, audience=AUDIENCE, key="concurrent-publish")
|
|
return original(*args, **kwargs)
|
|
monkeypatch.setattr(app.repository, "annotate", changed)
|
|
with pytest.raises(LifecycleError, match="record_changed"):
|
|
app.annotate(caller(), memo.id, 1, kind="comment", body={"text": "pending"}, key="comment")
|
|
original_read = app.repository.retrieve
|
|
def changed_read(*args, **kwargs):
|
|
result = original_read(*args, **kwargs)
|
|
app.repository.publish(AGENT, memo.id, 1, audience={"subjects": ["another"], "groups": []}, key="replace")
|
|
return result
|
|
monkeypatch.setattr(app.repository, "retrieve", changed_read)
|
|
with pytest.raises(LifecycleError, match="record_changed"):
|
|
app.retrieve(caller(), memo.id)
|
|
|
|
|
|
def test_search_filters_denied_records_and_opaque_cursor_is_caller_scoped(workspace):
|
|
app, memo = service(workspace)
|
|
for index in range(51):
|
|
app.repository.register(AGENT, replace(memo, id=f"record-{index:03}"), key=f"register-{index}")
|
|
# Actual PolicyClient still validates these synthetic PDP responses.
|
|
def decide(data):
|
|
if data["binding"]["resource"]["id"] == "memo:record-001":
|
|
data["effect"] = "deny"
|
|
app.policy.transport.change = decide
|
|
first = app.search(caller())
|
|
assert len(first["decisions"]) == 49
|
|
assert "record-001" not in json.dumps(first)
|
|
assert first["next_cursor"] and "record-049" not in first["next_cursor"]
|
|
assert app.search(caller(), cursor=first["next_cursor"])["decisions"][0]["memo_id"] == "record-050"
|
|
with pytest.raises(ValueError):
|
|
app.search(caller(), query="different", cursor=first["next_cursor"])
|
|
foreign = replace(caller(), attribution=replace(AGENT, tenant="tenant:other"))
|
|
with pytest.raises(ValueError):
|
|
app.search(foreign, cursor=first["next_cursor"])
|
|
with pytest.raises(ValueError):
|
|
LifecycleService(app.repository, policy()).search(caller(), cursor=first["next_cursor"])
|
|
app.policy.transport.change = lambda data: data.update(effect="deny")
|
|
with pytest.raises(LifecycleError):
|
|
app.search(caller(), cursor=first["next_cursor"])
|
|
|
|
|
|
def test_search_does_not_turn_policy_failure_into_empty_success(workspace):
|
|
app, memo = service(workspace)
|
|
app.repository.register(AGENT, memo, key="register")
|
|
app.policy.transport.status = 503
|
|
with pytest.raises(LifecycleError):
|
|
app.search(caller())
|