87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
"""Focused contracts for Hermes's local inference admission proxy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
HERMES = Path(__file__).parents[2] / "services/hermes"
|
|
|
|
|
|
def _model_gate_namespace() -> dict:
|
|
document = yaml.safe_load((HERMES / "model-gate-configmap.yaml").read_text())
|
|
namespace = {"__name__": "hermes_model_gate_test"}
|
|
exec(compile(document["data"]["model_gate.py"], "model_gate.py", "exec"), namespace)
|
|
return namespace
|
|
|
|
|
|
def test_model_gate_removes_hosted_only_reasoning_fields():
|
|
normalize = _model_gate_namespace()["_normalize_reasoning"]
|
|
|
|
top_level = json.loads(normalize(b'{"reasoning_effort":"xhigh"}'))
|
|
string_reasoning = json.loads(normalize(b'{"reasoning":"max"}'))
|
|
nested = json.loads(normalize(b'{"reasoning":{"effort":"xhigh"}}'))
|
|
|
|
assert "reasoning_effort" not in top_level
|
|
assert "reasoning" not in string_reasoning
|
|
assert "reasoning" not in nested
|
|
|
|
|
|
def test_model_gate_preserves_supported_and_non_json_requests():
|
|
normalize = _model_gate_namespace()["_normalize_reasoning"]
|
|
supported = b'{"model":"qwen2.5:14b-instruct-q4_0","messages":[]}'
|
|
non_json = b"streamed-body"
|
|
|
|
assert normalize(supported) == supported
|
|
assert normalize(non_json) == non_json
|
|
|
|
|
|
def test_model_gate_translates_switchyard_local_target_alias():
|
|
normalize = _model_gate_namespace()["_normalize_reasoning"]
|
|
|
|
routed = json.loads(
|
|
normalize(
|
|
b'{"model":"route/local/qwen2.5-14b/medium",'
|
|
b'"reasoning_effort":"medium","messages":[]}'
|
|
)
|
|
)
|
|
|
|
assert routed["model"] == "qwen2.5:14b-instruct-q4_0"
|
|
assert "reasoning_effort" not in routed
|
|
|
|
|
|
def test_model_gate_supplies_its_single_model_when_switchyard_omits_it():
|
|
normalize = _model_gate_namespace()["_normalize_reasoning"]
|
|
|
|
routed = json.loads(normalize(b'{"messages":[{"role":"user","content":"hi"}]}'))
|
|
|
|
assert routed["model"] == "qwen2.5:14b-instruct-q4_0"
|
|
|
|
|
|
def test_model_gate_runs_a_renderer_aware_ariadne_handoff():
|
|
"""Wolf cannot reach Ollama until the local image service reports idle."""
|
|
namespace = _model_gate_namespace()
|
|
assert "_wait_for_image_idle" in namespace
|
|
assert namespace["HANDOFF_PORT"] == 8081
|
|
handoff = namespace["HandoffHandler"]
|
|
assert callable(handoff.do_GET)
|
|
assert callable(handoff.do_POST)
|
|
|
|
|
|
def test_model_gate_can_only_read_the_local_image_deployment():
|
|
"""The dead-renderer escape hatch must not grant workload mutation access."""
|
|
documents = list(yaml.safe_load_all((HERMES / "model-gate-rbac.yaml").read_text()))
|
|
role = next(document for document in documents if document.get("kind") == "Role")
|
|
|
|
assert role["rules"] == [
|
|
{
|
|
"apiGroups": ["apps"],
|
|
"resources": ["deployments"],
|
|
"resourceNames": ["hermes-local-image"],
|
|
"verbs": ["get"],
|
|
}
|
|
]
|