feat: live AKS contract test environment + fix silently swallowed HTTP errors (#109)

.github/workflows/k8s-contract.yml

@@ -0,0 +1,180 @@
+name: K8s contract tests
+# Live-cluster contract suite (issue #109). Runs the destructive
+# write-path contract tests against the dedicated, stopped-at-rest
+# AKS cluster `aks-korvid-contract-test`. Never runs for pull requests
+# or forks: the Azure identity is scoped to the protected
+# `aks-contract-test` environment, which only `main` may use.
+
+on:
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: korvid-k8s-contract
+  cancel-in-progress: false
+
+env:
+  # The one cluster this workflow is allowed to touch. The guard steps
+  # fail closed when the (mutable) environment variables drift from
+  # these fixed, documented test-only names.
+  EXPECTED_RESOURCE_GROUP: rg-korvid-contract-test
+  EXPECTED_CLUSTER_NAME: aks-korvid-contract-test
+
+jobs:
+  contract:
+    # GitHub-hosted ephemeral runner — deliberately NOT the shared
+    # self-hosted `korvid-runners` pool, so cluster credentials never
+    # touch persistent shared infrastructure.
+    runs-on: ubuntu-latest
+    environment: aks-contract-test
+    timeout-minutes: 45
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      AKS_RESOURCE_GROUP: ${{ vars.AKS_RESOURCE_GROUP }}
+      AKS_CLUSTER_NAME: ${{ vars.AKS_CLUSTER_NAME }}
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+        with:
+          persist-credentials: false
+
+      - name: Guard — refuse any target other than the documented test cluster
+        run: |
+          if [ "$AKS_RESOURCE_GROUP" != "$EXPECTED_RESOURCE_GROUP" ] || \
+             [ "$AKS_CLUSTER_NAME" != "$EXPECTED_CLUSTER_NAME" ]; then
+            echo "::error title=Cluster identity mismatch::environment variables point at $AKS_RESOURCE_GROUP/$AKS_CLUSTER_NAME, expected $EXPECTED_RESOURCE_GROUP/$EXPECTED_CLUSTER_NAME. Refusing to run destructive tests."
+            exit 1
+          fi
+
+      - uses: azure/login@532459ea530d8321f2fb9bb10d1e0bcf23869a43 # v3.0.0
+        with:
+          client-id: ${{ vars.AZURE_CLIENT_ID }}
+          tenant-id: ${{ vars.AZURE_TENANT_ID }}
+          subscription-id: ${{ vars.AZURE_SUBSCRIPTION_ID }}
+
+      - name: Guard — verify test-only tags before touching the cluster
+        run: |
+          purpose=$(az aks show -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME" \
+            --query 'tags.purpose' -o tsv)
+          production=$(az aks show -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME" \
+            --query 'tags."production-use"' -o tsv)
+          if [ "$purpose" != "korvid-contract-testing" ] || [ "$production" != "prohibited" ]; then
+            echo "::error title=Not a test-only cluster::$AKS_CLUSTER_NAME is missing the purpose=korvid-contract-testing / production-use=prohibited tags. Refusing to run destructive tests."
+            exit 1
+          fi
+
+      - name: Start AKS cluster
+        run: |
+          state=$(az aks show -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME" --query powerState.code -o tsv)
+          if [ "$state" != "Running" ]; then
+            az aks start -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME"
+          fi
+
+      - uses: azure/use-kubelogin@0ce7c36141aa27d4934872cf00b0120804c98a29 # v1.3
+        with:
+          kubelogin-version: v0.2.15
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Fetch user credentials (no --admin, isolated kubeconfig)
+        run: |
+          export KUBECONFIG="$RUNNER_TEMP/contract-kubeconfig"
+          az aks get-credentials -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME" \
+            --overwrite-existing -f "$KUBECONFIG"
+          kubelogin convert-kubeconfig -l azurecli --kubeconfig "$KUBECONFIG"
+          chmod 600 "$KUBECONFIG"
+          echo "KUBECONFIG=$KUBECONFIG" >> "$GITHUB_ENV"
+
+      - uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        with:
+          python-version: "3.13"
+          enable-cache: true
+
+      - run: uv sync --locked --dev --all-extras
+
+      - name: Install helm
+        run: |
+          HELM_VERSION=v3.19.0
+          curl -fsSL "https://get.helm.sh/helm-${HELM_VERSION}-linux-amd64.tar.gz" \
+            | tar -xz -C "$RUNNER_TEMP" linux-amd64/helm
+          sudo install "$RUNNER_TEMP/linux-amd64/helm" /usr/local/bin/helm
+
+      - name: Janitor — remove stale fixtures from interrupted runs
+        run: uv run python -m tests.contract.janitor
+
+      - name: Run contract suite
+        timeout-minutes: 30
+        env:
+          KORVID_CONTRACT_RUN_ID: ${{ github.run_id }}-${{ github.run_attempt }}
+        run: uv run pytest -p no:randomly -m contract tests/contract/ -v
+
+      - name: Collect diagnostics
+        if: failure() || cancelled()
+        run: |
+          mkdir -p diag
+          kubectl get events -A --sort-by=.lastTimestamp > diag/events.txt || true
+          kubectl get pods -A -o wide > diag/pods.txt || true
+          kubectl describe pods -A > diag/pods-describe.txt || true
+          kubectl get nodes -o wide > diag/nodes.txt || true
+          kubectl get ns -l app.kubernetes.io/managed-by=korvid-contract -o yaml > diag/namespaces.yaml || true
+
+      - name: Upload diagnostics
+        if: failure() || cancelled()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: contract-diagnostics
+          path: diag/
+          retention-days: 14
+
+      - name: Delete kubeconfig
+        if: always()
+        run: |
+          if [ -n "${KUBECONFIG:-}" ] && [ -f "$KUBECONFIG" ]; then
+            rm -f "$KUBECONFIG"
+          fi
+
+  # Separate job so the stop runs even when the contract job hits its
+  # timeout-minutes (GitHub kills the job before later always() steps,
+  # which would leave the stopped-at-rest cluster running and billing).
+  stop-cluster:
+    needs: contract
+    if: always()
+    runs-on: ubuntu-latest
+    environment: aks-contract-test
+    timeout-minutes: 20
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      AKS_RESOURCE_GROUP: ${{ vars.AKS_RESOURCE_GROUP }}
+      AKS_CLUSTER_NAME: ${{ vars.AKS_CLUSTER_NAME }}
+    steps:
+      - name: Guard — refuse any target other than the documented test cluster
+        run: |
+          if [ "$AKS_RESOURCE_GROUP" != "$EXPECTED_RESOURCE_GROUP" ] || \
+             [ "$AKS_CLUSTER_NAME" != "$EXPECTED_CLUSTER_NAME" ]; then
+            echo "::error title=Cluster identity mismatch::environment variables point at $AKS_RESOURCE_GROUP/$AKS_CLUSTER_NAME, expected $EXPECTED_RESOURCE_GROUP/$EXPECTED_CLUSTER_NAME. Refusing to stop an unexpected cluster."
+            exit 1
+          fi
+
+      - uses: azure/login@532459ea530d8321f2fb9bb10d1e0bcf23869a43 # v3.0.0
+        with:
+          client-id: ${{ vars.AZURE_CLIENT_ID }}
+          tenant-id: ${{ vars.AZURE_TENANT_ID }}
+          subscription-id: ${{ vars.AZURE_SUBSCRIPTION_ID }}
+
+      - name: Stop AKS cluster
+        run: |
+          az aks stop -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME" --no-wait
+          for i in $(seq 1 30); do
+            state=$(az aks show -g "$AKS_RESOURCE_GROUP" -n "$AKS_CLUSTER_NAME" --query powerState.code -o tsv || echo unknown)
+            [ "$state" = "Stopped" ] && exit 0
+            sleep 20
+          done
+          echo "::error title=AKS stop failed::aks-korvid-contract-test did not reach Stopped state; idle compute may be accruing. Stop it manually: az aks stop -g $AKS_RESOURCE_GROUP -n $AKS_CLUSTER_NAME"
+          exit 1

docs/dev/README.md

@@ -10,6 +10,8 @@ opposed to the user-facing feature docs that live directly under
 - [`plans/`](plans/) — dated implementation plans for individual phases
   and slices. Historical once executed; kept for traceability, not
   updated retroactively.
+- [`contract-tests.md`](contract-tests.md) — the live-cluster contract
+  suite and the Korvid test-only AKS infrastructure it runs against.
 
 If you are looking for how to *use* korvid, start at the
 [project README](../../README.md).

docs/dev/contract-tests.md

@@ -0,0 +1,87 @@
+# Live-cluster contract tests (issue #109)
+
+`tests/contract/` is a live-cluster contract suite that proves korvid's k8s
+layer against a real Kubernetes API server — dry-run previews cause no
+persistent mutation, execute paths mutate exactly once, and RBAC behaves as
+the permission probes assume. It exists because unit tests with faked
+transports cannot catch bugs like #103 (a preview reaching the real mutation
+path) or a client that silently swallows HTTP errors.
+
+## How it runs
+
+- **Fast PR gate is unchanged.** The suite is gated on the
+  `KORVID_CONTRACT_RUN_ID` environment variable: without it every
+  `contract`-marked test skips at collection time.
+- **Post-merge gate.** `.github/workflows/k8s-contract.yml` runs on pushes to
+  `main` (and protected manual dispatch) inside the `aks-contract-test`
+  GitHub environment, which only `main` may use. PR and fork workflows can
+  never obtain the Azure identity.
+- Run ID `${run_id}-${run_attempt}` labels every fixture
+  (`app.kubernetes.io/managed-by=korvid-contract`,
+  `korvid.dev/contract-run=<id>`) so cleanup is idempotent; a janitor
+  (`python -m tests.contract.janitor`) sweeps leftovers from interrupted runs
+  and uncordons any node a crashed run left unschedulable.
+
+## Test-only infrastructure
+
+Everything below is **Korvid test-only infrastructure** — no production
+workload runs there, and every resource is disposable:
+
+| Resource | Name |
+|---|---|
+| Resource group | `rg-korvid-contract-test` (koreacentral) |
+| AKS cluster | `aks-korvid-contract-test` (stopped at rest; started/stopped by the workflow; NAP disabled) |
+| Managed identity | `id-korvid-contract-test` (OIDC-federated to the `aks-contract-test` environment; no stored kubeconfig, secret, or admin cert) |
+
+Node pools:
+
+- `system` — tainted `CriticalAddonsOnly=true:NoSchedule`, labelled
+  `korvid.dev/pool=system`. Never targeted by tests.
+- `workload` — labelled `korvid.dev/pool=workload` and
+  `korvid.dev/disposable=true`. The **only** pool test pods schedule onto and
+  the only node the node-operation tests (cordon/drain/evict) will touch;
+  tests fail rather than fall back to a system node.
+
+The cluster has AAD + Azure RBAC enabled with local accounts disabled; the
+workflow identity holds a minimal custom role (read/start/stop/list-user-
+credentials) plus AKS RBAC Cluster Admin, both scoped to the one cluster.
+The provisioning Bicep and deployment record are attached to issue #109 —
+the infrastructure code deliberately lives outside this repository.
+
+## Running locally
+
+You need data-plane access to the cluster (AKS RBAC role on it),
+`kubelogin`, and Helm **3.13+** (`dry_run_install`/`dry_run_upgrade` pass
+`--hide-secret`, and `test_helm_release_lifecycle` skips silently when no
+`helm` binary is on `PATH`):
+
+```bash
+az aks start -g rg-korvid-contract-test -n aks-korvid-contract-test
+az aks get-credentials -g rg-korvid-contract-test -n aks-korvid-contract-test \
+  -f /tmp/contract-kubeconfig --overwrite-existing
+kubelogin convert-kubeconfig -l azurecli --kubeconfig /tmp/contract-kubeconfig
+
+export KUBECONFIG=/tmp/contract-kubeconfig
+uv run python -m tests.contract.janitor
+KORVID_CONTRACT_RUN_ID=local-$USER uv run pytest -p no:randomly -m contract tests/contract/
+
+az aks stop -g rg-korvid-contract-test -n aks-korvid-contract-test  # always
+```
+
+## Writing contract tests
+
+The contract pattern every test follows:
+
+1. **Preview** (`preview_*`, `dry_run_*`): call it, then **read state back
+   from the API server** and prove nothing persisted (uid, resourceVersion,
+   generation unchanged; no `deletionTimestamp`).
+2. **Execute**: call the real write, read back, and prove the mutation
+   happened **exactly once** (one generation bump, 404/409 on replay).
+3. Previews answer `None` on any failure by design (they never block the
+   approval flow), and they are pinned to a GET snapshot's resourceVersion —
+   retry via `conftest.preview_until_settled` when a controller may be
+   bumping the object concurrently.
+4. RBAC scenarios mint short-lived TokenRequest tokens for throwaway
+   ServiceAccounts; never legacy token Secrets.
+5. Label every created object with `conftest.run_labels()` and create it in
+   the per-test `namespace` fixture so teardown and the janitor can find it.

pyproject.toml

@@ -95,6 +95,9 @@ asyncio_mode = "auto"
 addopts = "--strict-markers --tb=short"
 testpaths = ["tests"]
 filterwarnings = ["error"]
+markers = [
+    "contract: live-cluster contract tests (opt-in via KORVID_CONTRACT_RUN_ID)",
+]
 
 [tool.coverage.run]
 branch = true

src/korvid/k8s/client.py

@@ -665,6 +665,7 @@ async def _request_write(
             # _preload_content=False the caller owns the response. Writes may
             # return empty or non-JSON bodies, so no decode is attempted here.
             raw: bytes = await resp.read()
+            _raise_for_status(resp, raw)
             return raw
         except k8s_client.exceptions.ApiException as exc:
             raise ApiStatusError(
@@ -1259,6 +1260,10 @@ async def stream_logs(
         except k8s_client.exceptions.ApiException as exc:
             raise ApiStatusError(int(exc.status or 0), str(exc.reason or "")) from exc
         try:
+            # Raw path never raises for HTTP errors (see _raise_for_status):
+            # check up front so an error Status body isn't streamed as logs.
+            if not 200 <= int(getattr(resp, "status", 0) or 0) <= 299:
+                _raise_for_status(resp, await resp.read())
             async for raw in resp.content:
                 if not raw:
                     continue
@@ -1339,13 +1344,21 @@ async def _watch_call(
             query_params: list[tuple[str, Any]] = [*extra_query, ("watch", "true")]
             if resource_version is not None:
                 query_params.append(("resourceVersion", resource_version))
-            return await api.call_api(
+            resp = await api.call_api(
                 path,
                 "GET",
                 auth_settings=["BearerToken"],
                 query_params=query_params,
                 _preload_content=False,
             )
+            # Watch never inspects resp.status: a non-2xx response would be
+            # retried forever (empty body) or surfaced as malformed events.
+            if not 200 <= int(getattr(resp, "status", 0) or 0) <= 299:
+                try:
+                    _raise_for_status(resp, await resp.read())
+                finally:
+                    resp.close()
+            return resp
 
         return _watch_call
 
@@ -1364,6 +1377,7 @@ async def _request_json(
                 _preload_content=False,
             )
             body = await resp.read()
+            _raise_for_status(resp, body)
             result: dict[str, Any] = json.loads(body)
             return result
         except k8s_client.exceptions.ApiException as exc:
@@ -1404,14 +1418,36 @@ async def close(self) -> None:
 
 
 async def _to_dict(resp: Any) -> dict[str, Any]:
-    """Normalize aiohttp response or dict into a plain dict."""
+    """Normalize aiohttp response or dict into a plain dict.
+
+    Raises ApiStatusError on a non-2xx status: with ``_preload_content=False``
+    kubernetes_asyncio never raises for HTTP errors, so an unchecked error
+    Status body would otherwise be parsed as if it were the requested object.
+    """
     if isinstance(resp, dict):
         return resp
     body = await resp.read()
+    _raise_for_status(resp, body)
     result: dict[str, Any] = json.loads(body)
     return result
 
 
+def _raise_for_status(resp: Any, body: bytes) -> None:
+    """Raise ApiStatusError for a non-2xx raw (``_preload_content=False``)
+    response. kubernetes_asyncio's rest layer only raises ApiException when
+    it preloads the body, so raw-path callers must check the status
+    themselves — otherwise refused writes (409 uid precondition, 429 PDB
+    denial) silently look like successes (caught live by the contract
+    suite, issue #109)."""
+    status = int(getattr(resp, "status", 0) or 0)
+    if not 200 <= status <= 299:
+        raise ApiStatusError(
+            status,
+            str(getattr(resp, "reason", "") or ""),
+            body=body.decode("utf-8", errors="replace"),
+        )
+
+
 def _parse_resource_list(data: dict[str, Any], *, group: str, version: str) -> list[ResourceMeta]:
     out = []
     for r in data.get("resources", []):

tests/contract/__init__.py

@@ -0,0 +1,8 @@
+"""Live-cluster contract tests (issue #109).
+
+Every test in this package runs against the dedicated
+``aks-korvid-contract-test`` cluster and proves real API-server
+semantics: previews cause no persistent mutation, executes mutate
+exactly once. The suite is opt-in — it only runs when
+``KORVID_CONTRACT_RUN_ID`` is set (see ``conftest.py``).
+"""