From d64c3c9f39df1a11f2441d6a1967b7f7c4800aaf Mon Sep 17 00:00:00 2001 From: Nguyen Minh Phuc Date: Tue, 21 Jul 2026 01:31:11 +0000 Subject: [PATCH] fix: three correctness bugs found in review + a dropped-log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fail() blanket-marked the instance `failed` for every dead-lettered task kind. Only provision is correct. The others each left the instance in a state the UPDATE then corrupted: - deprovision: `deleting` -> `failed` stranded the instance, because due_for_deprovision only re-selects ready/deleting, leaking the release reconcile.py promises to reclaim. - upgrade: helm --atomic rolled back, so the instance was still `ready` and serving the old version; `failed` mislabelled a healthy service and dropped it off the upgrade work-list. - verify: handle_verify already halted the rollout; the instance was `ready`. Now gated on kind == provision, with a regression test per kind (control-tested against the blanket UPDATE, which fails all three). The API exception handler was registered on fastapi.HTTPException, a subclass of starlette's. Starlette matches handlers by walking type(exc).__mro__, so framework-raised 404/405 never hit it and returned {"detail": ...} instead of ErrorBody. Registered on the starlette parent, and added a RequestValidationError handler so body-validation 422s share the shape too. Tests assert the ErrorBody shape for framework 404, 405, and a forbidden field. helm._list_releases_via_api built the httpx client with verify=, which loads the CA eagerly and raises OSError on a half-mounted ServiceAccount — an error the except clause did not catch, crashing the reconciler tick as a bare bug. Now the CA is checked for readability alongside the token, and a missing one means "not in-cluster" and falls back to helm. Also documents the no-limit pagination invariant and pins it with a test. redis.py logged through stdlib logging with extra={"team": team}, which the structlog bridge drops on the floor — the trap notify.py documents. Switched to a bound logger with team as a kwarg. --- .../svcforge_core/adapters/helm.py | 36 +++++--- .../svcforge_core/adapters/redis.py | 26 ++++-- .../svcforge_core/svcforge_core/repo/tasks.py | 30 ++++-- services/api/main.py | 27 +++++- tests/integration/test_api.py | 40 ++++++++ tests/integration/test_tasks.py | 80 ++++++++++++++++ tests/unit/test_helm_argv.py | 91 ++++++++++++++++++- 7 files changed, 299 insertions(+), 31 deletions(-) diff --git a/libs/svcforge_core/svcforge_core/adapters/helm.py b/libs/svcforge_core/svcforge_core/adapters/helm.py index 53f578e..abef23b 100644 --- a/libs/svcforge_core/svcforge_core/adapters/helm.py +++ b/libs/svcforge_core/svcforge_core/adapters/helm.py @@ -278,12 +278,12 @@ class HelmProvisioner: sweep. They are not orphans. They were never svcforge's to know about. Cost is why this does not shell out to helm when it does not have to. `--selector` is - NOT pushed down as a server-side selector — helm fetches and decompresses every - release secret in the cluster regardless, then filters what it already parsed. - Measured: unscoped 23 releases in 4392ms, scoped to 0 in 3988ms, a saving of about - 10%, not the order of magnitude the flag's shape suggests. + applied by helm after it has already fetched and decompressed every release secret in + the cluster, so it saves almost nothing: measured unscoped 23 releases in 4392ms, + scoped to 0 in 3988ms, about 10%. The flag's shape suggests a server-side filter; it + is a client-side one. - That was not academic. With the CPU request mutated to 0 by a cluster policy, the + The cost was real. With the CPU request mutated to 0 by a cluster policy, the call took over 330s and timed out on every tick, against ~4s given real CPU. A check that never completes reports no drift, which looks exactly like no drift existing. @@ -347,19 +347,27 @@ class HelmProvisioner: treating it as missing would have the reconciler re-provision on top of it. * helm writes one secret per revision, so a release can still appear more than once — 20 releases here had up to 10 revisions each. The newest `version` label wins. - Skipping this would report a release that exists as several, which for the - reconciler's set difference is harmless, and for anything counting releases is not. + Skipping this would report one release as several; the reconciler's set difference + tolerates that, but any caller that counts releases would over-count. `chart`, `status`, `revision` and `app_version` on the returned ReleaseInfo are the - subset the labels give away free. `chart` in particular is empty rather than wrong, - because the chart name lives only in the compressed payload. Narrowing the model - instead would have been honest about this method and dishonest about the helm path, - which does populate them. + subset the labels give away free. `chart` is empty here because the chart name lives + only in the compressed payload. The model keeps those fields so the helm fallback + path, which does populate them from `helm list`, returns the same shape. """ try: token = _SA_TOKEN.read_text(encoding="utf-8").strip() except OSError: return None + # The CA has to be readable too, and it is checked here rather than left to httpx. + # httpx loads the CA eagerly when the client is built, and that load raises OSError, + # which is not in the (httpx.HTTPError, json.JSONDecodeError) except below. A + # half-mounted ServiceAccount — token present, ca.crt absent or late — would then + # crash the tick as a bare bug instead of falling back. A complete ServiceAccount is + # the real in-cluster signal, so a missing CA means "not in-cluster" like a missing + # token does. + if not token or not os.access(_SA_CA, os.R_OK): + return None host, port = ( os.environ.get("KUBERNETES_SERVICE_HOST"), os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", "443"), @@ -370,6 +378,12 @@ class HelmProvisioner: selector = f"{_HELM_OWNER_LABEL},{MANAGED_BY_LABEL}={MANAGED_BY_VALUE},{_LIVE_STATUSES}" try: async with httpx.AsyncClient(verify=str(_SA_CA), timeout=_API_TIMEOUT_S) as client: + # No `limit` param, and that is load-bearing: the apiserver only returns a + # `metadata.continue` token when the client sets `limit`, so with none set it + # returns the full matching set in one response and the single read below is + # complete. Adding `limit` here without also looping on `continue` would + # silently truncate the list, and the reconciler would read the missing + # releases as orphans to delete or as vanished releases to re-provision. resp = await client.get( f"https://{host}:{port}/api/v1/secrets", params={"labelSelector": selector}, diff --git a/libs/svcforge_core/svcforge_core/adapters/redis.py b/libs/svcforge_core/svcforge_core/adapters/redis.py index 845ab40..d2a3d90 100644 --- a/libs/svcforge_core/svcforge_core/adapters/redis.py +++ b/libs/svcforge_core/svcforge_core/adapters/redis.py @@ -39,7 +39,6 @@ you cared about. from __future__ import annotations import asyncio -import logging import math from dataclasses import dataclass from datetime import UTC, datetime @@ -51,6 +50,7 @@ from pydantic import ValidationError from redis.asyncio import Redis from redis.exceptions import RedisError +from svcforge_core import obs from svcforge_core.adapters.clock import Clock, SystemClock from svcforge_core.domain.models import Instance @@ -59,7 +59,10 @@ if TYPE_CHECKING: from svcforge_core.settings import Settings -_log = logging.getLogger(__name__) +# structlog via obs, not stdlib logging. The stdlib bridge builds the event dict from the +# record message alone and drops `extra=` fields — the same trap notify.py documents. A bound +# logger takes fields as kwargs (team=team) and keeps them. Bound per instance in __init__, +# which runs after obs.setup() has configured structlog, never at import time. # --- The budget metric ------------------------------------------------------------------ # @@ -209,6 +212,7 @@ class RateLimiter: def __init__(self, r: Redis, limit: int, window_s: int, *, clock: Clock | None = None) -> None: """`clock` is injectable so the window boundary is testable without sleeping.""" + self._log = obs.get_logger(__name__) if limit < 1: raise ValueError("limit must be >= 1") if window_s < 1: @@ -244,10 +248,10 @@ class RateLimiter: allowed, remaining = await self._script(keys=[key], args=[self._limit, self._window_s]) except _REDIS_DOWN: REDIS_ERRORS.labels(op="ratelimit").inc() - _log.warning( + self._log.warning( "rate limiter degraded: redis unavailable, failing OPEN", exc_info=True, - extra={"team": team}, + team=team, ) return RateLimitResult( allowed=True, @@ -294,6 +298,7 @@ class IdempotencyStore: def __init__(self, r: Redis, ttl_s: int = 86400) -> None: """A day is the window a client might reasonably retry in; then the key is garbage.""" + self._log = obs.get_logger(__name__) if ttl_s < 1: raise ValueError("ttl_s must be >= 1") self._r = r @@ -320,7 +325,7 @@ class IdempotencyStore: existing = await self._r.get(redis_key) except _REDIS_DOWN: REDIS_ERRORS.labels(op="idempotency").inc() - _log.warning( + self._log.warning( "idempotency degraded: redis unavailable, falling through to the DB constraint", exc_info=True, ) @@ -333,7 +338,7 @@ class IdempotencyStore: try: return UUID(_as_text(existing)) except ValueError: - _log.warning("idempotency key holds a non-UUID value; ignoring it") + self._log.warning("idempotency key holds a non-UUID value; ignoring it") return None @@ -371,6 +376,7 @@ class InstanceCache: """ def __init__(self, r: Redis, ttl_s: int = 30) -> None: + self._log = obs.get_logger(__name__) if ttl_s < 1: raise ValueError("ttl_s must be >= 1") self._r = r @@ -391,7 +397,7 @@ class InstanceCache: raw = await self._r.get(self._key(instance_id)) except _REDIS_DOWN: REDIS_ERRORS.labels(op="cache_get").inc() - _log.warning("cache read degraded: redis unavailable, falling through to Postgres") + self._log.warning("cache read degraded: redis unavailable, falling through to Postgres") return None if raw is None: return None @@ -400,7 +406,7 @@ class InstanceCache: except ValidationError: # A model change deployed over a warm cache. Treat it as a miss and let the TTL # take the old shape out. Not an error: the truth is in Postgres either way. - _log.info("cache entry failed validation; treating as a miss") + self._log.info("cache entry failed validation; treating as a miss") return None async def put(self, inst: Instance) -> None: @@ -410,7 +416,7 @@ class InstanceCache: await self._r.set(self._key(inst.id), inst.model_dump_json(), ex=self._ttl_s) except _REDIS_DOWN: REDIS_ERRORS.labels(op="cache_put").inc() - _log.warning("cache write degraded: redis unavailable") + self._log.warning("cache write degraded: redis unavailable") async def invalidate(self, instance_id: UUID) -> None: """One DEL. Called by the worker inside the code path that writes the state. @@ -423,4 +429,4 @@ class InstanceCache: await self._r.delete(self._key(instance_id)) except _REDIS_DOWN: REDIS_ERRORS.labels(op="cache_del").inc() - _log.warning("cache invalidate degraded: redis unavailable; entry expires within the TTL") + self._log.warning("cache invalidate degraded: redis unavailable; entry expires within the TTL") diff --git a/libs/svcforge_core/svcforge_core/repo/tasks.py b/libs/svcforge_core/svcforge_core/repo/tasks.py index f3f830a..4984034 100644 --- a/libs/svcforge_core/svcforge_core/repo/tasks.py +++ b/libs/svcforge_core/svcforge_core/repo/tasks.py @@ -215,14 +215,28 @@ class TaskRepo: where id = %s""", (err[-2000:], task_id), ) - # `state = any(%s)` keeps this honest: a deprovision that exhausts its - # retries against an already-deleted instance records nothing rather than - # resurrecting it into `failed`. - await cur.execute( - """update instances set error=%s, state=%s, updated_at=now() - where id=%s and state = any(%s)""", - (err[-2000:], InstanceState.FAILED.value, instance_id, list(_CAN_FAIL)), - ) + # Dead-lettering the task is correct for every kind. Moving the INSTANCE to + # `failed` is correct only for provision: a provisioning instance that never + # came up is failed, and nothing recovers it but a human. The other kinds + # must leave the instance where it is, because for each of them the instance + # is still healthy and something else is responsible for recovery: + # deprovision — still `deleting`, which is exactly what lets + # due_for_deprovision re-enqueue it on the next sweep. `failed` + # drops it out of that query and leaks the release forever. + # upgrade — helm --atomic rolled back, so it is still `ready` and + # serving the previous version. check_version_drift retries on + # the next window; `failed` would mislabel a working service + # and drop it off the upgrade work-list. + # verify — handle_verify already halted the rollout; the instance is + # `ready`, and drift re-provisions it if its release vanished. + # The dead-letter metric and its alert are the operator signal for all four, + # so leaving the instance alone loses no visibility. + if row["kind"] == TaskKind.PROVISION.value: + await cur.execute( + """update instances set error=%s, state=%s, updated_at=now() + where id=%s and state = any(%s)""", + (err[-2000:], InstanceState.FAILED.value, instance_id, list(_CAN_FAIL)), + ) # Counted here, not in the worker: this is the only place that knows the # difference between "attempt 2 of 5 failed" and "this task is done trying". TASKS_DEAD_LETTERED.labels(kind=str(row["kind"])).inc() diff --git a/services/api/main.py b/services/api/main.py index 5b9d089..a56767b 100644 --- a/services/api/main.py +++ b/services/api/main.py @@ -11,9 +11,11 @@ import asyncio from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from fastapi import FastAPI, HTTPException, Request +from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from jwt import PyJWKClient +from starlette.exceptions import HTTPException from services.api.models import ErrorBody from services.api.routes import health, instances @@ -89,6 +91,13 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo Handlers raise `detail={"code": ..., "message": ...}`; FastAPI's default would nest that under `{"detail": {...}}`. Plain-string details (raised by FastAPI itself, e.g. a 405) are wrapped so clients never have to branch on the body's type. + + Registered on starlette's HTTPException, not fastapi's. fastapi.HTTPException is a + subclass, and Starlette matches handlers by walking type(exc).__mro__, so a handler + keyed on the subclass never fires for a framework-raised 404 or 405 — which are + starlette.HTTPException instances. Keying on the parent catches both: app handlers + raise the FastAPI subclass with a dict detail, the framework raises the parent with a + str detail, and the branch below renders each into ErrorBody. """ assert isinstance(exc, HTTPException) # noqa: S101 - registered only for HTTPException # Widened to object deliberately. Starlette types `detail` as str, but FastAPI passes @@ -102,6 +111,21 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo return JSONResponse(status_code=exc.status_code, content=body.model_dump(), headers=exc.headers) +async def _validation_exception_handler(request: Request, exc: Exception) -> JSONResponse: + """Render request-validation failures as ErrorBody too. + + A body that fails validation (a forbidden extra field, a bad type, an out-of-range + ttl_days) raises RequestValidationError, which the HTTPException handler above never + sees. Without this it returns FastAPI's default `{"detail": [...]}` — a second 422 shape + alongside the ErrorBody 422s the handlers raise. This gives every 422 one shape. + """ + assert isinstance(exc, RequestValidationError) # noqa: S101 - registered only for this + return JSONResponse( + status_code=422, + content=ErrorBody(code="validation_error", message=str(exc.errors())).model_dump(), + ) + + def create_app(settings: Settings | None = None) -> FastAPI: """App factory: lifespan, routers, exception handler, /metrics.""" settings = settings or load_settings() @@ -132,6 +156,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: app.include_router(instances.router) app.add_exception_handler(HTTPException, _http_exception_handler) + app.add_exception_handler(RequestValidationError, _validation_exception_handler) return app diff --git a/tests/integration/test_api.py b/tests/integration/test_api.py index d476382..ccfb6fc 100644 --- a/tests/integration/test_api.py +++ b/tests/integration/test_api.py @@ -298,6 +298,46 @@ async def test_ttl_out_of_range_is_422(client: httpx.AsyncClient, token: str) -> assert resp.status_code == 422 +def _is_error_body(payload: object) -> bool: + """The uniform error shape: a dict with `code` and `message`, and no default `detail`.""" + return isinstance(payload, dict) and "code" in payload and "message" in payload + + +async def test_framework_404_uses_the_error_body_shape(client: httpx.AsyncClient) -> None: + """A 404 raised by the router, not a handler, must still be ErrorBody. + + Starlette raises its own HTTPException for an unknown route. The exception handler is + registered on that parent class precisely so this body is ErrorBody and not FastAPI's + default `{"detail": "Not Found"}` — one shape for every error. + """ + resp = await client.get("/v1/no-such-route") + assert resp.status_code == 404 + assert _is_error_body(resp.json()), resp.text + + +async def test_framework_405_uses_the_error_body_shape(client: httpx.AsyncClient) -> None: + """A wrong-method 405 comes from the router too, and must be ErrorBody.""" + resp = await client.delete("/v1/instances") # collection route has no DELETE + assert resp.status_code == 405 + assert _is_error_body(resp.json()), resp.text + + +async def test_body_validation_422_uses_the_error_body_shape(client: httpx.AsyncClient, token: str) -> None: + """A RequestValidationError 422 must match the handler-raised 422 shape. + + A forbidden extra field trips pydantic's `extra="forbid"` and raises + RequestValidationError, which the dedicated handler renders as ErrorBody rather than the + default `{"detail": [...]}`. + """ + resp = await client.post( + "/v1/instances", + headers=auth(token), + json={"service_type": "redis", "size": "small", "surprise": "field"}, + ) + assert resp.status_code == 422 + assert _is_error_body(resp.json()), resp.text + + # --------------------------------------------------------------------------- authn diff --git a/tests/integration/test_tasks.py b/tests/integration/test_tasks.py index af3db27..184a622 100644 --- a/tests/integration/test_tasks.py +++ b/tests/integration/test_tasks.py @@ -110,6 +110,86 @@ async def test_fail_does_not_resurrect_a_deleted_instance(pool: DictPool) -> Non assert inst.error is None +async def test_fail_of_deprovision_leaves_the_instance_deleting_to_be_retried( + pool: DictPool, +) -> None: + """A dead-lettered deprovision must not strand the instance in `failed`. + + The instance is `deleting`, which is one of the states that CAN legally become `failed`, + so the naive blanket UPDATE would move it there. due_for_deprovision only re-selects + `ready`(expired) and `deleting`, so `failed` would take the instance out of the recovery + sweep and leak the helm release forever. reconcile.py documents that a deprovision which + exhausts its retries stays re-enqueueable; this pins that guarantee. + """ + iid = await make_instance(pool, state=InstanceState.DELETING) + tasks, instances = TaskRepo(pool), InstanceRepo(pool) + tid = await tasks.enqueue_standalone(iid, TaskKind.DEPROVISION) + + claimed = await tasks.claim("w1") + assert claimed is not None + async with pool.connection() as conn, conn.cursor() as cur: + await cur.execute("update tasks set attempts = 5 where id = %s", (tid,)) + + assert await tasks.fail(tid, "cluster unreachable", "w1", max_attempts=5) is True + + assert (await _task_row(pool, tid))["state"] == "failed" + inst = await instances.get(iid, team="platform") + assert inst is not None + assert inst.state is InstanceState.DELETING, "a stranded deprovision leaks the release" + assert inst.error is None + + +async def test_fail_of_upgrade_leaves_a_working_instance_ready(pool: DictPool) -> None: + """A dead-lettered upgrade must not mark a healthy instance `failed`. + + helm --atomic rolls the release back, so after a failed upgrade the instance is still + `ready` and serving the previous version. Marking it `failed` mislabels a working + service and drops it off the upgrade work-list. check_version_drift retries on the next + window; the dead-letter metric is the operator signal. + """ + iid = await make_instance(pool, state=InstanceState.READY) + tasks, instances = TaskRepo(pool), InstanceRepo(pool) + tid = await tasks.enqueue_standalone(iid, TaskKind.UPGRADE) + + claimed = await tasks.claim("w1") + assert claimed is not None + async with pool.connection() as conn, conn.cursor() as cur: + await cur.execute("update tasks set attempts = 5 where id = %s", (tid,)) + + assert await tasks.fail(tid, "upgrade to 1.4.0 kept timing out", "w1", max_attempts=5) is True + + assert (await _task_row(pool, tid))["state"] == "failed" + inst = await instances.get(iid, team="platform") + assert inst is not None + assert inst.state is InstanceState.READY, "a failed upgrade mislabelled a healthy instance" + assert inst.error is None + + +async def test_fail_of_verify_leaves_the_instance_ready(pool: DictPool) -> None: + """A dead-lettered verify must not mark the instance `failed`. + + handle_verify halts the rollout for the service type; the instance itself is `ready`, + and drift re-provisions it if its release vanished. `failed` would take it out of both + recovery paths. + """ + iid = await make_instance(pool, state=InstanceState.READY) + tasks, instances = TaskRepo(pool), InstanceRepo(pool) + tid = await tasks.enqueue_standalone(iid, TaskKind.VERIFY) + + claimed = await tasks.claim("w1") + assert claimed is not None + async with pool.connection() as conn, conn.cursor() as cur: + await cur.execute("update tasks set attempts = 5 where id = %s", (tid,)) + + assert await tasks.fail(tid, "release vanished after upgrade", "w1", max_attempts=5) is True + + assert (await _task_row(pool, tid))["state"] == "failed" + inst = await instances.get(iid, team="platform") + assert inst is not None + assert inst.state is InstanceState.READY + assert inst.error is None + + async def test_fail_truncates_error_to_2kb(pool: DictPool) -> None: iid = await make_instance(pool) repo = TaskRepo(pool) diff --git a/tests/unit/test_helm_argv.py b/tests/unit/test_helm_argv.py index be6251b..c74ac0b 100644 --- a/tests/unit/test_helm_argv.py +++ b/tests/unit/test_helm_argv.py @@ -21,6 +21,7 @@ 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 @@ -131,8 +132,10 @@ def _fake_api( 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", tmp_path / "ca.crt") + monkeypatch.setattr(helm, "_SA_CA", ca) monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1") monkeypatch.setenv("KUBERNETES_SERVICE_PORT_HTTPS", "443") @@ -267,3 +270,89 @@ async def test_api_read_handles_kubernetes_serialising_empty_as_null( 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"]