79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
|
import importlib.util
|
||
|
|
from io import BytesIO
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
spec = importlib.util.spec_from_file_location(
|
||
|
|
"sitting_bind_preflight",
|
||
|
|
Path(__file__).resolve().parents[1] / "tools" / "sitting_bind_preflight.py",
|
||
|
|
)
|
||
|
|
preflight = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(preflight)
|
||
|
|
evaluate, load_sitting, probe_origin = preflight.evaluate, preflight.load_sitting, preflight.probe_origin
|
||
|
|
|
||
|
|
|
||
|
|
class _Resp:
|
||
|
|
def __init__(self, status, body=b"", headers=None):
|
||
|
|
self.status = status
|
||
|
|
self.headers = headers or {}
|
||
|
|
self._body = BytesIO(body)
|
||
|
|
|
||
|
|
def getcode(self):
|
||
|
|
return self.status
|
||
|
|
|
||
|
|
def read(self, n=-1):
|
||
|
|
return self._body.read(n)
|
||
|
|
|
||
|
|
def __enter__(self):
|
||
|
|
return self
|
||
|
|
|
||
|
|
def __exit__(self, *exc):
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def test_unsigned_sitting_is_not_ready_to_sit():
|
||
|
|
report = evaluate(load_sitting())
|
||
|
|
assert report["ready_to_sit"] is False
|
||
|
|
assert report["agent_disposition"] == "forbidden"
|
||
|
|
assert report["memo_count"] == 8
|
||
|
|
assert "missing_act_binding" in report["gates"]
|
||
|
|
assert "principal_not_live_subject" in report["gates"]
|
||
|
|
assert len(report["missing_act_binding"]) == 8
|
||
|
|
assert report["live_accept"] is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_origin_readyz_503_is_live_accept_closed():
|
||
|
|
bodies = {
|
||
|
|
"/healthz": _Resp(200, b'{"status":"ok"}'),
|
||
|
|
"/readyz": _Resp(503, b'{"status":"incomplete","reason":"approval_path_not_connected"}'),
|
||
|
|
"/auth/start": _Resp(303, b"", {"Location": "https://kc.coulomb.social/authorize?client_id=informed-decision-approver"}),
|
||
|
|
}
|
||
|
|
|
||
|
|
def opener(req, timeout=10):
|
||
|
|
return bodies[req.full_url.split("https://decisions.coulomb.social", 1)[1]]
|
||
|
|
|
||
|
|
probe = probe_origin("https://decisions.coulomb.social", opener=opener)
|
||
|
|
report = evaluate(load_sitting(), probe)
|
||
|
|
assert report["live_accept"] == "closed"
|
||
|
|
assert "live_accept_closed" in report["gates"]
|
||
|
|
assert report["ready_to_sit"] is False
|
||
|
|
assert probe["auth_start"]["http"] == 303
|
||
|
|
assert probe["auth_start"]["location_host"] == "kc.coulomb.social"
|
||
|
|
assert "body" not in probe["auth_start"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_origin_ready_still_blocked_on_unsigned_memos():
|
||
|
|
bodies = {
|
||
|
|
"/healthz": _Resp(200, b'{"status":"ok"}'),
|
||
|
|
"/readyz": _Resp(200, b'{"status":"ready"}'),
|
||
|
|
"/auth/start": _Resp(303, b"", {"Location": "https://kc.example/authorize"}),
|
||
|
|
}
|
||
|
|
|
||
|
|
def opener(req, timeout=10):
|
||
|
|
return bodies[req.full_url.split("https://origin.example", 1)[1]]
|
||
|
|
|
||
|
|
report = evaluate(load_sitting(), probe_origin("https://origin.example", opener=opener))
|
||
|
|
assert report["live_accept"] == "open"
|
||
|
|
assert "live_accept_closed" not in report["gates"]
|
||
|
|
assert report["ready_to_sit"] is False
|
||
|
|
assert "missing_act_binding" in report["gates"]
|