atlas-iac/testing/tests/test_hermes_cli_lane_routing.py

82 lines
2.3 KiB
Python

"""Regression tests for durable Hermes worker-route model resolution."""
from __future__ import annotations
import importlib.util
import json
import sys
from pathlib import Path
SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
sys.path.insert(0, str(SCRIPTS))
SCRIPT = SCRIPTS / "cli_lane_runner.py"
SPEC = importlib.util.spec_from_file_location("cli_lane_runner_routing_test", SCRIPT)
assert SPEC and SPEC.loader
lanes = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = lanes
SPEC.loader.exec_module(lanes)
class SwitchyardResponse:
"""Return a tier header and independently resolved broker content."""
def __init__(self, selected: str, resolved_target: str):
self.headers = {
"x-model-router-selected-model": selected,
"x-model-router-rationale": "test route",
}
self.resolved_target = resolved_target
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self) -> bytes:
return json.dumps(
{
"model": self.headers["x-model-router-selected-model"],
"choices": [
{"message": {"content": self.resolved_target}}
],
}
).encode()
def test_switchyard_tier_is_resolved_before_invoking_codex():
"""Codex receives the exact account-visible ID, not a tier shorthand."""
route = lanes.select_route(
"Repair the failing implementation.",
"cli-auto",
open_request=lambda *_args, **_kwargs: SwitchyardResponse(
"worker/codex/terra/xhigh",
"worker/codex/gpt-5.6-terra/xhigh",
),
)
assert (route.provider, route.model, route.effort) == (
"codex",
"gpt-5.6-terra",
"xhigh",
)
def test_mismatched_broker_resolution_cannot_change_route_decision():
"""Resolved content cannot change Switchyard's provider or effort."""
route = lanes.select_route(
"Repair it.",
"cli-auto",
open_request=lambda *_args, **_kwargs: SwitchyardResponse(
"worker/codex/terra/xhigh",
"worker/claude/claude-opus-5/high",
),
)
assert (route.provider, route.model, route.effort) == (
"codex",
"terra",
"xhigh",
)