448 lines
16 KiB
Python
448 lines
16 KiB
Python
"""Isolated mediator, SCM gate, and model-facing API contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
SCRIPTS = ROOT / "services/hermes/scripts"
|
|
SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts"
|
|
sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)]
|
|
|
|
import execution_pool_client as client # noqa: E402
|
|
import execution_pool_protocol as protocol # noqa: E402
|
|
import execution_pool_scm as scm # noqa: E402
|
|
|
|
|
|
KEY = b"d" * 64
|
|
RESULT = {
|
|
"status": "completed",
|
|
"summary": "Completed safely.",
|
|
"changed_files": ["safe.py"],
|
|
"tests_run": ["pytest"],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
}
|
|
|
|
|
|
def binding(**changes):
|
|
value = {
|
|
"board": "metis", "task_id": "t_deadbeef", "run_id": "42",
|
|
"worker_ordinal": 0, "attempt": 1,
|
|
}
|
|
value.update(changes)
|
|
return value
|
|
|
|
|
|
def payload(**changes):
|
|
value = {
|
|
"context": "safe objective",
|
|
"repo_url": "https://scm.bstein.dev/titan/metis.git",
|
|
"branch": "wt/t_deadbeef",
|
|
"base_branch": "main",
|
|
}
|
|
value.update(changes)
|
|
return value
|
|
|
|
|
|
def assignment(**payload_changes):
|
|
return protocol.sign_envelope(
|
|
KEY, "assignment", binding(), payload(**payload_changes)
|
|
)
|
|
|
|
|
|
class Response:
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self, _size=-1):
|
|
return self.body
|
|
|
|
|
|
class FakeSCM:
|
|
def __init__(self):
|
|
self.submitted = []
|
|
|
|
def checkout(self, _envelope):
|
|
return {"workspace": "/workspace/run", "baseline_sha": "a" * 40}
|
|
|
|
def submit(self, envelope, request):
|
|
self.submitted.append((envelope, request))
|
|
return {"pull_request": "https://scm.bstein.dev/titan/metis/pulls/7"}
|
|
|
|
|
|
def test_result_schema_validation_rejects_every_unsafe_shape():
|
|
invalid = [
|
|
None,
|
|
{},
|
|
{"structured": []},
|
|
{"structured": {**RESULT, "extra": True}},
|
|
{"structured": {**RESULT, "status": "unknown"}},
|
|
{"structured": {**RESULT, "summary": ""}},
|
|
{"structured": {**RESULT, "tests_run": "pytest"}},
|
|
{"structured": {**RESULT, "tests_run": [7]}},
|
|
]
|
|
for value in invalid:
|
|
with pytest.raises(protocol.ProtocolError):
|
|
client._validate_result(value)
|
|
assert client._validate_result({"structured": dict(RESULT)})["structured"] == RESULT
|
|
|
|
|
|
def test_client_post_verifies_response_and_bounds_body(monkeypatch):
|
|
boundary = client.ClientBoundary(KEY, FakeSCM())
|
|
ack = protocol.sign_envelope(KEY, "ack", binding(), {"accepted": True})
|
|
monkeypatch.setattr(client.urllib.request, "urlopen", lambda *_a, **_k: Response(protocol.canonical_json(ack)))
|
|
assert boundary._post("/v1/result", ack)["kind"] == "ack"
|
|
monkeypatch.setattr(
|
|
client.urllib.request,
|
|
"urlopen",
|
|
lambda *_a, **_k: Response(b"x" * (protocol.MAX_WIRE_BYTES + 1)),
|
|
)
|
|
with pytest.raises(protocol.ProtocolError, match="exceeds"):
|
|
boundary._post("/v1/result", ack)
|
|
|
|
|
|
def test_poll_materializes_only_ordinal_owned_assignment(monkeypatch):
|
|
monkeypatch.setattr(client, "ORDINAL", 0)
|
|
boundary = client.ClientBoundary(KEY, FakeSCM())
|
|
boundary._post = lambda *_a: protocol.sign_envelope(
|
|
KEY,
|
|
"ack",
|
|
binding(board="", task_id="", run_id="", attempt=0),
|
|
{"assignment": None},
|
|
)
|
|
assert boundary.poll() == {"assignment": None}
|
|
|
|
boundary._post = lambda *_a: assignment()
|
|
result = boundary.poll()["assignment"]
|
|
assert result["workspace"] == "/workspace/run"
|
|
assert result["protocol_version"] == 2
|
|
assert "signature" not in result
|
|
|
|
boundary._post = lambda *_a: protocol.sign_envelope(
|
|
KEY, "assignment", binding(worker_ordinal=1), payload()
|
|
)
|
|
with pytest.raises(protocol.ProtocolError, match="foreign"):
|
|
boundary.poll()
|
|
|
|
|
|
def test_heartbeat_and_finish_require_exact_current_binding(monkeypatch):
|
|
fake_scm = FakeSCM()
|
|
boundary = client.ClientBoundary(KEY, fake_scm)
|
|
current = assignment()
|
|
boundary.current = current
|
|
|
|
def post(path, envelope):
|
|
kind = "heartbeat" if path.endswith("heartbeat") else "result"
|
|
protocol.verify_envelope(KEY, envelope, expected_kind=kind)
|
|
return protocol.sign_envelope(
|
|
KEY, "ack", binding(), {"accepted": True, "duplicate": False}
|
|
)
|
|
|
|
boundary._post = post
|
|
assert boundary.heartbeat(
|
|
{"binding": binding(), "payload": {"note": "active"}}
|
|
)["ack"]["accepted"]
|
|
request = {
|
|
"binding": binding(),
|
|
"payload": {"structured": dict(RESULT), "returncode": 0},
|
|
"title": "Safe change",
|
|
"body": "Evidence",
|
|
}
|
|
finished = boundary.finish(request)
|
|
assert finished["ack"]["accepted"] and boundary.current is None
|
|
assert fake_scm.submitted
|
|
assert finished["structured"]["artifacts"] == [
|
|
"https://scm.bstein.dev/titan/metis/pulls/7"
|
|
]
|
|
|
|
boundary.current = current
|
|
with pytest.raises(protocol.ProtocolError, match="heartbeat payload"):
|
|
boundary.heartbeat({"binding": binding(), "payload": "bad"})
|
|
with pytest.raises(protocol.ProtocolError, match="object"):
|
|
boundary._current_for(None)
|
|
with pytest.raises(protocol.ProtocolError, match="does not own"):
|
|
boundary._current_for(binding(attempt=2))
|
|
|
|
|
|
def test_noncompleted_finish_never_invokes_scm_and_bad_ack_is_rejected():
|
|
fake_scm = FakeSCM()
|
|
boundary = client.ClientBoundary(KEY, fake_scm)
|
|
boundary.current = assignment()
|
|
bad = protocol.sign_envelope(KEY, "ack", binding(attempt=2), {"accepted": True})
|
|
boundary._post = lambda *_a: bad
|
|
blocked = {**RESULT, "status": "blocked", "blockers": ["capacity"]}
|
|
with pytest.raises(protocol.ProtocolError, match="binding changed"):
|
|
boundary.finish(
|
|
{"binding": binding(), "payload": {"structured": blocked, "returncode": 1}}
|
|
)
|
|
assert fake_scm.submitted == []
|
|
|
|
boundary._post = lambda *_a: protocol.sign_envelope(
|
|
KEY, "result", binding(), {"accepted": True}
|
|
)
|
|
with pytest.raises(protocol.ProtocolError, match="binding changed"):
|
|
boundary.heartbeat({"binding": binding(), "payload": {}})
|
|
|
|
|
|
def http_request(handler, path, *, body=None):
|
|
server = protocol.BoundedHTTPServer(("127.0.0.1", 0), handler, max_workers=2)
|
|
thread = threading.Thread(target=server.handle_request)
|
|
thread.start()
|
|
request = urllib.request.Request(
|
|
f"http://127.0.0.1:{server.server_port}{path}",
|
|
data=body,
|
|
method="POST" if body is not None else "GET",
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=3) as response:
|
|
return response.status, json.loads(response.read())
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, json.loads(error.read())
|
|
finally:
|
|
thread.join(timeout=3)
|
|
server.server_close()
|
|
|
|
|
|
def test_model_api_exposes_only_gated_state_machine_operations():
|
|
class Boundary:
|
|
poll = staticmethod(lambda: {"assignment": None})
|
|
heartbeat = staticmethod(lambda request: {"heartbeat": request["payload"]})
|
|
finish = staticmethod(lambda request: {"finish": request["payload"]})
|
|
|
|
handler = client.handler_factory(Boundary())
|
|
assert http_request(handler, "/ready") == (
|
|
200, {"protocol_version": 2, "ready": True}
|
|
)
|
|
assert http_request(handler, "/missing")[0] == 404
|
|
for operation in ("poll", "heartbeat", "finish"):
|
|
body = protocol.canonical_json(
|
|
{"operation": operation, "binding": binding(), "payload": {}}
|
|
)
|
|
assert http_request(handler, "/v1/client", body=body)[0] == 200
|
|
for value in (
|
|
{"operation": "bypass"},
|
|
{"operation": "poll", "authority": "steal"},
|
|
):
|
|
status, response = http_request(
|
|
handler, "/v1/client", body=protocol.canonical_json(value)
|
|
)
|
|
assert status == 409 and response.get("error")
|
|
|
|
|
|
def test_client_main_validates_ordinal_and_starts_bounded_server(monkeypatch):
|
|
monkeypatch.setattr(client, "ORDINAL", -1)
|
|
with pytest.raises(SystemExit, match="ORDINAL"):
|
|
client.main()
|
|
started = []
|
|
|
|
class Server:
|
|
def __init__(self, address, _handler, max_workers):
|
|
started.append((address, max_workers))
|
|
|
|
def serve_forever(self):
|
|
return
|
|
|
|
monkeypatch.setattr(client, "ORDINAL", 2)
|
|
monkeypatch.setattr(client, "read_key", lambda _path: KEY)
|
|
monkeypatch.setattr(client, "BoundedHTTPServer", Server)
|
|
monkeypatch.setattr(client, "SCMBoundary", lambda _key: FakeSCM())
|
|
assert client.main() == 0
|
|
assert started == [(("0.0.0.0", client.PORT), 4)]
|
|
|
|
|
|
def init_checkout(path, branch="wt/t_deadbeef"):
|
|
subprocess.run(["git", "init", "-q", str(path)], check=True)
|
|
commands = [
|
|
("remote", "add", "origin", "https://scm.bstein.dev/titan/metis.git"),
|
|
(
|
|
"remote",
|
|
"add",
|
|
"hermes-broker",
|
|
"http://hermes-scm-broker.hermes-scm.svc.cluster.local:9081/"
|
|
"git/atlas/metis.git",
|
|
),
|
|
("checkout", "-qb", branch),
|
|
("config", "user.email", "test@example.com"),
|
|
("config", "user.name", "Test"),
|
|
]
|
|
for command in commands:
|
|
subprocess.run(["git", "-C", str(path), *command], check=True)
|
|
(path / "tracked").write_text("safe\n")
|
|
subprocess.run(["git", "-C", str(path), "add", "tracked"], check=True)
|
|
subprocess.run(["git", "-C", str(path), "commit", "-qm", "initial"], check=True)
|
|
return subprocess.check_output(
|
|
["git", "-C", str(path), "rev-parse", "HEAD"], text=True
|
|
).strip()
|
|
|
|
|
|
def test_scm_run_and_binding_validation(tmp_path, monkeypatch):
|
|
checkout = tmp_path / "checkout"
|
|
head = init_checkout(checkout)
|
|
assert scm._run("rev-parse", "HEAD", cwd=checkout) == head
|
|
with pytest.raises(RuntimeError):
|
|
scm._run("rev-parse", "missing", cwd=checkout)
|
|
monkeypatch.setattr(scm, "MAX_STATUS_BYTES", 1)
|
|
with pytest.raises(protocol.ProtocolError, match="output"):
|
|
scm._run("rev-parse", "HEAD", cwd=checkout)
|
|
|
|
monkeypatch.setattr(scm, "ORDINAL", 0)
|
|
assert scm._binding(assignment())[1:] == ("metis", "wt/t_deadbeef", "main")
|
|
for envelope in (
|
|
{**assignment(), "kind": "result"},
|
|
{**assignment(), "payload": []},
|
|
assignment(repo_url="https://evil.example/metis.git"),
|
|
assignment(branch="main"),
|
|
assignment(base_branch="../main"),
|
|
protocol.sign_envelope(KEY, "assignment", binding(worker_ordinal=1), payload()),
|
|
):
|
|
with pytest.raises(protocol.ProtocolError):
|
|
scm._binding(envelope)
|
|
|
|
|
|
def test_scm_private_paths_and_text_are_bounded(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
|
|
target = scm.workspace_path(assignment())
|
|
assert target.name == "42"
|
|
assert scm._state_path(assignment()).name == "metis-t_deadbeef-42.json"
|
|
bad = {**assignment(), "task_id": "../bad"}
|
|
with pytest.raises(protocol.ProtocolError, match="binding"):
|
|
scm.workspace_path(bad)
|
|
state = tmp_path / "regular"
|
|
state.write_text("safe")
|
|
assert scm._regular_text(state, 10) == "safe"
|
|
with pytest.raises(protocol.ProtocolError, match="invalid"):
|
|
scm._regular_text(state, 1)
|
|
binary = tmp_path / "binary"
|
|
binary.write_bytes(b"\xff")
|
|
with pytest.raises(protocol.ProtocolError, match="malformed"):
|
|
scm._regular_text(binary, 10)
|
|
linked_state = tmp_path / "linked-state"
|
|
linked_state.mkdir()
|
|
(tmp_path / "state-link").symlink_to(linked_state, target_is_directory=True)
|
|
monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state-link")
|
|
with pytest.raises(protocol.ProtocolError):
|
|
scm._state_path(assignment())
|
|
|
|
|
|
def test_boundary_verification_existing_checkout_and_missing_baseline(tmp_path, monkeypatch):
|
|
workspace = tmp_path / "workspace"
|
|
state_root = tmp_path / "state"
|
|
monkeypatch.setattr(scm, "WORKSPACE_ROOT", workspace)
|
|
monkeypatch.setattr(scm, "SCM_ROOT", state_root)
|
|
monkeypatch.setattr(scm, "ORDINAL", 0)
|
|
boundary = scm.Boundary(KEY)
|
|
assert boundary.verify(assignment())["kind"] == "assignment"
|
|
destination = scm.workspace_path(assignment())
|
|
baseline = init_checkout(destination)
|
|
protocol.atomic_json(
|
|
scm._state_path(assignment()),
|
|
{"baseline_sha": baseline, "repo": "metis", "branch": "wt/t_deadbeef"},
|
|
)
|
|
assert boundary.checkout(assignment())["baseline_sha"] == baseline
|
|
scm._state_path(assignment()).write_text("{}")
|
|
with pytest.raises(protocol.ProtocolError, match="baseline"):
|
|
boundary.checkout(assignment())
|
|
|
|
|
|
@pytest.mark.parametrize("feature_exists", [True, False])
|
|
def test_new_checkout_uses_broker_and_safe_base_fallback(
|
|
tmp_path, monkeypatch, feature_exists
|
|
):
|
|
monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
|
|
monkeypatch.setattr(scm, "ORDINAL", 0)
|
|
calls = []
|
|
|
|
def run(*arguments, cwd=None, timeout=300):
|
|
calls.append((arguments, cwd, timeout))
|
|
if arguments[0] == "clone":
|
|
branch = arguments[arguments.index("--branch") + 1]
|
|
destination = Path(arguments[-1])
|
|
if branch == "wt/t_deadbeef" and not feature_exists:
|
|
raise RuntimeError("missing branch")
|
|
(destination / ".git").mkdir(parents=True)
|
|
return ""
|
|
|
|
monkeypatch.setattr(scm, "_run", run)
|
|
monkeypatch.setattr(scm, "_workspace_identity", lambda *_a: "a" * 40)
|
|
result = scm.Boundary(KEY).checkout(assignment())
|
|
assert result["baseline_sha"] == "a" * 40
|
|
assert any("hermes-scm-broker" in str(call) for call in calls)
|
|
if not feature_exists:
|
|
assert any(call[0][0] == "checkout" for call in calls)
|
|
|
|
|
|
def test_failed_clone_never_deletes_unmanaged_state(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
|
|
monkeypatch.setattr(scm, "ORDINAL", 0)
|
|
|
|
def fail(*arguments, **_kwargs):
|
|
destination = Path(arguments[-1])
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
(destination / "preserved").write_text("owner data")
|
|
raise RuntimeError("clone failed")
|
|
|
|
monkeypatch.setattr(scm, "_run", fail)
|
|
with pytest.raises(protocol.ProtocolError, match="unmanaged"):
|
|
scm.Boundary(KEY).checkout(assignment())
|
|
assert (scm.workspace_path(assignment()) / "preserved").read_text() == "owner data"
|
|
|
|
|
|
def test_draft_reuse_create_and_submit_gates(tmp_path, monkeypatch):
|
|
existing = json.dumps([{"html_url": "https://scm/pulls/1"}]).encode()
|
|
monkeypatch.setattr(scm.scm_broker_client, "read", lambda _path: existing)
|
|
assert scm.Boundary._draft("metis", "wt/task", "main", "a" * 40, "t", "b") == "https://scm/pulls/1"
|
|
monkeypatch.setattr(scm.scm_broker_client, "read", lambda _path: b"[]")
|
|
monkeypatch.setattr(
|
|
scm.scm_broker_client, "create_draft",
|
|
lambda *_a, **_k: json.dumps({"html_url": "https://scm/pulls/2"}).encode(),
|
|
)
|
|
assert scm.Boundary._draft("metis", "wt/task", "main", "a" * 40, "t", "b") == "https://scm/pulls/2"
|
|
|
|
monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
|
|
monkeypatch.setattr(scm, "ORDINAL", 0)
|
|
destination = scm.workspace_path(assignment())
|
|
destination.mkdir(parents=True)
|
|
state = scm._state_path(assignment())
|
|
protocol.atomic_json(state, {"baseline_sha": "a" * 40})
|
|
boundary = scm.Boundary(KEY)
|
|
monkeypatch.setattr(scm, "_workspace_identity", lambda *_a: "b" * 40)
|
|
outputs = {"status": "", "rev-list": "1", "push": ""}
|
|
|
|
def run(*arguments, **_kwargs):
|
|
return outputs.get(arguments[0], "")
|
|
|
|
monkeypatch.setattr(scm, "_run", run)
|
|
monkeypatch.setattr(boundary, "_draft", lambda *_a: "https://scm/pulls/3")
|
|
result = boundary.submit(assignment(), {"title": "safe", "body": "evidence"})
|
|
assert result["pull_request"] == "https://scm/pulls/3"
|
|
outputs["status"] = "?? untracked"
|
|
with pytest.raises(protocol.ProtocolError, match="uncommitted"):
|
|
boundary.submit(assignment(), {})
|
|
outputs["status"] = ""
|
|
outputs["rev-list"] = "0"
|
|
assert boundary.submit(assignment(), {})["pull_request"] == ""
|
|
with pytest.raises(protocol.ProtocolError, match="metadata"):
|
|
boundary.submit(assignment(), {"title": "x" * 513, "body": "x"})
|