hermes: add daemonless agent image release lane

ci/Jenkinsfile.hermes-agent-image

@@ -19,6 +19,7 @@ spec:
   nodeSelector:
     kubernetes.io/arch: arm64
     node-role.kubernetes.io/worker: "true"
+    hardware: rpi5
   affinity:
     nodeAffinity:
       requiredDuringSchedulingIgnoredDuringExecution:
@@ -33,14 +34,6 @@ spec:
                   - titan-19
                   - titan-22
                   - titan-24
-      preferredDuringSchedulingIgnoredDuringExecution:
-        - weight: 100
-          preference:
-            matchExpressions:
-              - key: hardware
-                operator: In
-                values:
-                  - rpi5
   imagePullSecrets:
     - name: harbor-bstein-robot
   containers:

dockerfiles/hermes-kaniko-heredoc-runner.py

@@ -86,14 +86,13 @@ def extract_blocks(source: str) -> list[tuple[tuple[str, ...], str]]:
                 continue
             logical, index = _logical_instruction(lines, index, escape)
             stripped = logical.lstrip()
-            if (
-                stripped[:3].upper() == "RUN"
-                and (len(stripped) == 3 or stripped[3].isspace())
-                and "<<" in stripped
+            if stripped[:3].upper() == "RUN" and (
+                len(stripped) == 3 or stripped[3].isspace()
             ):
-                raise ValueError(
-                    "unsupported RUN heredoc outside the exact reviewed contract"
-                )
+                if "<<" in stripped:
+                    raise ValueError(
+                        "unsupported RUN heredoc outside the exact reviewed contract"
+                    )
             index += 1
             continue
         end, command = marker

services/bstein-dev-home/kustomization.yaml

@@ -20,9 +20,9 @@ resources:
   - ingress.yaml
 images:
   - name: registry.bstein.dev/bstein/bstein-dev-home-frontend
-    newTag: 0.1.1-471 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-frontend:tag"}
+    newTag: 0.1.1-470 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-frontend:tag"}
   - name: registry.bstein.dev/bstein/bstein-dev-home-backend
-    newTag: 0.1.1-471 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-backend:tag"}
+    newTag: 0.1.1-470 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-backend:tag"}
 configMapGenerator:
   - name: chat-ai-gateway
     namespace: bstein-dev-home

services/hermes/NOTES.md

@@ -111,11 +111,6 @@ privileged container, or writable Git credential. It accepts only an exact
 40-character revision that is both the checked-out commit and current
 `atlas/titan-iac` `main`, so a human must merge the source PR first.
 
-The build pod requires an ARM64 worker and gives `hardware=rpi5` the maximum
-scheduler preference. Healthy rpi4 workers remain valid fallback capacity when
-the rpi5 pool cannot fit the full request. Cordoned, unhealthy, amd64, and
-reserved last-resort nodes remain excluded by the required placement boundary.
-
 Kaniko runs as UID 0 because it must unpack an image root filesystem and enter
 that filesystem for Dockerfile `RUN` instructions. Its capability set is the
 minimum proven by an exact ARM64 no-push build of this Dockerfile:

testing/quality_contract.json

@@ -15,9 +15,6 @@
   ],
   "managed_modules": [
     "ci/scripts/hermes_image_release.py",
-    "dockerfiles/hermes-kaniko-heredoc-runner.py",
-    "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py",
-    "services/hermes/scripts/jenkins_image_build_trigger.py",
     "ci/scripts/publish_test_metrics.py",
     "ci/scripts/publish_test_metrics_quality.py",
     "ci/scripts/semgrep_report.py",
@@ -40,9 +37,6 @@
   ],
   "lint_paths": [
     "ci/scripts/hermes_image_release.py",
-    "dockerfiles/hermes-kaniko-heredoc-runner.py",
-    "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py",
-    "services/hermes/scripts/jenkins_image_build_trigger.py",
     "ci/scripts/publish_test_metrics.py",
     "ci/scripts/publish_test_metrics_quality.py",
     "ci/scripts/semgrep_report.py",
@@ -65,10 +59,7 @@
       "junit": "build/junit-unit.xml",
       "coverage_sources": [
         "ci/scripts",
-        "dockerfiles",
         "scripts.render.dashboards_render_atlas",
-        "services/harbor/scripts",
-        "services/hermes/scripts",
         "services/mailu/scripts",
         "testing"
       ],
@@ -180,9 +171,6 @@
     "minimum_percent": 95.0,
     "tracked_files": [
       "ci/scripts/hermes_image_release.py",
-      "dockerfiles/hermes-kaniko-heredoc-runner.py",
-      "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py",
-      "services/hermes/scripts/jenkins_image_build_trigger.py",
       "ci/scripts/publish_test_metrics.py",
       "ci/scripts/publish_test_metrics_quality.py",
       "ci/scripts/semgrep_report.py",

testing/tests/test_hermes_image_builder.py

@@ -93,6 +93,27 @@ def test_builder_pod_is_daemonless_and_kernel_bounded() -> None:
     assert "add" not in containers["python"]["securityContext"]["capabilities"]
 
 
+def test_builder_is_restricted_to_healthy_rpi5_capacity() -> None:
+    """Disposable builds must stay off unhealthy and reserved Atlas nodes."""
+    spec = _pod_spec()
+    assert spec["nodeSelector"]["hardware"] == "rpi5"
+    expressions = spec["affinity"]["nodeAffinity"][
+        "requiredDuringSchedulingIgnoredDuringExecution"
+    ]["nodeSelectorTerms"][0]["matchExpressions"]
+    host_rule = next(
+        rule for rule in expressions if rule["key"] == "kubernetes.io/hostname"
+    )
+    assert host_rule["operator"] == "NotIn"
+    assert set(host_rule["values"]) >= {
+        "titan-04",
+        "titan-14",
+        "titan-18",
+        "titan-19",
+        "titan-22",
+        "titan-24",
+    }
+
+
 def test_pipeline_requires_reviewed_main_and_runtime_credentials() -> None:
     """Publishing requires explicit confirmation and a reviewed main commit."""
     source = PIPELINE_PATH.read_text(encoding="utf-8")

testing/tests/test_hermes_image_builder_coverage.py

@@ -1,456 +0,0 @@
-"""Boundary coverage for the three security-critical image-lane helpers."""
-
-from __future__ import annotations
-
-import importlib.util
-import io
-import json
-import sys
-import urllib.error
-from pathlib import Path
-
-import pytest
-
-
-REPO_ROOT = Path(__file__).resolve().parents[2]
-RUNNER = REPO_ROOT / "dockerfiles/hermes-kaniko-heredoc-runner.py"
-DOCKERFILE = REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent"
-HARBOR = (
-    REPO_ROOT
-    / "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py"
-)
-TRIGGER = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
-
-
-def _load(path: Path, name: str):
-    spec = importlib.util.spec_from_file_location(name, path)
-    assert spec is not None and spec.loader is not None
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
-class Response(io.BytesIO):
-    """Context-managed urllib response used without network access."""
-
-    def __init__(
-        self,
-        status: int,
-        body: bytes = b"",
-        headers: dict[str, str] | None = None,
-    ) -> None:
-        super().__init__(body)
-        self.status = status
-        self.headers = headers or {}
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, *_args):
-        self.close()
-
-
-class FakeClient:
-    """Ordered Harbor response fake with the pinned API origin."""
-
-    origin = "https://registry.bstein.dev/api/v2.0"
-
-    def __init__(self, responses) -> None:
-        self.responses = list(responses)
-        self.calls = []
-
-    def request(self, method, path, payload=None):
-        self.calls.append((method, path, payload))
-        return self.responses.pop(0)
-
-
-def _count(value: int) -> dict[str, str]:
-    return {"X-Total-Count": str(value)}
-
-
-def _robot(module, *, immutable: bool = False, duration=-1) -> dict:
-    access = [
-        {"resource": "repository", "action": "pull"},
-        {"resource": "repository", "action": "push"},
-    ]
-    if immutable:
-        access.append({"resource": "immutable-tag", "action": "list"})
-    return {
-        "id": 41,
-        "name": module.PUBLISH_ROBOT,
-        "description": "publisher",
-        "level": "system",
-        "duration": duration,
-        "editable": True,
-        "disable": False,
-        "permissions": [
-            {
-                "kind": "project",
-                "namespace": module.PROJECT,
-                "access": access,
-            }
-        ],
-    }
-
-
-def _robot_list(module, *, robot_id=41, name=None):
-    return json.dumps(
-        [{"id": robot_id, "name": name or module.PUBLISH_ROBOT}]
-    ).encode()
-
-
-def test_runner_rejects_directive_and_heredoc_boundaries(tmp_path: Path) -> None:
-    """Malformed parser state, bodies, sizes, and indices all fail closed."""
-    module = _load(RUNNER, "runner_boundary_coverage")
-    assert module._escape_character([""]) == "\\"
-    with pytest.raises(ValueError, match="multiple"):
-        module._escape_character(["# escape=\\", "# escape=\\"])
-    with pytest.raises(ValueError, match="unsupported"):
-        module._escape_character(["# escape=^"])
-    with pytest.raises(ValueError, match="unterminated.*continuation"):
-        module._logical_instruction(["R\\"], 0, "\\")
-    with pytest.raises(ValueError, match="unterminated node"):
-        module.extract_blocks("RUN node <<'NODE'\nbody")
-
-    source = DOCKERFILE.read_text(encoding="utf-8")
-    marker = "RUN node <<'NODE'\n"
-    body_start = source.index(marker) + len(marker)
-    body_end = source.index("\nNODE", body_start)
-    empty = source[:body_start] + source[body_end + 1 :]
-    with pytest.raises(ValueError, match="empty node"):
-        module.extract_blocks(empty)
-
-    dockerfile = tmp_path / "Dockerfile"
-    dockerfile.write_text("", encoding="utf-8")
-    with pytest.raises(ValueError, match="size"):
-        module.replay(dockerfile, 1)
-    with dockerfile.open("wb") as stream:
-        stream.truncate(module.MAX_DOCKERFILE_BYTES + 1)
-    with pytest.raises(ValueError, match="size"):
-        module.replay(dockerfile, 1)
-    dockerfile.write_text(source, encoding="utf-8")
-    for index in (0, 10):
-        with pytest.raises(ValueError, match="index"):
-            module.replay(dockerfile, index)
-
-
-def test_runner_main_dispatches_the_exact_path_and_index(
-    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
-) -> None:
-    """The command-line wrapper cannot omit either bounded input."""
-    module = _load(RUNNER, "runner_main_coverage")
-    dockerfile = tmp_path / "Dockerfile"
-    seen = []
-    monkeypatch.setattr(module, "replay", lambda path, index: seen.append((path, index)))
-    monkeypatch.setattr(
-        sys,
-        "argv",
-        ["runner", "--dockerfile", str(dockerfile), "--block-index", "4"],
-    )
-    assert module.main() == 0
-    assert seen == [(dockerfile, 4)]
-
-
-def test_trigger_redirect_wrapper_accepts_only_plugin_303(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """The low-level opener returns a real response or the one expected redirect."""
-    module = _load(TRIGGER, "trigger_opener_coverage")
-    assert module._NoRedirect().redirect_request(None, None, 0, None, None, None) is None
-    request = module.urllib.request.Request(module.JENKINS_BUILD_URL)
-
-    class Opener:
-        def __init__(self, result) -> None:
-            self.result = result
-
-        def open(self, _request, timeout):
-            assert timeout == 9
-            if isinstance(self.result, BaseException):
-                raise self.result
-            return self.result
-
-    success = Response(201)
-    monkeypatch.setattr(
-        module.urllib.request, "build_opener", lambda *_args: Opener(success)
-    )
-    assert module._open_without_redirect(request, 9) is success
-    redirect = module.urllib.error.HTTPError(
-        request.full_url, 303, "queued", {"Location": "/queue/item/1/"}, None
-    )
-    monkeypatch.setattr(
-        module.urllib.request, "build_opener", lambda *_args: Opener(redirect)
-    )
-    assert module._open_without_redirect(request, 9) is redirect
-    denied = module.urllib.error.HTTPError(request.full_url, 403, "denied", {}, None)
-    monkeypatch.setattr(
-        module.urllib.request, "build_opener", lambda *_args: Opener(denied)
-    )
-    with pytest.raises(urllib.error.HTTPError):
-        module._open_without_redirect(request, 9)
-
-
-def test_trigger_main_reports_safe_success_and_errors(
-    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
-) -> None:
-    """CLI output contains safe metadata and converts expected failures to JSON."""
-    module = _load(TRIGGER, "trigger_main_coverage")
-    revision = "a" * 40
-    monkeypatch.setattr(sys, "argv", ["trigger", revision])
-    monkeypatch.setattr(
-        module,
-        "trigger_build",
-        lambda value: {"job": module.JOB_NAME, "source_revision": value, "status": 201},
-    )
-    assert module.main() == 0
-    assert json.loads(capsys.readouterr().out)["source_revision"] == revision
-    monkeypatch.setattr(
-        module, "trigger_build", lambda _value: (_ for _ in ()).throw(OSError("closed"))
-    )
-    assert module.main() == 1
-    assert json.loads(capsys.readouterr().out) == {"error": "closed"}
-
-
-def test_harbor_http_client_bounds_requests_and_failures(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Credentials stay same-origin and all response/error paths remain bounded."""
-    module = _load(HARBOR, "harbor_client_coverage")
-    assert module.NoRedirect().redirect_request(None, None, 0, None, None, None) is None
-
-    class Opener:
-        result = Response(200, b"ok", {"X-Test": "yes"})
-
-        def open(self, request, timeout):
-            self.request = request
-            assert timeout == 20
-            if isinstance(self.result, BaseException):
-                raise self.result
-            return self.result
-
-    opener = Opener()
-    monkeypatch.setattr(module.urllib.request, "build_opener", lambda *_args: opener)
-    client = module.HarborClient(module.EXPECTED_ORIGIN + "/", "admin", "private")
-    status, body, headers = client.request("POST", "/rules", {"value": 1})
-    assert (status, body, headers["X-Test"]) == (200, b"ok", "yes")
-    assert opener.request.get_header("Content-type") == "application/json"
-
-    opener.result = module.urllib.error.HTTPError(
-        module.EXPECTED_ORIGIN, 404, "missing", {}, io.BytesIO(b"missing")
-    )
-    assert client.request("GET", "/missing")[:2] == (404, b"missing")
-    opener.result = module.urllib.error.URLError("offline")
-    with pytest.raises(module.HarborUnavailable):
-        client.request("GET", "/rules")
-    opener.result = Response(200, b"x" * (module.MAX_RESPONSE + 1))
-    with pytest.raises(RuntimeError, match="size"):
-        client.request("GET", "/rules")
-
-
-@pytest.mark.parametrize("body", [b"\xff", b"not-json", b"{}", b"[1]"])
-def test_harbor_json_lists_reject_invalid_shapes(body: bytes) -> None:
-    """Malformed encodings, JSON, objects, and scalar members are never lists."""
-    module = _load(HARBOR, f"harbor_json_{body!r}")
-    with pytest.raises(RuntimeError, match="invalid"):
-        module._json_list(body, "fixture")
-
-
-@pytest.mark.parametrize(
-    "robot,match",
-    [
-        ({"permissions": "bad"}, "permissions"),
-        ({"permissions": [1]}, "permissions"),
-        ({"permissions": []}, "one existing"),
-        (
-            {
-                "permissions": [
-                    {"kind": "project", "namespace": "bstein", "access": []},
-                    {"kind": "project", "namespace": "bstein", "access": []},
-                ]
-            },
-            "one existing",
-        ),
-        (
-            {
-                "permissions": [
-                    {"kind": "project", "namespace": "bstein", "access": "bad"}
-                ]
-            },
-            "access",
-        ),
-        (
-            {
-                "permissions": [
-                    {"kind": "project", "namespace": "bstein", "access": []}
-                ]
-            },
-            "pull/push",
-        ),
-    ],
-)
-def test_publisher_scope_rejects_malformed_or_incomplete_permissions(
-    robot: dict, match: str
-) -> None:
-    """Robot mutations require one structurally exact existing push scope."""
-    module = _load(HARBOR, f"harbor_scope_{match}")
-    with pytest.raises(RuntimeError, match=match):
-        module._publisher_scope(robot)
-    assert module._verified_publisher(robot) is False
-
-
-@pytest.mark.parametrize(
-    "responses,match,unavailable",
-    [
-        ([(503, b"", {})], "503", True),
-        ([(403, b"", {})], "403", False),
-        ([(200, b"[]", _count(0))], "exactly one", False),
-        ([(200, b"[]", _count(0))], "exactly one", False),
-        ([(200, b'[{\"name\":\"robot$jenkins-pipelines\",\"id\":0}]', _count(1))], "valid ID", False),
-    ],
-)
-def test_publisher_list_rejects_status_count_and_id(
-    responses, match: str, unavailable: bool
-) -> None:
-    """Publisher discovery distinguishes readiness from permanent policy errors."""
-    module = _load(HARBOR, f"harbor_publisher_list_{match}_{unavailable}")
-    error = module.HarborUnavailable if unavailable else RuntimeError
-    with pytest.raises(error, match=match):
-        module.ensure_publisher_can_read_rule(FakeClient(responses))
-
-
-@pytest.mark.parametrize(
-    "second,match,unavailable",
-    [
-        ((503, b"", {}), "503", True),
-        ((403, b"", {}), "403", False),
-        ((200, b"not-json", {}), "invalid robot JSON", False),
-        ((200, b"[]", {}), "invalid shape", False),
-        ((200, b'{}', {}), "not active and exact", False),
-    ],
-)
-def test_publisher_read_rejects_status_and_identity(
-    second, match: str, unavailable: bool
-) -> None:
-    """The detailed robot reread must be available, valid, and exact."""
-    module = _load(HARBOR, f"harbor_publisher_read_{match}_{unavailable}")
-    responses = [(200, _robot_list(module), _count(1)), second]
-    error = module.HarborUnavailable if unavailable else RuntimeError
-    with pytest.raises(error, match=match):
-        module.ensure_publisher_can_read_rule(FakeClient(responses))
-
-
-def test_publisher_update_rejects_duration_reference_and_status(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Only the preserved scope and a successful policy PUT can be accepted."""
-    module = _load(HARBOR, "harbor_publisher_update_coverage")
-    for duration in ("forever",):
-        responses = [
-            (200, _robot_list(module), _count(1)),
-            (200, json.dumps(_robot(module, duration=duration)).encode(), {}),
-        ]
-        with pytest.raises(RuntimeError, match="duration"):
-            module.ensure_publisher_can_read_rule(FakeClient(responses))
-
-    robot = _robot(module)
-    monkeypatch.setattr(module, "_publisher_scope", lambda _robot: ({}, []))
-    responses = [
-        (200, _robot_list(module), _count(1)),
-        (200, json.dumps(robot).encode(), {}),
-    ]
-    with pytest.raises(RuntimeError, match="normalization"):
-        module.ensure_publisher_can_read_rule(FakeClient(responses))
-
-    monkeypatch.undo()
-    for status, error in ((503, module.HarborUnavailable), (409, RuntimeError)):
-        responses = [
-            (200, _robot_list(module), _count(1)),
-            (200, json.dumps(_robot(module)).encode(), {}),
-            (status, b"", {}),
-        ]
-        with pytest.raises(error):
-            module.ensure_publisher_can_read_rule(FakeClient(responses))
-
-
-def test_publisher_and_rule_verification_retries_are_bounded(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Stale reads may retry, but never silently become successful evidence."""
-    module = _load(HARBOR, "harbor_retry_coverage")
-    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
-    responses = [
-        (200, _robot_list(module), _count(1)),
-        (200, json.dumps(_robot(module)).encode(), {}),
-        (200, b"", {}),
-        *((500, b"", {}) for _ in range(5)),
-    ]
-    with pytest.raises(RuntimeError, match="did not verify"):
-        module.ensure_publisher_can_read_rule(FakeClient(responses))
-
-    rule_path = "/api/v2.0/projects/bstein/immutabletagrules/23"
-    responses = [
-        (200, b"[]", _count(0)),
-        (201, b"", {"Location": rule_path}),
-        *((200, b"[]", _count(0)) for _ in range(5)),
-    ]
-    with pytest.raises(RuntimeError, match="did not verify"):
-        module.ensure_rule(FakeClient(responses))
-
-
-def test_rule_rejects_ids_create_status_and_location_suffix() -> None:
-    """Neither existing nor newly created rules can omit their exact positive ID."""
-    module = _load(HARBOR, "harbor_rule_id_coverage")
-    bad_rule = {"id": 0, **module.EXPECTED_RULE}
-    with pytest.raises(RuntimeError, match="valid ID"):
-        module.ensure_rule(
-            FakeClient([(200, json.dumps([bad_rule]).encode(), _count(1))])
-        )
-    with pytest.raises(module.HarborUnavailable):
-        module.ensure_rule(
-            FakeClient([(200, b"[]", _count(0)), (503, b"", {})])
-        )
-    with pytest.raises(RuntimeError, match="invalid ID"):
-        module.ensure_rule(
-            FakeClient(
-                [
-                    (200, b"[]", _count(0)),
-                    (
-                        201,
-                        b"",
-                        {
-                            "location": "/api/v2.0/projects/bstein/immutabletagrules/nope"
-                        },
-                    ),
-                ]
-            )
-        )
-
-
-def test_harbor_main_retries_then_reports_exact_ids(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-    capsys: pytest.CaptureFixture[str],
-) -> None:
-    """Runtime credentials are required and transient startup failures are bounded."""
-    module = _load(HARBOR, "harbor_main_coverage")
-    password_file = tmp_path / "password"
-    password_file.write_text("private\n", encoding="utf-8")
-    monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN)
-    monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password_file))
-    monkeypatch.setattr(module, "HarborClient", lambda *_args: object())
-    attempts = iter([module.HarborUnavailable("warming"), 17])
-
-    def ensure_rule(_client):
-        result = next(attempts)
-        if isinstance(result, BaseException):
-            raise result
-        return result
-
-    monkeypatch.setattr(module, "ensure_rule", ensure_rule)
-    monkeypatch.setattr(module, "ensure_publisher_can_read_rule", lambda _client: 41)
-    monkeypatch.setattr(module.time, "sleep", lambda seconds: None)
-    assert module.main() == 0
-    assert "id=17" in capsys.readouterr().out
-
-    password_file.write_text("\n", encoding="utf-8")
-    with pytest.raises(RuntimeError, match="empty"):
-        module.main()

testing/tests/test_hermes_image_builder_fresh_review.py

@@ -32,41 +32,6 @@ def _pod_spec() -> dict:
     return yaml.safe_load(pod_yaml)["spec"]
 
 
-def test_builder_prefers_rpi5_with_healthy_arm64_worker_fallback() -> None:
-    """Disposable builds prefer rpi5 without excluding schedulable rpi4 workers."""
-    spec = _pod_spec()
-    assert spec["nodeSelector"] == {
-        "kubernetes.io/arch": "arm64",
-        "node-role.kubernetes.io/worker": "true",
-    }
-    affinity = spec["affinity"]["nodeAffinity"]
-    assert affinity["preferredDuringSchedulingIgnoredDuringExecution"] == [
-        {
-            "weight": 100,
-            "preference": {
-                "matchExpressions": [
-                    {"key": "hardware", "operator": "In", "values": ["rpi5"]}
-                ]
-            },
-        }
-    ]
-    expressions = affinity["requiredDuringSchedulingIgnoredDuringExecution"][
-        "nodeSelectorTerms"
-    ][0]["matchExpressions"]
-    host_rule = next(
-        rule for rule in expressions if rule["key"] == "kubernetes.io/hostname"
-    )
-    assert host_rule["operator"] == "NotIn"
-    assert set(host_rule["values"]) >= {
-        "titan-04",
-        "titan-14",
-        "titan-18",
-        "titan-19",
-        "titan-22",
-        "titan-24",
-    }
-
-
 @pytest.mark.parametrize(
     "unsupported",
     [