tegwick
ff59199c3b
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029c-c003-7283-acc9-930b434e6cae
96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
"""Hub decision write-back behind INFD_HUB_WRITEBACK (INFD-WP-0005-T03). Offline."""
|
|
import pytest
|
|
|
|
from informed_decision.hub_writeback import Answer, build_resolution, write_back
|
|
|
|
D1 = "62aa95e4-3ccc-4867-8806-5e3db005b483"
|
|
BASE = "http://hub.test"
|
|
ON = {"INFD_HUB_WRITEBACK": "1"}
|
|
|
|
|
|
def answer(**kw):
|
|
data = dict(memo_id="infd-20261003-p01", memo_version=1, verb="accept", disposition_id="disp-1",
|
|
decided_at="2026-10-03T12:00:00Z", reviewer_display="Bernd Worsch",
|
|
approval_id="ba5ce2d8-8b6d-40be-af89-2e8c147029a3",
|
|
approval_binding_digest="sha256:" + "a" * 64, view_hash="sha256:" + "b" * 64)
|
|
data.update(kw)
|
|
return Answer(**data)
|
|
|
|
|
|
class Hub:
|
|
def __init__(self, *responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def __call__(self, method, url, body):
|
|
self.calls.append((method, url, body))
|
|
return self.responses.pop(0)
|
|
|
|
|
|
def test_rationale_carries_memo_id_and_pins():
|
|
payload = build_resolution(answer())
|
|
assert "memo infd-20261003-p01 v1" in payload["rationale"]
|
|
assert "approval ba5ce2d8" in payload["rationale"]
|
|
assert payload["decided_by"] == "Bernd Worsch via informed-decision"
|
|
|
|
|
|
def test_agents_cannot_resolve():
|
|
with pytest.raises(ValueError):
|
|
build_resolution(answer(actor_kind="agent"))
|
|
|
|
|
|
def test_flag_off_builds_but_does_not_send():
|
|
hub = Hub()
|
|
out = write_back(answer(), {"hub_decision_id": D1, "resolves": "on-answer"}, base_url=BASE,
|
|
transport=hub, env={})
|
|
assert out.status == "skipped" and out.payload and hub.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("verb,serves", [
|
|
("return", {"hub_decision_id": D1, "resolves": "on-answer"}),
|
|
("discuss", {"hub_decision_id": D1, "resolves": "on-answer"}),
|
|
("accept", {"hub_decision_id": D1, "resolves": "never"}),
|
|
("accept", None),
|
|
])
|
|
def test_not_applicable(verb, serves):
|
|
hub = Hub()
|
|
out = write_back(answer(verb=verb), serves, base_url=BASE, transport=hub, env=ON)
|
|
assert out.status == "not_applicable" and hub.calls == []
|
|
|
|
|
|
def test_resolves_when_on():
|
|
hub = Hub((200, {"id": D1, "status": "resolved"}))
|
|
out = write_back(answer(verb="decline"), {"hub_decision_id": D1, "resolves": "on-answer"},
|
|
base_url=BASE, transport=hub, env=ON)
|
|
assert out.status == "resolved"
|
|
assert hub.calls[0][:2] == ("POST", f"{BASE}/decisions/{D1}/resolve")
|
|
|
|
|
|
def test_409_is_idempotent_only_for_our_memo():
|
|
ours = Hub((409, {}), (200, {"rationale": "accept in informed-decision memo infd-20261003-p01 v1; ..."}))
|
|
assert write_back(answer(), {"hub_decision_id": D1, "resolves": "on-answer"}, base_url=BASE,
|
|
transport=ours, env=ON).status == "already_resolved"
|
|
theirs = Hub((409, {}), (200, {"rationale": "resolved by hand"}))
|
|
assert write_back(answer(), {"hub_decision_id": D1, "resolves": "on-answer"}, base_url=BASE,
|
|
transport=theirs, env=ON).status == "conflict"
|
|
|
|
|
|
def test_transport_error_is_failed_not_raised():
|
|
def broken(*_):
|
|
raise ConnectionRefusedError()
|
|
out = write_back(answer(), {"hub_decision_id": D1, "resolves": "on-answer"}, base_url=BASE,
|
|
transport=broken, env=ON)
|
|
assert out.status == "failed"
|
|
|
|
|
|
def test_bad_decision_id_refused():
|
|
with pytest.raises(ValueError):
|
|
write_back(answer(), {"hub_decision_id": "D1", "resolves": "on-answer"}, base_url=BASE,
|
|
transport=Hub(), env=ON)
|
|
|
|
|
|
@pytest.mark.parametrize('read_status,body', [(503, {}), (404, {}), (200, {}), (200, None)])
|
|
def test_unreadable_conflict_remains_retryable(read_status, body):
|
|
hub = Hub((409, {}), (read_status, body))
|
|
assert write_back(answer(), {'hub_decision_id': D1, 'resolves': 'on-answer'},
|
|
base_url=BASE, transport=hub, env=ON).status == 'failed'
|