tegwick
b4f637326b
state-hub d61ba32 accepts requires_new_bind, required_ack, and sealed. The p01 cross-repo check now expects a delivered pose that keeps those bools. T08 stays waiting on the service account, grant, and T05/T06. Assistant: grok Assistant-Session: 01a10286-9185-7bb2-b411-89cf6ab4c0c6
201 lines
9.4 KiB
Python
201 lines
9.4 KiB
Python
"""Delivery contracts, recovery and opt-in runtime composition."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from informed_decision.audit import AuditDeliveryError, OutboxWorker
|
|
from informed_decision.evidence import EventClass
|
|
from informed_decision.intake_delivery import IntakeDeliveryWorker
|
|
from informed_decision.intake_hub import HubDelivery
|
|
from informed_decision.runtime import AuditPump, Runtime
|
|
from informed_decision.store import Store
|
|
from test_durable_component import receiver
|
|
from test_intake import envelope
|
|
from test_intake_service import setup, request
|
|
from test_review_runtime import config
|
|
|
|
|
|
class HubTransport:
|
|
def __init__(self, status=201):
|
|
self.status, self.calls = status, []
|
|
|
|
def request(self, method, url, *, headers, body):
|
|
payload = json.loads(body) if body else None
|
|
self.calls.append((method, url, headers, body))
|
|
if url.endswith('/notifications'):
|
|
receipt = {'message_id': str(uuid.uuid4()), 'delivery_id': headers['Idempotency-Key'],
|
|
'status': 'accepted', 'authoritative_review': 'signed-in-overview', **payload}
|
|
else:
|
|
receipt = {'id': url.split('/decisions/')[1].split('/')[0],
|
|
'status': 'resolved', **(payload or {})}
|
|
return self.status, receipt
|
|
|
|
|
|
def test_notification_replay_uses_identical_request_and_checks_receipt():
|
|
transport = HubTransport()
|
|
hub = HubDelivery('https://hub.test', transport=transport)
|
|
ident = str(uuid.uuid4())
|
|
metadata = {'memo_id': 'review-1', 'question': 'Proceed?', 'memo_class': 'credential-lane', 'submitter': 'state-hub'}
|
|
assert hub.notify(ident, metadata)
|
|
assert hub.notify(ident, metadata)
|
|
assert transport.calls[0] == transport.calls[1]
|
|
assert json.loads(transport.calls[0][3]) == metadata
|
|
assert transport.calls[0][1].endswith('/informed-decision/notifications')
|
|
assert transport.calls[0][2]['Idempotency-Key'] == ident
|
|
transport.status = 202
|
|
assert not hub.notify(ident, metadata) # Relay queue acknowledgement is not a receipt.
|
|
with pytest.raises(ValueError):
|
|
hub.notify(ident, {**metadata, 'packet': 'private'})
|
|
transport.request = lambda *args, **kwargs: (201, {'id': str(uuid.uuid4())})
|
|
assert not hub.notify(ident, metadata)
|
|
|
|
|
|
def test_resolution_is_fixed_origin_and_idempotent():
|
|
transport = HubTransport(status=200)
|
|
hub = HubDelivery('https://hub.test', transport=transport)
|
|
url = 'https://hub.test/decisions/' + str(uuid.uuid4())
|
|
for _ in range(2):
|
|
hub.resolution('POST', url + '/resolve', {'rationale': 'memo review-1', 'decided_by': 'human'})
|
|
assert transport.calls[0] == transport.calls[1]
|
|
hub.resolution('GET', url, {})
|
|
transport.status = 202
|
|
assert hub.resolution('POST', url + '/resolve', {'rationale': 'memo review-1', 'decided_by': 'human'}) == (503, {})
|
|
for method, target in [('GET', url + '/resolve'), ('POST', url),
|
|
('POST', url.replace('hub.test', 'other.test') + '/resolve'),
|
|
('POST', 'https://hub.test/decisions/' + '-' * 36 + '/resolve')]:
|
|
with pytest.raises(ValueError):
|
|
hub.resolution(method, target, {})
|
|
|
|
|
|
def test_runtime_delivery_is_opt_in_and_writeback_remains_off(tmp_path, monkeypatch):
|
|
monkeypatch.delenv('INFD_HUB_WRITEBACK', raising=False)
|
|
path, data = config(tmp_path)
|
|
assert Runtime.from_file(path).pump.intake_worker is None
|
|
data['intake_delivery'] = {'hub_origin': 'https://hub.test'}
|
|
path.write_text(json.dumps(data))
|
|
worker = Runtime.from_file(path).pump.intake_worker
|
|
assert worker is not None and worker.audit and worker.notify and worker.hub_transport
|
|
from informed_decision.hub_writeback import enabled
|
|
assert not enabled(worker.env)
|
|
for invalid in [None, {}, {**data['intake_delivery'], 'allow_all': True},
|
|
{'hub_origin': 'http://public.example.com'}]:
|
|
data['intake_delivery'] = invalid
|
|
path.write_text(json.dumps(data))
|
|
with pytest.raises(ValueError):
|
|
Runtime.from_file(path)
|
|
|
|
|
|
def test_posed_evidence_survives_restart_and_is_immutable(setup):
|
|
service, store, _, _ = setup
|
|
assert request(service, envelope())[0] == 201
|
|
store.stage_intake_audit()
|
|
before = store.evidence()
|
|
assert len(before) == 1 and before[0]['class'] == EventClass.MEMO_POSED.value
|
|
Store(store.path).stage_intake_audit()
|
|
assert store.evidence() == before and len(store.outbox()) == 1
|
|
with store._connection() as db:
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute('DELETE FROM evidence')
|
|
assert store.intake_audit_pending()
|
|
|
|
|
|
def test_actual_posed_receiver_lost_reply_deduplicates_and_reconciles(setup, receiver):
|
|
service, store, _, _ = setup
|
|
assert request(service, envelope())[0] == 201
|
|
sink, transport, _ = receiver
|
|
store.stage_intake_audit()
|
|
first = store.claim_delivery(now=10)
|
|
transport.lose_reply = True
|
|
with pytest.raises(AuditDeliveryError):
|
|
sink.deliver(first[0], first[2])
|
|
replay = Store(store.path).claim_delivery(now=41)
|
|
assert replay[0] == first[0] and replay[2] == first[2]
|
|
reference = sink.deliver(replay[0], replay[2])
|
|
store.finish_delivery(replay[0], replay[1], reference=reference)
|
|
assert store.settle_intake_audit() == 1 and not store.intake_audit_pending()
|
|
report = OutboxWorker(store, sink).reconcile('2020-01-01T00:00:00Z', '2100-01-01T00:00:00Z')
|
|
assert report['count_values_match']
|
|
assert report['counts'][EventClass.MEMO_POSED.value] == {'source': 1, 'receiver': 1}
|
|
assert store.outbox()[0]['receiver_reference'] == 'audit:' + first[0]
|
|
|
|
|
|
def test_pump_includes_posed_evidence_and_retains_failed_notifications(setup, receiver):
|
|
service, store, _, _ = setup
|
|
assert request(service, envelope())[0] == 201
|
|
sink, _, _ = receiver
|
|
worker = IntakeDeliveryWorker(store, audit=sink, notify=lambda *args: False)
|
|
pump = AuditPump(store, sink, intake_worker=worker)
|
|
pump.tick()
|
|
assert pump.ready() and not store.intake_audit_pending()
|
|
with store._connection() as db:
|
|
assert db.execute("SELECT state FROM intake_deliveries WHERE kind='notification'").fetchone()[0] == 'pending'
|
|
snapshot = json.loads((store.path.parent / 'audit-reconciliation.json').read_text())
|
|
assert snapshot['count_values_match'] and EventClass.MEMO_POSED.value in snapshot['counts']
|
|
|
|
|
|
def test_failed_posed_delivery_blocks_readiness_and_no_empty_heartbeat(setup, receiver):
|
|
service, store, _, _ = setup
|
|
assert request(service, envelope())[0] == 201
|
|
sink, _, _ = receiver
|
|
sink.token_provider = lambda: 'invalid-synthetic-token'
|
|
pump = AuditPump(store, sink, intake_worker=IntakeDeliveryWorker(store, audit=sink))
|
|
pump.tick()
|
|
assert not pump.ready() and store.intake_audit_pending()
|
|
heartbeats = [json.loads(row['envelope']) for row in store.evidence()
|
|
if row['class'] == EventClass.HEARTBEAT.value]
|
|
assert all(row['data']['class'] != EventClass.MEMO_POSED.value for row in heartbeats)
|
|
|
|
|
|
@pytest.fixture
|
|
def hub_contract():
|
|
source = os.environ.get('INFD_STATE_HUB_SOURCE')
|
|
if not source:
|
|
pytest.skip('set INFD_STATE_HUB_SOURCE for actual hub contract checks')
|
|
assert (Path(source) / 'api/services/memo_forwarding.py').is_file()
|
|
sys.path.insert(0, source)
|
|
from api.services import informed_decision_contract, memo_forwarding
|
|
return informed_decision_contract, memo_forwarding
|
|
|
|
|
|
def test_actual_hub_forwarder_preserves_governed_boolean_fields(setup, hub_contract, tmp_path):
|
|
service, store, creates, _ = setup
|
|
contract, forwarding = hub_contract
|
|
body = envelope()
|
|
body['idempotency_key'] = contract.stable_idempotency_key(body['serves']['hub_decision_id'], body['memo']['id'])
|
|
calls = []
|
|
def transport(method, url, payload, headers):
|
|
assert method == 'POST' and url == 'https://intake.test/v1/memos'
|
|
assert headers['Authorization'] == 'Bearer projected-token'
|
|
calls.append(payload)
|
|
return request(service, payload)
|
|
fixture = forwarding.DevIntakeFixture(tmp_path / 'hub.sqlite', transport=transport,
|
|
intake_base_url='https://intake.test', hub_base_url='https://hub.test',
|
|
token_reader=lambda: 'projected-token')
|
|
# Owner commit d61ba32 accepts the shared schema's bool leaves. Do not strip
|
|
# or coerce requires_new_bind, required_ack, or sealed.
|
|
posed = fixture.submit(body)
|
|
assert posed.status == 'delivered' and len(calls) == 1 and len(creates) == 1
|
|
memo = calls[0]['memo']
|
|
assert memo['binding']['target']['requires_new_bind'] is False
|
|
assert memo['highlights'][0]['required_ack'] is True
|
|
assert memo['sealed'] is False
|
|
assert store.memos_for('keycape-human')
|
|
assert b'projected-token' not in Path(fixture.path).read_bytes()
|
|
|
|
|
|
def test_notification_body_matches_actual_hub_parser(hub_contract):
|
|
contract, _ = hub_contract
|
|
class Transport(HubTransport):
|
|
def request(self, method, url, *, headers, body):
|
|
parsed = contract.parse_notification(json.loads(body))
|
|
assert parsed.as_body() == json.loads(body)
|
|
assert contract.parse_delivery_id(headers['Idempotency-Key'])
|
|
return super().request(method, url, headers=headers, body=body)
|
|
hub = HubDelivery('https://hub.test', transport=Transport())
|
|
assert hub.notify(str(uuid.uuid4()), {'memo_id': 'review-1', 'question': 'Proceed?',
|
|
'memo_class': 'credential-lane', 'submitter': 'state-hub'})
|