diff --git a/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py b/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py index a313cc80..06475c50 100644 --- a/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py +++ b/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()) diff --git a/testing/tests/test_hermes_image_builder_branch_edges.py b/testing/tests/test_hermes_image_builder_branch_edges.py new file mode 100644 index 00000000..8ada5cbb --- /dev/null +++ b/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