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,
    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")
    monkeypatch.setattr(helm, "_SA_TOKEN", token)
    monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt")
    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"]