hermes: close release-lane branch coverage gaps

Cover the uncovered branches in the image release script (redirect
handler, policy-read credential and size bounds, count and shape
rejections, EOF manifest scan) and in the Harbor immutability helper
(scope-failure verification, absent robot duration, stale-read retries,
created-rule ID binding, bounded startup retries). Exclude the
__main__ guard like the sibling release script.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py

@@ -374,5 +374,5 @@ def main() -> int:
     return 0
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover - exercised through main()
     raise SystemExit(main())

testing/tests/test_hermes_image_builder_branch_edges.py

@@ -0,0 +1,227 @@
+"""Branch-edge coverage for the Hermes release and Harbor policy scripts."""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+RELEASE = REPO_ROOT / "ci/scripts/hermes_image_release.py"
+HARBOR = (
+    REPO_ROOT
+    / "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.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 {}
+
+
+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 _publisher(module, *, immutable: bool = False, with_duration: bool = True) -> dict:
+    access = [
+        {"resource": "repository", "action": "pull"},
+        {"resource": "repository", "action": "push"},
+    ]
+    if immutable:
+        access.append({"resource": "immutable-tag", "action": "list"})
+    robot = {
+        "id": 41,
+        "name": module.PUBLISH_ROBOT,
+        "description": "publisher",
+        "level": "system",
+        "editable": True,
+        "disable": False,
+        "permissions": [
+            {"kind": "project", "namespace": module.PROJECT, "access": access}
+        ],
+    }
+    if with_duration:
+        robot["duration"] = -1
+    return robot
+
+
+def test_release_redirect_handler_never_forwards_credentials() -> None:
+    """The registry opener refuses to follow any redirect target."""
+    module = _load(RELEASE, "release_redirect_edges")
+    handler = module._NoRedirect()
+    assert handler.redirect_request(None, None, 0, None, None, None) is None
+
+
+def test_release_policy_read_requires_credentials_and_bounded_body() -> None:
+    """The policy preflight rejects empty credentials and oversized responses."""
+    module = _load(RELEASE, "release_policy_read_edges")
+    with pytest.raises(RuntimeError, match="credentials are empty"):
+        module._immutable_rules_response(username="", password="private")
+    with pytest.raises(RuntimeError, match="size limit"):
+        module._immutable_rules_response(
+            username="robot",
+            password="private",
+            opener=lambda *_args: Response(200, b"x" * 1_048_577),
+        )
+
+
+@pytest.mark.parametrize("headers", [{}, {"X-Total-Count": "many"}])
+def test_release_policy_rejects_missing_or_invalid_count(headers: dict) -> None:
+    """A proxy that strips or mangles count evidence cannot prove completeness."""
+    module = _load(RELEASE, "release_policy_count_edges")
+    with pytest.raises(RuntimeError, match="total count"):
+        module.verify_immutable_policy(
+            username="robot",
+            password="private",
+            opener=lambda *_args, evidence=headers: Response(200, b"[]", evidence),
+        )
+
+
+@pytest.mark.parametrize("body", [b"{}", b"[1]"])
+def test_release_policy_rejects_non_rule_list_shapes(body: bytes) -> None:
+    """Valid JSON that is not a list of rule objects fails the preflight."""
+    module = _load(RELEASE, "release_policy_shape_edges")
+    with pytest.raises(RuntimeError, match="invalid shape"):
+        module.verify_immutable_policy(
+            username="robot",
+            password="private",
+            opener=lambda *_args, value=body: Response(200, value, _count(1)),
+        )
+
+
+def test_renderer_ignores_trailing_entry_without_digest_at_eof() -> None:
+    """A dangling final image entry cannot shadow the one complete digest."""
+    module = _load(RELEASE, "release_render_eof_edges")
+    digest = "sha256:" + "6" * 64
+    source = (
+        "images:\n"
+        f"  - name: {module.DEFAULT_IMAGE}\n"
+        "    digest: sha256:" + "0" * 64 + "\n"
+        f"  - name: {module.DEFAULT_IMAGE}\n"
+        "    newTag: dangling\n"
+    )
+    rendered = module.render_kustomization(source, digest)
+    assert f"digest: {digest}\n" in rendered
+    assert rendered.endswith("newTag: dangling\n")
+
+
+def test_harbor_verified_publisher_rejects_exact_identity_with_bad_scope() -> None:
+    """An exact robot identity still fails verification on a malformed scope."""
+    module = _load(HARBOR, "harbor_verified_scope_edges")
+    robot = _publisher(module)
+    robot["permissions"] = "bad"
+    assert module._verified_publisher(robot) is False
+
+
+def test_harbor_publisher_update_omits_absent_duration(monkeypatch) -> None:
+    """A robot without a configured duration is preserved without inventing one."""
+    module = _load(HARBOR, "harbor_duration_edges")
+    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+    verified = _publisher(module, immutable=True)
+    client = FakeClient(
+        [
+            (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
+            (200, json.dumps(_publisher(module, with_duration=False)).encode(), {}),
+            (200, b"", {}),
+            (200, json.dumps(verified).encode(), {}),
+        ]
+    )
+    assert module.ensure_publisher_can_read_rule(client) == 41
+    method, path, payload = client.calls[2]
+    assert (method, path) == ("PUT", "/robots/41")
+    assert "duration" not in payload
+
+
+def test_harbor_publisher_verify_retries_past_invalid_json(monkeypatch) -> None:
+    """One stale unreadable reread retries instead of passing or failing outright."""
+    module = _load(HARBOR, "harbor_verify_retry_edges")
+    sleeps = []
+    monkeypatch.setattr(module.time, "sleep", sleeps.append)
+    verified = _publisher(module, immutable=True)
+    client = FakeClient(
+        [
+            (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
+            (200, json.dumps(_publisher(module)).encode(), {}),
+            (200, b"", {}),
+            (200, b"not-json", {}),
+            (200, json.dumps(verified).encode(), {}),
+        ]
+    )
+    assert module.ensure_publisher_can_read_rule(client) == 41
+    assert sleeps == [1]
+
+
+def test_harbor_rule_verify_requires_the_created_rule_id(monkeypatch) -> None:
+    """A matching rule under a different ID is a stale read, not creation proof."""
+    module = _load(HARBOR, "harbor_rule_id_retry_edges")
+    sleeps = []
+    monkeypatch.setattr(module.time, "sleep", sleeps.append)
+    location = "/api/v2.0/projects/bstein/immutabletagrules/23"
+    client = FakeClient(
+        [
+            (200, b"[]", _count(0)),
+            (201, b"", {"Location": location}),
+            (200, json.dumps([{"id": 99, **module.EXPECTED_RULE}]).encode(), _count(1)),
+            (200, json.dumps([{"id": 23, **module.EXPECTED_RULE}]).encode(), _count(1)),
+        ]
+    )
+    assert module.ensure_rule(client) == 23
+    assert sleeps == [1]
+
+
+def test_harbor_main_reraises_after_bounded_unavailable_retries(
+    monkeypatch, tmp_path: Path
+) -> None:
+    """Persistent Harbor unavailability surfaces instead of looping forever."""
+    module = _load(HARBOR, "harbor_main_retry_edges")
+    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())
+
+    def never_ready(_client):
+        raise module.HarborUnavailable("still warming")
+
+    monkeypatch.setattr(module, "ensure_rule", never_ready)
+    sleeps = []
+    monkeypatch.setattr(module.time, "sleep", sleeps.append)
+    with pytest.raises(module.HarborUnavailable, match="still warming"):
+        module.main()
+    assert len(sleeps) == 11