130 lines
3.6 KiB
Python
130 lines
3.6 KiB
Python
"""Focused tests for Hermes-to-Herdr routing and the shared auth patch."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
|
|
|
|
|
|
def _load(name: str):
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
dispatch = _load("herdr_dispatch")
|
|
auth_patch = _load("patch_hermes_auth")
|
|
|
|
|
|
def test_herdr_plan_chooses_task_shape_and_caps_effort():
|
|
status = {
|
|
"routes": {
|
|
"codex-high": [
|
|
"openai-codex/gpt-5.6-sol",
|
|
"anthropic/claude-opus-5",
|
|
"custom/qwen2.5:14b-instruct-q4_0",
|
|
],
|
|
"claude-medium": [
|
|
"anthropic/claude-sonnet-5",
|
|
"openai-codex/gpt-5.6-terra",
|
|
],
|
|
}
|
|
}
|
|
implementation = dispatch.select_plan(status, "implementation", "high")
|
|
architecture = dispatch.select_plan(status, "architecture", "medium")
|
|
assert implementation["worker"] == "codex"
|
|
assert implementation["model"] == "gpt-5.6-sol"
|
|
assert architecture["worker"] == "claude"
|
|
assert architecture["model"] == "claude-sonnet-5"
|
|
with pytest.raises(ValueError, match="effort"):
|
|
dispatch.select_plan(status, "review", "max")
|
|
|
|
|
|
def test_claude_worker_waits_for_prompt_readiness(tmp_path: Path, monkeypatch):
|
|
herdr = tmp_path / "herdr"
|
|
herdr.touch()
|
|
project = tmp_path / "project"
|
|
project.mkdir()
|
|
calls = []
|
|
|
|
def fake_run(command, env):
|
|
calls.append(command)
|
|
if command[1:3] == ["workspace", "create"]:
|
|
return {"result": {"root_pane": {"pane_id": "w2:p1"}}}
|
|
return {"result": {"ok": True}}
|
|
|
|
monkeypatch.setattr(dispatch, "HERDR_BIN", herdr)
|
|
monkeypatch.setattr(dispatch, "_run", fake_run)
|
|
plan = {
|
|
"worker": "claude",
|
|
"model": "claude-haiku-4-5-20251001",
|
|
"effort": "low",
|
|
}
|
|
|
|
dispatch.launch_worker(plan, project, "review", "Check the implementation.")
|
|
|
|
ready = calls[-2]
|
|
assert ready[1:] == [
|
|
"pane",
|
|
"wait-output",
|
|
"w2:p1",
|
|
"--match",
|
|
"accept edits on",
|
|
"--source",
|
|
"recent",
|
|
"--lines",
|
|
"120",
|
|
"--timeout",
|
|
"120000",
|
|
]
|
|
prompt = calls[-1]
|
|
assert prompt[1:5] == [
|
|
"agent",
|
|
"prompt",
|
|
"claude-review",
|
|
"Check the implementation.",
|
|
]
|
|
assert prompt[-9:] == [
|
|
"--wait",
|
|
"--until",
|
|
"working",
|
|
"--until",
|
|
"done",
|
|
"--until",
|
|
"blocked",
|
|
"--timeout",
|
|
"15000",
|
|
]
|
|
|
|
|
|
def test_auth_patch_honors_explicit_shared_store(tmp_path: Path):
|
|
source = tmp_path / "auth.py"
|
|
destination = tmp_path / "patched" / "auth.py"
|
|
source.write_text(
|
|
"from pathlib import Path\nimport os\n\n"
|
|
"def _auth_file_path() -> Path:\n"
|
|
" path = get_hermes_home() / \"auth.json\"\n"
|
|
" return path\n",
|
|
encoding="utf-8",
|
|
)
|
|
auth_patch.patch(source, destination)
|
|
content = destination.read_text(encoding="utf-8")
|
|
assert 'os.environ.get("HERMES_AUTH_FILE"' in content
|
|
assert "Path(configured)" in content
|
|
|
|
|
|
def test_auth_patch_fails_closed_on_upstream_drift(tmp_path: Path):
|
|
source = tmp_path / "auth.py"
|
|
source.write_text("def changed():\n pass\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="context changed"):
|
|
auth_patch.patch(source, tmp_path / "patched.py")
|