svcforge/tests/unit/test_helm_argv.py

"""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"]