state-hub/tests/test_memo_dispatch.py

import json
from pathlib import Path
import stat
import subprocess
import sys

import pytest

from api.services.memo_dispatch import Dispatcher, Refused, token_file, transport, ORIGIN
from tests.test_memo_forwarding import envelope, receipt, Clock


def draft():
    return envelope(memo_class='decision-record')


def test_queue_is_explicit_and_receipt_survives_restart(tmp_path):
    calls = []
    def send(*args):
        calls.append(args)
        return 201, receipt()
    queue = tmp_path/'private'
    worker = Dispatcher(queue, send=send, read_token=lambda: 'synthetic-token')
    assert worker.enqueue(draft()).status == 'pending'
    assert calls == []
    assert worker.once().status == 'delivered'
    assert len(calls) == 1
    assert calls[0][0:2] == ('POST', ORIGIN+'/v1/memos')
    reopened = Dispatcher(queue, send=send, read_token=lambda: 'synthetic-token')
    assert reopened.enqueue(draft()).status == 'replay'
    assert reopened.once() is None
    assert reopened.counts() == {'delivered': 1}
    assert stat.S_IMODE((queue/'forward.sqlite').stat().st_mode) == 0o600
    assert b'synthetic-token' not in (queue/'forward.sqlite').read_bytes()


def test_rotation_retry_and_uncertain_native_create_blocks(tmp_path):
    clock = Clock()
    tokens = tmp_path/'token';tokens.write_text('first')
    seen=[]
    def send(method,url,body,headers):
        seen.append(headers['Authorization'])
        if len(seen)==1: raise OSError('hidden response and token')
        return 409, {'error':'approval_create_unresolved'}
    worker=Dispatcher(tmp_path/'private',send=send,read_token=lambda:token_file(tokens),clock=clock)
    worker.enqueue(draft())
    assert worker.once().status=='pending'
    tokens.write_text('second');clock.now+=10
    assert worker.once().status=='blocked'
    clock.now+=1000
    assert worker.once() is None
    assert seen==['Bearer first','Bearer second']
    assert 'hidden response' not in worker.path.read_bytes().decode(errors='ignore')


def test_lock_serializes_separate_processes(tmp_path):
    worker=Dispatcher(tmp_path/'private')
    with worker.lock():
        result=subprocess.run([sys.executable,'-m','api.services.memo_dispatch','--queue',str(worker.directory),'once'],capture_output=True,text=True)
    assert result.returncode==1
    assert json.loads(result.stdout)=={'error':'dispatch_refused'}


def test_unsafe_custody_and_wrong_routing_refused(tmp_path):
    unsafe=tmp_path/'unsafe';unsafe.mkdir(mode=0o755)
    with pytest.raises(Refused): Dispatcher(unsafe)
    private=tmp_path/'private';private.mkdir(mode=0o700)
    target=tmp_path/'target';target.write_text('secret')
    (private/'forward.sqlite').symlink_to(target)
    with pytest.raises(Refused): Dispatcher(private)
    worker=Dispatcher(tmp_path/'queue')
    with pytest.raises(Refused): worker.enqueue(envelope())
    assert worker.counts()=={}


def test_transport_refuses_arbitrary_destinations_before_io():
    with pytest.raises(Refused): transport('POST','https://other.example/v1/memos',{}, {'Authorization':'Bearer hidden'})


def test_cli_errors_never_emit_input_content(tmp_path):
    path=tmp_path/'bad';path.write_text('synthetic-secret')
    result=subprocess.run([sys.executable,'-m','api.services.memo_dispatch','--queue',str(tmp_path/'private'),'enqueue','--envelope',str(path)],capture_output=True,text=True)
    assert result.returncode==1 and 'synthetic-secret' not in result.stdout+result.stderr


def test_retained_lifecycle_handoff_has_no_invented_hub_relationship(tmp_path):
    from api.services.informed_decision_contract import ContractError, validate_envelope
    body=draft();body.pop('serves')
    body['memo']['id']='review-12345678-1234-1234-1234-123456789abc'
    body['idempotency_key']=body['memo']['id']
    assert validate_envelope(body)==body
    worker=Dispatcher(tmp_path/'private')
    assert worker.enqueue(body).status=='pending'
    assert 'serves' not in json.loads(worker.ledger.pose(body['idempotency_key'])['envelope_json'])
    body['idempotency_key']='changed-key'
    with pytest.raises(ContractError): validate_envelope(body)


def test_transport_suppresses_proxy_redirect_and_oversized_response(monkeypatch):
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    import threading
    from api.services import memo_dispatch
    calls=[]
    mode=['redirect']
    class Handler(BaseHTTPRequestHandler):
        def log_message(self,*args): pass
        def do_POST(self):
            calls.append(self.path)
            self.rfile.read(int(self.headers['Content-Length']))
            if mode[0]=='redirect':
                self.send_response(307);self.send_header('Location','/stolen');self.end_headers()
            else:
                self.send_response(200);self.end_headers();self.wfile.write(b'x'*(memo_dispatch.MAX_RESPONSE+1))
    server=ThreadingHTTPServer(('127.0.0.1',0),Handler)
    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
    origin=f'http://127.0.0.1:{server.server_port}'
    monkeypatch.setattr(memo_dispatch,'ORIGIN',origin)
    monkeypatch.setenv('http_proxy','http://127.0.0.1:1')
    try:
        assert transport('POST',origin+'/v1/memos',{}, {'Authorization':'Bearer synthetic'})[0]==307
        assert calls==['/v1/memos']
        mode[0]='oversized'
        with pytest.raises(OSError,match='transport_unavailable'):
            transport('POST',origin+'/v1/memos',{}, {})
    finally:
        server.shutdown();server.server_close();thread.join(timeout=2)


def test_kubernetes_queue_init_handles_setgid_without_taking_over_data(tmp_path):
    manifest = json.loads((Path(__file__).parents[1] / 'deploy/informed-decision/dispatcher.candidate.json').read_text())
    pod = next(item for item in manifest['items'] if item['kind'] == 'Deployment')['spec']['template']['spec']
    init = pod['initContainers'][0]
    assert init['image'] == pod['containers'][0]['image']
    tmp_path.chmod(0o2770)
    private = tmp_path / 'private'
    code = init['command'][2].replace('/data/private', str(private))
    def run():
        return subprocess.run([sys.executable, '-c', code], capture_output=True).returncode
    assert run() == 0
    assert stat.S_IMODE(private.stat().st_mode) == 0o700
    retained = private / 'retained'
    retained.write_text('existing evidence')
    assert run() == 0
    private.chmod(0o2700)
    assert run() != 0  # Existing nonempty custody is never repaired silently.
    retained.unlink()
    assert run() == 0  # Empty directory left by the initial failed startup.
    private.chmod(0o755)
    assert run() != 0