"""The helm argv that the reconciler's correctness depends on. install() writes a label; list_releases() reads it back. Neither is checked by anything at runtime — if the two ever disagree, `helm list --selector` matches nothing, the reconciler sees an empty cluster, and every ready instance looks like it lost its release. That failure is silent and reads as "no drift", so it gets a test rather than a comment. These assert argv, not behaviour against a real cluster: tests/e2e covers that. The point here is that the two sides of the label agree, and that neither drops out under a refactor. """ from __future__ import annotations from pathlib import Path from typing import Any import httpx import pytest from svcforge_core.adapters import helm from svcforge_core.adapters.helm import ( MANAGED_BY_LABEL, MANAGED_BY_VALUE, HelmError, HelmProvisioner, ) from svcforge_core.domain.models import CatalogEntry, SizeSpec ENTRY = CatalogEntry( service_type="redis", chart="oci://example/redis", chart_version="1.2.3", sizes={"small": SizeSpec(replicas=1, resources={})}, ) def _capture(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]: """Record every argv HelmProvisioner would exec, and run none of them.""" seen: list[list[str]] = [] async def fake_run_helm(self: HelmProvisioner, argv: list[str]) -> str: seen.append(list(argv)) return "[]" monkeypatch.setattr(HelmProvisioner, "_run_helm", fake_run_helm, raising=True) return seen def _pair(argv: list[str], flag: str) -> str | None: """The value following `flag`, or None. Positional, because helm takes `--flag value`.""" return argv[argv.index(flag) + 1] if flag in argv else None @pytest.mark.asyncio async def test_install_labels_the_release_as_svcforge_managed( monkeypatch: pytest.MonkeyPatch, ) -> None: seen = _capture(monkeypatch) await HelmProvisioner(kubeconfig=Path("/dev/null")).install( "acme-redis", "tenant-acme", ENTRY, {"replicas": 1} ) assert len(seen) == 1 assert _pair(seen[0], "--labels") == f"{MANAGED_BY_LABEL}={MANAGED_BY_VALUE}" @pytest.mark.asyncio async def test_list_releases_asks_only_for_svcforge_releases( monkeypatch: pytest.MonkeyPatch, ) -> None: seen = _capture(monkeypatch) await HelmProvisioner(kubeconfig=Path("/dev/null")).list_releases() assert len(seen) == 1 argv = seen[0] assert _pair(argv, "--selector") == f"{MANAGED_BY_LABEL}={MANAGED_BY_VALUE}" # Still every namespace. Tenants get their own, so scoping to one would hide releases # rather than the cluster's; the label is what narrows this, not the namespace. assert "--all-namespaces" in argv @pytest.mark.asyncio async def test_the_label_written_is_the_label_read(monkeypatch: pytest.MonkeyPatch) -> None: """The regression that motivated this file. Asserted against each other rather than against a literal on both sides, so a rename that updates only one of install/list_releases fails here instead of in production as an empty drift check. """ seen = _capture(monkeypatch) prov = HelmProvisioner(kubeconfig=Path("/dev/null")) await prov.install("acme-redis", "tenant-acme", ENTRY, {}) await prov.list_releases() assert _pair(seen[0], "--labels") == _pair(seen[1], "--selector") def _meta(name: str, ns: str, version: str, status: str = "deployed") -> dict[str, object]: """One PartialObjectMetadata item as the API server returns it for a release secret.""" return { "metadata": { "name": f"sh.helm.release.v1.{name}.v{version}", "namespace": ns, "labels": {"name": name, "owner": "helm", "status": status, "version": version}, } } class _FakeResponse: def __init__(self, items: list[dict[str, object]]) -> None: self._items = items def raise_for_status(self) -> None: return None def json(self) -> dict[str, object]: return {"items": self._items} def _fake_api( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, items: list[dict[str, object]] ) -> dict[str, Any]: """Stand in for the in-cluster ServiceAccount and the API call it authenticates. The token is a real file at a repointed constant rather than a patched method: the module decides it is in-cluster by whether that file reads, so faking the decision at the filesystem keeps the test honest about what it is exercising. """ captured: dict[str, Any] = {} token = tmp_path / "token" token.write_text("tok", encoding="utf-8") ca = tmp_path / "ca.crt" ca.write_text("ca", encoding="utf-8") # a complete SA has both; the readability check needs it monkeypatch.setattr(helm, "_SA_TOKEN", token) monkeypatch.setattr(helm, "_SA_CA", ca) monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1") monkeypatch.setenv("KUBERNETES_SERVICE_PORT_HTTPS", "443") class _Client: def __init__(self, **kw: object) -> None: captured["client_kwargs"] = kw async def __aenter__(self) -> _Client: return self async def __aexit__(self, *exc: object) -> None: return None async def get(self, url: str, **kw: object) -> _FakeResponse: captured["url"] = url captured.update(kw) return _FakeResponse(items) monkeypatch.setattr(httpx, "AsyncClient", _Client) return captured @pytest.mark.asyncio async def test_api_read_collapses_a_release_to_its_newest_revision( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """helm writes one secret per revision. Ten revisions is one release, not ten.""" _fake_api( monkeypatch, tmp_path, [_meta("acme-redis", "tenant-acme", v) for v in ("1", "2", "10", "9")], ) out = await HelmProvisioner().list_releases() assert [(r.name, r.namespace) for r in out] == [("acme-redis", "tenant-acme")] # 10, not 9: string ordering would pick "9" and quietly report a stale revision. assert out[0].revision == 10 @pytest.mark.asyncio async def test_api_read_keeps_same_named_releases_in_different_namespaces( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """Two tenants may both call their instance `redis`. Namespace is part of the identity.""" _fake_api(monkeypatch, tmp_path, [_meta("redis", "tenant-a", "1"), _meta("redis", "tenant-b", "1")]) out = await HelmProvisioner().list_releases() assert {(r.name, r.namespace) for r in out} == {("redis", "tenant-a"), ("redis", "tenant-b")} @pytest.mark.asyncio async def test_api_read_asks_for_metadata_only_and_scopes_the_selector( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """The Accept header is the difference between metadata and megabytes of gzipped payload.""" cap = _fake_api(monkeypatch, tmp_path, []) await HelmProvisioner().list_releases() assert "PartialObjectMetadataList" in cap["headers"]["accept"] selector = cap["params"]["labelSelector"] assert "owner=helm" in selector assert f"{MANAGED_BY_LABEL}={MANAGED_BY_VALUE}" in selector assert "superseded" not in selector # dropped server-side by asking only for live states assert cap["url"].endswith("/api/v1/secrets") @pytest.mark.asyncio async def test_api_read_ignores_secrets_that_are_not_releases( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """A malformed or unrelated secret must not become a phantom release.""" _fake_api( monkeypatch, tmp_path, [{"metadata": {"namespace": "x", "labels": {}}}, _meta("real", "tenant-a", "1")], ) out = await HelmProvisioner().list_releases() assert [r.name for r in out] == ["real"] @pytest.mark.asyncio async def test_no_service_account_falls_back_to_helm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """Out of cluster there is nothing to authenticate with, so e2e and laptops keep working.""" seen = _capture(monkeypatch) monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "absent") await HelmProvisioner(kubeconfig=Path("/dev/null")).list_releases() assert seen and seen[0][:2] == ["helm", "list"] @pytest.mark.asyncio async def test_api_read_handles_kubernetes_serialising_empty_as_null( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """`"items": null`, not `[]`, is what an empty list looks like on the wire. The key is present, so `.get("items", [])` returns None and the default never fires. This shipped and failed in production on the first tick that matched no releases: TypeError: 'NoneType' object is not iterable. The earlier tests all passed a real empty list, which is the one shape that cannot catch it. """ cap = _fake_api(monkeypatch, tmp_path, []) cap["force_null_items"] = True class _NullResponse: def raise_for_status(self) -> None: return None def json(self) -> dict[str, object]: return {"apiVersion": "meta.k8s.io/v1", "kind": "PartialObjectMetadataList", "items": None} class _NullClient: def __init__(self, **kw: object) -> None: pass async def __aenter__(self) -> _NullClient: return self async def __aexit__(self, *exc: object) -> None: return None async def get(self, url: str, **kw: object) -> _NullResponse: return _NullResponse() monkeypatch.setattr(httpx, "AsyncClient", _NullClient) assert await HelmProvisioner().list_releases() == [] @pytest.mark.asyncio async def test_api_read_passes_tls_verify_and_timeout( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """The API client must verify against the SA CA and carry the read timeout. A refactor that dropped `verify` to the default (or None) is a TLS regression the happy path would not reveal, so it is pinned here off the captured client kwargs. """ cap = _fake_api(monkeypatch, tmp_path, []) await HelmProvisioner().list_releases() assert cap["client_kwargs"]["verify"] == str(helm._SA_CA) assert cap["client_kwargs"]["timeout"] == helm._API_TIMEOUT_S @pytest.mark.asyncio async def test_api_read_sends_no_limit_param(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """No `limit`, so the apiserver returns the full set and the single read is complete. Pins the pagination invariant: adding `limit` without consuming `metadata.continue` would silently truncate the release list. """ cap = _fake_api(monkeypatch, tmp_path, []) await HelmProvisioner().list_releases() assert "limit" not in cap["params"] @pytest.mark.asyncio async def test_api_error_raises_helmerror_and_does_not_fall_back( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """A reachable-but-erroring apiserver raises HelmError; it must not shell out to helm. Falling back would swap a visible error for the 330s helm timeout this method exists to remove. The fallback is only for a ServiceAccount that is not present at all. """ (tmp_path / "ca.crt").write_text("ca", encoding="utf-8") monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "token") (tmp_path / "token").write_text("tok", encoding="utf-8") monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt") monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1") seen = _capture(monkeypatch) class _ErrClient: def __init__(self, **kw: object) -> None: pass async def __aenter__(self) -> _ErrClient: return self async def __aexit__(self, *exc: object) -> None: return None async def get(self, url: str, **kw: object) -> object: raise httpx.ConnectError("connection reset by peer") monkeypatch.setattr(httpx, "AsyncClient", _ErrClient) with pytest.raises(HelmError): await HelmProvisioner().list_releases() assert seen == [], "an API error must not fall back to `helm list`" @pytest.mark.asyncio async def test_missing_ca_falls_back_to_helm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """A token without a readable CA is a half-mounted SA: fall back rather than crash. httpx loads the CA when the client is built, raising OSError that the API except clause does not catch, so the readability check has to happen before the request. A missing CA means "not in-cluster", the same as a missing token. """ monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "token") (tmp_path / "token").write_text("tok", encoding="utf-8") monkeypatch.setattr(helm, "_SA_CA", tmp_path / "absent-ca.crt") # never created monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1") seen = _capture(monkeypatch) await HelmProvisioner(kubeconfig=Path("/dev/null")).list_releases() assert seen and seen[0][:2] == ["helm", "list"]