285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""Hermes chat images contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from test_hermes_chat_support import (
|
|
HERMES,
|
|
Path,
|
|
SimpleNamespace,
|
|
_documents,
|
|
importlib,
|
|
sys,
|
|
time,
|
|
yaml,
|
|
)
|
|
|
|
|
|
def test_compact_image_edit_resolves_latest_tenant_artifact(tmp_path, monkeypatch):
|
|
"""Follow-up edits resolve the source server-side and keep tool JSON small."""
|
|
provider_module = SimpleNamespace(
|
|
DEFAULT_ASPECT_RATIO="square",
|
|
ImageGenProvider=object,
|
|
error_response=lambda **value: value,
|
|
normalize_reference_images=lambda value: value,
|
|
resolve_aspect_ratio=lambda value: value,
|
|
save_b64_image=lambda *_args, **_kwargs: tmp_path / "saved.png",
|
|
success_response=lambda **value: value,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "agent", SimpleNamespace())
|
|
monkeypatch.setitem(sys.modules, "agent.image_gen_provider", provider_module)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"hermes_image_plugin",
|
|
HERMES / "plugins" / "image-gen-broker" / "__init__.py",
|
|
)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
image_dir = tmp_path / "cache" / "images"
|
|
image_dir.mkdir(parents=True)
|
|
older = image_dir / "atlas_flux-old.png"
|
|
newest = image_dir / "atlas_gpt-image-new.png"
|
|
older.write_bytes(b"older")
|
|
newest.write_bytes(b"newest")
|
|
older.touch()
|
|
time.sleep(0.001)
|
|
newest.touch()
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
module,
|
|
"_handle_image_generate",
|
|
lambda args, route: calls.append((args, route)) or "ok",
|
|
)
|
|
assert module._handle_hosted_edit({"prompt": "make it a clown"}) == "ok"
|
|
assert calls == [
|
|
(
|
|
{
|
|
"prompt": "make it a clown",
|
|
"image_url": str(newest.resolve()),
|
|
},
|
|
"hosted",
|
|
)
|
|
]
|
|
|
|
def test_chat_reasoning_uses_switchyard_without_owner_credentials():
|
|
"""Family pods use AUTO/manual routes without mounting owner credentials."""
|
|
configmap = _documents(HERMES / "chat-configmap.yaml")[0]
|
|
config = yaml.safe_load(configmap["data"]["config.yaml"])
|
|
assert config["model"] == {
|
|
"provider": "atlas-switchyard",
|
|
"default": "atlas/auto/fast",
|
|
"model": "atlas/auto/fast",
|
|
}
|
|
assert config["providers"]["atlas-switchyard"] == {
|
|
"name": "Automatic Router",
|
|
"api": "http://hermes-switchyard.hermes.svc.cluster.local:9005/v1",
|
|
"api_key": "atlas-switchyard",
|
|
"default_model": "atlas/auto/fast",
|
|
"transport": "chat_completions",
|
|
}
|
|
assert config["platforms"]["api_server"]["extra"]["model_routes"] == {
|
|
route: {"provider": "atlas-switchyard", "model": route}
|
|
for route in [
|
|
"atlas/auto/fast",
|
|
"atlas/auto/balanced",
|
|
"atlas/auto/deep",
|
|
"atlas/auto/maximum",
|
|
"atlas/manual/codex/luna",
|
|
"atlas/manual/codex/terra",
|
|
"atlas/manual/codex/sol",
|
|
"atlas/manual/claude/haiku",
|
|
"atlas/manual/claude/fable",
|
|
"atlas/manual/claude/sonnet",
|
|
"atlas/manual/claude/opus",
|
|
"atlas/manual/local/qwen-14b",
|
|
]
|
|
}
|
|
|
|
agent = _documents(HERMES / "agent-deployment.yaml")[0]
|
|
containers = agent["spec"]["template"]["spec"]["containers"]
|
|
broker = next(item for item in containers if item["name"] == "codex-broker")
|
|
assert broker["ports"] == [
|
|
{"name": "codex-broker", "containerPort": 9003, "protocol": "TCP"}
|
|
]
|
|
assert broker["securityContext"]["readOnlyRootFilesystem"] is True
|
|
assert broker["securityContext"]["runAsNonRoot"] is True
|
|
assert {item["name"]: item["value"] for item in broker["env"]}.items() >= {
|
|
"PYTHONPATH": "/opt/hermes",
|
|
"HERMES_CODEX_BROKER_LISTEN_PORT": "9003",
|
|
"HERMES_ROUTING_CATALOG_PATH": "/routing-catalog/catalog.json",
|
|
}.items()
|
|
|
|
services = _documents(HERMES / "service.yaml")
|
|
service = next(
|
|
item for item in services if item["metadata"]["name"] == "hermes-codex-broker"
|
|
)
|
|
assert service["spec"]["selector"] == {"app": "hermes-agent"}
|
|
assert service["spec"]["ports"] == [
|
|
{
|
|
"name": "http",
|
|
"port": 9003,
|
|
"targetPort": "codex-broker",
|
|
"protocol": "TCP",
|
|
}
|
|
]
|
|
|
|
statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
|
|
assert statefulset["spec"]["template"]["metadata"]["annotations"][
|
|
"ai.bstein.dev/config-rev"
|
|
] == "20260816-telegram-topics"
|
|
pod_spec = statefulset["spec"]["template"]["spec"]
|
|
patch_init = next(
|
|
item for item in pod_spec["initContainers"]
|
|
if item["name"] == "patch-stream-recovery"
|
|
)
|
|
assert patch_init["command"][-2:] == [
|
|
"/opt/hermes/agent/conversation_loop.py",
|
|
"/patched/conversation_loop.py",
|
|
]
|
|
hermes = next(
|
|
item
|
|
for item in pod_spec["containers"]
|
|
if item["name"] == "hermes"
|
|
)
|
|
assert {
|
|
"name": "stream-recovery-patch",
|
|
"mountPath": "/opt/hermes/agent/conversation_loop.py",
|
|
"subPath": "conversation_loop.py",
|
|
} in hermes["volumeMounts"]
|
|
api_session_init = next(
|
|
item for item in pod_spec["initContainers"]
|
|
if item["name"] == "patch-api-server-sessions"
|
|
)
|
|
assert "patch_api_server_sessions.py" in api_session_init["args"][0]
|
|
assert "migrate_telegram_api_sessions.py" in api_session_init["args"][0]
|
|
assert {
|
|
"name": "api-server-patch",
|
|
"mountPath": "/opt/hermes/gateway/platforms/api_server.py",
|
|
"subPath": "api_server.py",
|
|
} in hermes["volumeMounts"]
|
|
assert any(
|
|
volume["name"] == "api-server-patch" for volume in pod_spec["volumes"]
|
|
)
|
|
assert not any(
|
|
mount["mountPath"].endswith("/.codex")
|
|
for mount in hermes["volumeMounts"]
|
|
)
|
|
|
|
policies = _documents(HERMES / "networkpolicy.yaml")
|
|
agent_policy = next(
|
|
item for item in policies if item["metadata"]["name"] == "hermes-agent-isolation"
|
|
)
|
|
broker_ingress = next(
|
|
rule
|
|
for rule in agent_policy["spec"]["ingress"]
|
|
if {port["port"] for port in rule["ports"]} == {9002, 9003}
|
|
)
|
|
assert broker_ingress["from"][0]["podSelector"]["matchLabels"] == {
|
|
"app": "hermes-chat-tenant"
|
|
}
|
|
|
|
def test_image_broker_returns_bytes_and_removes_owner_cache(tmp_path: Path, monkeypatch):
|
|
"""The broker must not retain a family user's generated image."""
|
|
broker_path = HERMES / "scripts" / "image_broker.py"
|
|
spec = importlib.util.spec_from_file_location("hermes_image_broker", broker_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
generated = tmp_path / "generated.png"
|
|
generated.write_bytes(b"\x89PNG\r\n\x1a\nprivate-image")
|
|
|
|
class Provider:
|
|
def generate(self, prompt, aspect, **kwargs):
|
|
assert prompt == "paint a blue sphere"
|
|
assert aspect == "square"
|
|
return {
|
|
"success": True,
|
|
"image": str(generated),
|
|
"model": "gpt-image-2-high",
|
|
"quality": "high",
|
|
}
|
|
|
|
monkeypatch.setattr(module, "_PROVIDER", Provider())
|
|
result = module._generate(
|
|
{
|
|
"prompt": "paint a blue sphere",
|
|
"aspect_ratio": "square",
|
|
"model": "gpt-image-2-high",
|
|
}
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["image_b64"]
|
|
assert "image" not in result
|
|
assert not generated.exists()
|
|
|
|
def test_image_broker_auto_falls_back_to_local_and_honors_explicit_routes(
|
|
monkeypatch,
|
|
):
|
|
"""AUTO is hosted-first while explicit local never calls the hosted lane."""
|
|
broker_path = HERMES / "scripts" / "image_broker.py"
|
|
spec = importlib.util.spec_from_file_location("hermes_image_router", broker_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
calls = []
|
|
|
|
def hosted(payload, model, prompt, aspect):
|
|
calls.append(("hosted", model, prompt, aspect))
|
|
return {"success": False, "error": "hosted refusal"}
|
|
|
|
def local(payload, timeout=1800.0):
|
|
calls.append(("local", payload["model"], timeout))
|
|
return {
|
|
"success": True,
|
|
"image_b64": "aW1hZ2U=",
|
|
"model": "flux-2-klein-4b-local",
|
|
"route": "local",
|
|
}
|
|
|
|
monkeypatch.setattr(module, "_generate_hosted", hosted)
|
|
monkeypatch.setattr(module, "_local_request", local)
|
|
auto = module._generate(
|
|
{
|
|
"prompt": "colorize this family photograph",
|
|
"aspect_ratio": "portrait",
|
|
"model": "atlas-image-auto-high",
|
|
}
|
|
)
|
|
assert auto["success"] is True
|
|
assert auto["route"] == "local"
|
|
assert auto["hosted_fallback_reason"] == "hosted refusal"
|
|
assert [call[0] for call in calls] == ["hosted", "local"]
|
|
|
|
calls.clear()
|
|
explicit = module._generate(
|
|
{
|
|
"prompt": "make a local landscape",
|
|
"aspect_ratio": "landscape",
|
|
"model": "flux-2-klein-4b-local",
|
|
}
|
|
)
|
|
assert explicit["route"] == "local"
|
|
assert [call[0] for call in calls] == ["local"]
|
|
|
|
def test_image_broker_policy_is_narrow_and_operator_extensible(tmp_path: Path, monkeypatch):
|
|
"""Family-photo restoration stays allowed while the hard boundary remains."""
|
|
broker_path = HERMES / "scripts" / "image_broker.py"
|
|
spec = importlib.util.spec_from_file_location("hermes_image_policy", broker_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
policy = tmp_path / "policy.json"
|
|
policy.write_text('{"additional_blocked_phrases":["site-specific block"]}')
|
|
monkeypatch.setattr(module, "POLICY_PATH", policy)
|
|
|
|
assert module._policy_error(
|
|
"Colorize my baby photograph with a lighter natural skin tone"
|
|
) is None
|
|
assert "minors" in module._policy_error("Create a sexual image of a child")
|
|
assert module._policy_error("A site-specific block request") == (
|
|
"request is blocked by the operator image policy"
|
|
)
|