"""Contracts for automatic deployment of validated Hermes image releases.""" from __future__ import annotations from pathlib import Path import re import pytest import yaml ROOT = Path(__file__).resolve().parents[2] SERVICE = ROOT / "services/hermes" APPLICATIONS = ROOT / "clusters/atlas/flux-system/applications" AGENT_MARKER = '"$imagepolicy": "hermes:hermes-agent-release:digest"' HOLD_KEYS = { "hermes.bstein.dev/agent-image-release-hold", "hermes.bstein.dev/agent-image-release-hold-digest", "hermes.bstein.dev/agent-image-release-hold-minimum-source", "hermes.bstein.dev/agent-image-release-hold-reason", } def _agent_release_binding(agent_text: str) -> str: """Validate exactly one normal setter or one explicit temporary hold.""" document = yaml.safe_load(agent_text) annotations = document.get("metadata", {}).get("annotations", {}) setter_count = agent_text.count(AGENT_MARKER) hold_present = bool(HOLD_KEYS & set(annotations)) if setter_count == 1: assert not hold_present, "agent image cannot have both setter and release hold" return "setter" assert setter_count == 0, "agent image has duplicate release setters" assert annotations.get("hermes.bstein.dev/agent-image-release-hold") == "true" image = next( item for item in document.get("images", []) if item.get("name") == "registry.bstein.dev/bstein/hermes-agent" ) digest = annotations.get("hermes.bstein.dev/agent-image-release-hold-digest", "") assert re.fullmatch(r"sha256:[0-9a-f]{64}", digest), "hold digest is invalid" assert image.get("digest") == digest, "hold digest does not match the pinned image" source = annotations.get("hermes.bstein.dev/agent-image-release-hold-minimum-source", "") assert re.fullmatch(r"[0-9a-f]{40}", source), "hold source must be a full commit" assert annotations.get("hermes.bstein.dev/agent-image-release-hold-reason", "").strip(), ( "hold reason is required" ) assert HOLD_KEYS <= set(annotations), "agent image hold metadata is incomplete" return "hold" def test_image_policies_observe_only_validated_release_tags() -> None: """Candidates remain invisible until Jenkins publishes the release suffix.""" documents = list( yaml.safe_load_all((SERVICE / "image.yaml").read_text(encoding="utf-8")) ) repositories = { item["metadata"]["name"]: item for item in documents if item["kind"] == "ImageRepository" } policies = { item["metadata"]["name"]: item for item in documents if item["kind"] == "ImagePolicy" } assert set(repositories) == { "hermes-agent-release", "hermes-chat-router-release", "hermes-webui-release", "hermes-stt-release", "hermes-tts-release", } assert set(policies) == set(repositories) for name, policy in policies.items(): assert policy["metadata"]["namespace"] == "hermes" assert policy["spec"]["imageRepositoryRef"]["name"] == name assert policy["spec"]["filterTags"] == { "pattern": ("^git-[0-9a-f]{40}-build-" "(?P[1-9][0-9]*)-release$"), "extract": "$build", } assert policy["spec"]["policy"] == {"numerical": {"order": "asc"}} assert policy["spec"]["digestReflectionPolicy"] == "Always" def test_flux_updates_only_the_reviewed_hermes_image_digests() -> None: """Flux persists selected digests to Git and rolls all matching workloads.""" service_kustomization = (SERVICE / "kustomization.yaml").read_text(encoding="utf-8") applications_kustomization = (APPLICATIONS / "kustomization.yaml").read_text( encoding="utf-8" ) automation = yaml.safe_load( (APPLICATIONS / "hermes/image-automation.yaml").read_text(encoding="utf-8") ) agent = (SERVICE / "kustomization.yaml").read_text(encoding="utf-8") chat = (SERVICE / "chat-statefulset.yaml").read_text(encoding="utf-8") dashboard = (SERVICE / "deployment.yaml").read_text(encoding="utf-8") voice = (SERVICE / "voice-deployment.yaml").read_text(encoding="utf-8") router = (SERVICE / "chat-router.yaml").read_text(encoding="utf-8") assert " - image.yaml" in service_kustomization assert " - hermes/image-automation.yaml" in applications_kustomization assert automation["spec"]["git"]["checkout"]["ref"]["branch"] == "main" assert automation["spec"]["git"]["push"]["branch"] == "main" assert automation["spec"]["update"] == { "strategy": "Setters", "path": "services/hermes", } assert _agent_release_binding(agent) in {"setter", "hold"} webui_marker = '"$imagepolicy": "hermes:hermes-webui-release"' chat_object = yaml.safe_load(chat) containers = chat_object["spec"]["template"]["spec"]["containers"] hux = next((item for item in containers if item["name"] == "hux"), None) producer = next( (item for item in containers if item["name"] == "hux-evidence-producer"), None, ) # Before activation the chat StatefulSet carries exactly one WebUI # consumer and no HUX setters; after activation the HUX sidecar (and the # HUX-12 evidence producer) consume the very same reviewed image line, # with tag and digest setters binding the sidecar's build metadata. assert producer is None or hux is not None consumers = 1 + (1 if hux else 0) + (1 if producer else 0) assert chat.count(webui_marker) == consumers assert dashboard.count(webui_marker) == 1 assert chat.count( '"$imagepolicy": "hermes:hermes-webui-release:tag"' ) == (1 if hux else 0) assert chat.count( '"$imagepolicy": "hermes:hermes-webui-release:digest"' ) == (1 if hux else 0) if hux: hux_env = { item["name"]: item["value"] for item in hux["env"] if "value" in item } release = re.fullmatch( r"git-([0-9a-f]{40})-build-[1-9][0-9]*-release", hux_env["HUX_IMAGE_TAG"], ) assert release is not None assert hux["image"].split(":git-", 1)[1].split("@", 1)[0] == hux_env[ "HUX_IMAGE_TAG" ].removeprefix("git-") assert hux["image"].endswith("@" + hux_env["HUX_IMAGE_DIGEST"]) if producer is not None: assert producer["image"] == hux["image"] # A digest-only setter replaces the complete YAML scalar with ``sha256:...``. # Whole-image setters must retain the registry and repository in pod specs. for workload in (chat, dashboard): marked_line = next( line for line in workload.splitlines() if webui_marker in line ) assert "registry.bstein.dev/bstein/hermes-webui:" in marked_line assert "@sha256:" in marked_line for component in ("stt", "tts"): marker = f'"$imagepolicy": "hermes:hermes-{component}-release"' assert voice.count(marker) == 1 marked_line = next(line for line in voice.splitlines() if marker in line) assert f"registry.bstein.dev/bstein/hermes-jetson-{component}" in marked_line assert "@sha256:" in marked_line router_marker = '"$imagepolicy": "hermes:hermes-chat-router-release"' assert router.count(router_marker) == 1 marked_line = next(line for line in router.splitlines() if router_marker in line) assert "registry.bstein.dev/bstein/hermes-chat-router" in marked_line assert "@sha256:" in marked_line def test_agent_release_hold_rejects_mismatched_digest_and_blank_reason() -> None: """A temporary hold must bind the exact image and explain its purpose.""" image_digest = "sha256:" + "a" * 64 def hold_text(*, digest: str = image_digest, reason: str = "verified hold") -> str: return yaml.safe_dump( { "apiVersion": "kustomize.config.k8s.io/v1beta1", "kind": "Kustomization", "metadata": { "annotations": { "hermes.bstein.dev/agent-image-release-hold": "true", "hermes.bstein.dev/agent-image-release-hold-digest": digest, "hermes.bstein.dev/agent-image-release-hold-minimum-source": "b" * 40, "hermes.bstein.dev/agent-image-release-hold-reason": reason, } }, "images": [ { "name": "registry.bstein.dev/bstein/hermes-agent", "digest": image_digest, } ], } ) bad_digest = hold_text(digest="sha256:" + "0" * 64) with pytest.raises(AssertionError, match="digest"): _agent_release_binding(bad_digest) blank_reason = hold_text(reason="") with pytest.raises(AssertionError, match="reason"): _agent_release_binding(blank_reason)