test: close merged image lane quality gaps

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_contract.py

@@ -0,0 +1,181 @@
+"""Exercise exact fail-closed branches in the reviewed Hermes image lane."""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+RELEASE = ROOT / "ci/scripts/hermes_image_release.py"
+HARBOR = 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):
+    """Small bounded response for same-process HTTP contract tests."""
+
+    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:
+    """Return one reviewed sequence of Harbor API responses."""
+
+    origin = "https://registry.bstein.dev/api/v2.0"
+
+    def __init__(self, responses) -> None:
+        self.responses = list(responses)
+
+    def request(self, _method, _path, _payload=None):
+        return self.responses.pop(0)
+
+
+def _count(value: int) -> dict[str, str]:
+    return {"X-Total-Count": str(value)}
+
+
+def _robot(module, *, immutable: bool, duration: int | None = None) -> 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 duration is not None:
+        robot["duration"] = duration
+    return robot
+
+
+def test_release_policy_response_edges_are_fail_closed() -> None:
+    """Credentials, bounds, page evidence, JSON shape, and redirects stay strict."""
+    module = _load(RELEASE, "hermes_release_policy_branch_contract")
+    assert module._NoRedirect().redirect_request(None, None, 302, "", {}, "") is None
+    with pytest.raises(RuntimeError, match="credentials"):
+        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),
+        )
+    with pytest.raises(RuntimeError, match="valid total count"):
+        module._require_complete_rule_page([], {})
+    with pytest.raises(RuntimeError, match="invalid shape"):
+        module.verify_immutable_policy(
+            username="robot",
+            password="private",
+            opener=lambda *_args: Response(200, b"{}", _count(0)),
+        )
+
+
+def test_release_renderer_exhausts_a_trailing_incomplete_image() -> None:
+    """A trailing matching name cannot hide the one complete digest entry."""
+    module = _load(RELEASE, "hermes_release_trailing_image_branch")
+    digest = "sha256:" + "9" * 64
+    source = (
+        "images:\n"
+        f"  - name: {module.DEFAULT_IMAGE}\n"
+        "    digest: sha256:" + "0" * 64 + "\n"
+        f"  - name: {module.DEFAULT_IMAGE}\n"
+    )
+    assert module.render_kustomization(source, digest).count(digest) == 1
+
+
+def test_harbor_publisher_handles_absent_duration_and_stale_reread(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An absent optional duration is preserved while stale policy reads retry."""
+    module = _load(HARBOR, "harbor_publisher_branch_contract")
+    current = _robot(module, immutable=False)
+    malformed_persisted = {**_robot(module, immutable=True), "permissions": "bad"}
+    exact = _robot(module, immutable=True)
+    client = FakeClient(
+        [
+            (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
+            (200, json.dumps(current).encode(), {}),
+            (200, b"", {}),
+            (200, json.dumps(malformed_persisted).encode(), {}),
+            (200, json.dumps(exact).encode(), {}),
+        ]
+    )
+    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+    assert module.ensure_publisher_can_read_rule(client) == 41
+
+
+def test_harbor_created_rule_retries_an_id_mismatch(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A stale exact-looking rule with the wrong ID is never accepted."""
+    module = _load(HARBOR, "harbor_rule_id_retry_branch")
+    path = "/api/v2.0/projects/bstein/immutabletagrules/23"
+    wrong = {"id": 22, **module.EXPECTED_RULE}
+    exact = {"id": 23, **module.EXPECTED_RULE}
+    client = FakeClient(
+        [
+            (200, b"[]", _count(0)),
+            (201, b"", {"Location": path}),
+            (200, json.dumps([wrong]).encode(), _count(1)),
+            (200, json.dumps([exact]).encode(), _count(1)),
+        ]
+    )
+    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+    assert module.ensure_rule(client) == 23
+
+
+def test_harbor_main_covers_immediate_success_and_final_retry(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """Startup exits directly on success and re-raises the final unavailable result."""
+    module = _load(HARBOR, "harbor_main_branch_contract")
+    password = tmp_path / "password"
+    password.write_text("private\n", encoding="utf-8")
+    monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN)
+    monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password))
+    monkeypatch.setattr(module, "HarborClient", lambda *_args: object())
+    monkeypatch.setattr(module, "ensure_rule", lambda _client: 17)
+    monkeypatch.setattr(module, "ensure_publisher_can_read_rule", lambda _client: 41)
+    assert module.main() == 0
+
+    def unavailable(_client):
+        raise module.HarborUnavailable("still warming")
+
+    monkeypatch.setattr(module, "ensure_rule", unavailable)
+    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+    with pytest.raises(module.HarborUnavailable, match="still warming"):
+        module.main()

testing/tests/test_hermes_runtime_access.py

@@ -451,9 +451,8 @@ def test_manifests_never_seed_access_material_into_persistent_env():
     assert triage_annotations[
         "vault.hashicorp.com/agent-inject-secret-triage-api-key"
     ] == "kv/data/atlas/hermes/triage-api"
-    gitea_api = (
-        ROOT / "services/hermes/scm-common/scripts/gitea_api.py"
-    ).read_text(encoding="utf-8")
+    gitea_path = ROOT / "services/hermes/scm-common/scripts/gitea_api.py"
+    gitea_api = gitea_path.read_text(encoding="utf-8")
     assert "scm_broker_client" in gitea_api
     assert "GITEA_TOKEN" not in gitea_api
     assert "gitea-token" not in annotations