Upload 42 files

projects/k3s-test-app/.gitignore

@@ -0,0 +1,14 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+# agentic stuffs
+.agent/
+review.txt

projects/k3s-test-app/.python-version

@@ -0,0 +1 @@
+3.14

projects/k3s-test-app/Containerfile

@@ -0,0 +1,17 @@
+# Stage 1: build dependencies
+FROM python:3.14-slim AS builder
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+WORKDIR /app
+COPY pyproject.toml uv.lock README.md ./
+RUN uv sync --frozen --no-install-project --no-dev
+COPY src/ src/
+RUN uv sync --frozen --no-dev
+
+# Stage 2: runtime
+FROM python:3.14-slim
+COPY --from=builder /app /app
+WORKDIR /app
+ENV PATH="/app/.venv/bin:$PATH"
+EXPOSE 8000
+USER nobody
+CMD ["uvicorn", "k3s_test_app.main:app", "--host", "0.0.0.0", "--port", "8000", "--timeout-graceful-shutdown", "30"]

projects/k3s-test-app/DOCS/ARCHITECTURE.md

@@ -0,0 +1,52 @@
+# ARCHITECTURE.md
+
+## App Flow
+Client → k3s Ingress (Traefik) → Service → Pod(s) → FastAPI app → (ConfigMap/Secret/PVC/other Service)
+
+## Folder Structure
+```
+k3s-test-app/
+├── app/
+│   ├── main.py              # app entrypoint, router includes
+│   ├── routers/
+│   │   ├── health.py        # /healthz, /ready
+│   │   ├── config.py        # /config, /secret
+│   │   ├── stress.py        # /stress/cpu, /stress/mem
+│   │   ├── lifecycle.py     # /crash, /sleep, shutdown handler
+│   │   ├── identity.py      # /whoami
+│   │   ├── storage.py       # /storage
+│   │   ├── network.py       # /delay/{s}, /call-internal, /ws, /stream
+│   │   ├── upload.py        # /upload
+│   │   └── version.py       # /version
+│   ├── core/
+│   │   ├── config.py        # pydantic Settings (env vars)
+│   │   └── logging.py
+│   └── metrics.py           # prometheus_client instrumentation
+├── k8s/
+│   ├── deployment.yaml
+│   ├── service.yaml
+│   ├── ingress.yaml
+│   ├── configmap.yaml
+│   ├── secret.yaml
+│   ├── pvc.yaml
+│   └── hpa.yaml
+├── Dockerfile
+├── requirements.txt
+├── PRD.md
+├── ARCHITECTURE.md
+├── RULES.md
+└── PHASES.md
+```
+
+## Tech Stack
+- **Runtime**: Python 3.12, FastAPI, Uvicorn (with `--timeout-graceful-shutdown`)
+- **Metrics**: `prometheus-client` / `prometheus-fastapi-instrumentator`
+- **Config**: `pydantic-settings` (env var driven)
+- **Container**: multi-stage Dockerfile, non-root user
+- **Orchestration**: k3s, Traefik ingress (built-in), standard Deployment/Service/PVC/HPA manifests
+- **Load testing**: `hey` / `k6` (external, not part of app)
+
+## Data Flow Notes
+- `/storage` reads/writes to a path backed by a PVC (`local-path` provisioner default in k3s)
+- `/call-internal` hits another Service by in-cluster DNS name (`svc.namespace.svc.cluster.local`)
+- `/metrics` scraped by Prometheus (or manual `curl`) in Prometheus text format

projects/k3s-test-app/DOCS/PHASES.md

@@ -0,0 +1,40 @@
+# PHASES.md
+
+## Phase 1 — Core App + Health
+- Scaffold FastAPI app, Dockerfile, requirements.txt
+- `/healthz`, `/ready`, `/version`
+- Deployment + Service manifests, deploy to k3s
+
+## Phase 2 — Config & Secrets
+- `/config` reads ConfigMap-injected env vars
+- `/secret` reads mounted Secret
+- ConfigMap + Secret manifests
+
+## Phase 3 — Identity & Networking
+- `/whoami` (hostname, pod IP)
+- `/delay/{seconds}`
+- `/call-internal` (in-cluster service-to-service call)
+- Ingress manifest + Traefik routing test
+
+## Phase 4 — Storage
+- `/storage` read/write endpoint
+- PVC manifest, verify persistence across pod restarts
+
+## Phase 5 — Load & Resilience
+- `/stress/cpu`, `/stress/mem`
+- `/crash`, `/sleep`
+- Resource limits + HPA manifest
+- Graceful SIGTERM shutdown handler
+
+## Phase 6 — Realtime & Uploads
+- `/ws` or `/stream` (SSE)
+- `/upload`
+- Ingress body-size/timeout tuning
+
+## Phase 7 — Observability
+- `/metrics` (Prometheus format)
+- Optional: Prometheus + Grafana deploy for scraping/dashboards
+
+## Phase 8 — Rolling Update Validation
+- `/version` bump workflow
+- Rolling update + rollback test (`kubectl rollout`)

projects/k3s-test-app/DOCS/PRD.md

@@ -0,0 +1,31 @@
+# PRD.md
+
+## Project
+k3s-test-app — FastAPI service for exercising k3s cluster behaviors (probes, config, scaling, storage, networking).
+
+## What to Build
+A single FastAPI app exposing test endpoints (health, config/secret readers, stress, crash, storage, delay, metrics, websocket, upload, graceful shutdown) to validate k3s deployment primitives end-to-end.
+
+## Targeted User
+Solo dev / infra engineer learning or validating a k3s cluster setup. Not a production consumer-facing app.
+
+## Features
+- [ ] Liveness/readiness probes (`/healthz`, `/ready`)
+- [ ] Config/secret introspection (`/config`, `/secret`)
+- [ ] Load/stress simulation (`/stress/cpu`, `/stress/mem`)
+- [ ] Controlled crash/sleep (`/crash`, `/sleep`)
+- [ ] Pod identity (`/whoami` — hostname, pod IP)
+- [ ] Persistent volume read/write (`/storage`)
+- [ ] Configurable delay (`/delay/{seconds}`)
+- [ ] Build/version info (`/version`)
+- [ ] Internal service call (`/call-internal`)
+- [ ] WebSocket/SSE (`/ws`, `/stream`)
+- [ ] File upload (`/upload`)
+- [ ] Prometheus metrics (`/metrics`)
+- [ ] Graceful SIGTERM shutdown
+
+## Out of Scope
+Auth, persistence beyond test PVC, multi-tenant concerns, UI.
+
+## Success Criteria
+Each endpoint proves one specific k3s behavior (see list above) reproducibly via `kubectl` observation.

projects/k3s-test-app/DOCS/RULES.md

@@ -0,0 +1,40 @@
+# RULES.md
+
+## Use
+- FastAPI + Uvicorn (async endpoints where I/O bound)
+- `pydantic-settings` for all config — no hardcoded values
+- `prometheus-client` for metrics
+- Structured logging (`logging` + JSON formatter), log to stdout only
+- Type hints on all functions
+- `httpx` for internal service calls (async)
+
+## Preferred Libraries
+| Purpose | Library |
+|---|---|
+| Web framework | fastapi |
+| ASGI server | uvicorn[standard] |
+| Config | pydantic-settings |
+| HTTP client | httpx |
+| Metrics | prometheus-fastapi-instrumentator |
+| Testing | pytest, pytest-asyncio |
+
+## Error Handling
+- No bare `except:` — always catch specific exceptions
+- Every endpoint returns proper HTTP status codes (4xx client, 5xx server)
+- Use FastAPI `HTTPException` for expected failures; let unexpected ones 500 + log traceback
+- All external calls (`/call-internal`, storage I/O) wrapped with timeout + explicit error response, never hang silently
+- Startup/shutdown events must not crash the app — log and degrade gracefully
+
+## AI Boundaries / Constraints
+- AI may scaffold code, write endpoints matching PRD.md features, write k8s manifests
+- AI must NOT introduce new libraries not listed in "Preferred Libraries" without explicit approval
+- AI must NOT change project structure defined in ARCHITECTURE.md without approval
+- AI must NOT add auth, persistence, or scope creep beyond PRD.md
+- AI must ask before deleting files or modifying k8s manifests already applied to a live cluster
+
+## Avoid
+- Unapproved libraries (Flask, Django, requests in async context, etc.)
+- Silent failures / swallowed exceptions
+- Global mutable state outside app lifespan
+- Hardcoded secrets/config (must come from env/ConfigMap/Secret)
+- AI making architectural or scope decisions outside PRD.md/ARCHITECTURE.md without flagging it

projects/k3s-test-app/DOCS/TESTING.md

@@ -0,0 +1,88 @@
+# TESTING.md
+
+## Overview
+The test suite lives in `tests/` and is organized by phase, matching the implementation in `PHASES.md`. Tests use `pytest` with `pytest-asyncio` and `httpx`'s `AsyncClient` pointed at the ASGI app — no real server is started.
+
+## Structure
+```
+tests/
+├── __init__.py
+├── conftest.py                      # Shared fixtures (async TestClient)
+├── test_phase1_health.py            # /healthz, /ready, /version
+├── test_phase2_config.py            # /config, /secret
+├── test_phase3_identity_network.py  # /whoami, /delay, /call-internal
+├── test_phase4_storage.py           # /storage
+├── test_phase5_resilience.py        # /stress/cpu, /stress/mem, /crash, /sleep
+├── test_phase6_realtime_upload.py   # /ws, /upload
+└── test_phase7_observability.py     # /metrics
+```
+
+## Running Tests
+
+### Install dev dependencies
+```bash
+uv sync --group dev
+```
+
+### Run all tests
+```bash
+uv run pytest
+```
+
+### Run a specific phase
+```bash
+uv run pytest tests/test_phase1_health.py
+uv run pytest tests/test_phase5_resilience.py
+```
+
+### Run with verbose output
+```bash
+uv run pytest -v
+```
+
+### Run with coverage (if you add pytest-cov later)
+```bash
+uv run pytest --cov=k3s_test_app --cov-report=term-missing
+```
+
+## Test Philosophy
+- Each test file covers one phase's endpoints.
+- External I/O (cluster DNS, secret files) is mocked using `unittest.mock.patch` and `monkeypatch` — tests run fully offline without a cluster.
+- Async endpoints are tested with `pytest-asyncio` in `asyncio_mode = auto` (configured in `pytest.ini`).
+- Settings are patched inline using `monkeypatch.setattr` on `cfg_module.settings` — no env file hacks needed.
+
+## Notable Patterns
+
+### Async test client (conftest.py)
+```python
+@pytest_asyncio.fixture
+async def client() -> AsyncClient:
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
+        yield ac
+```
+
+### Patching settings at runtime
+```python
+from k3s_test_app.core import config as cfg_module
+monkeypatch.setattr(cfg_module.settings, "ready", False)
+```
+
+### Mocking httpx calls for /call-internal
+```python
+with patch("k3s_test_app.routers.network.httpx.AsyncClient") as mock_http:
+    mock_http.return_value.__aenter__.return_value.get.side_effect = httpx.ConnectError("Name or service not known")
+    response = await client.get("/call-internal")
+    assert response.status_code == 502
+```
+
+### Mocking built-in loops safely (test_phase5_resilience.py)
+```python
+def mock_time_gen():
+    t = 0
+    while True:
+        yield t
+        t += 10
+
+with patch("k3s_test_app.routers.stress.time.time", side_effect=mock_time_gen()):
+    response = await client.get("/stress/cpu?seconds=5")
+```

projects/k3s-test-app/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Gaurav5189
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

projects/k3s-test-app/README.md

@@ -0,0 +1,111 @@
+# k3s-test-app 🚀
+
+`k3s-test-app` is a lightweight, purpose-built FastAPI application designed to validate and test various Kubernetes capabilities within a `k3s` environment. It provides a comprehensive suite of endpoints to simulate real-world behaviors—ranging from configuration injection and persistent storage to load-induced autoscaling and graceful degradation.
+
+## Features ✨
+
+This application is built in phases, exposing endpoints to test specific k8s features:
+
+- **Core & Probes**: `/healthz` and `/ready` for liveness and readiness probes.
+- **Configuration & Secrets**: `/config` and `/secret` to validate ConfigMap environment variable injection and Secret volume mounting.
+- **Identity & Networking**: `/whoami` to verify Downward API (`POD_IP`) injection, `/delay` for timeout testing, and `/call-internal` for cluster DNS resolution.
+- **Storage Persistence**: `GET /storage` and `POST /storage` to validate read/write operations against PVCs.
+- **Load & Resilience**: `/stress/cpu` and `/stress/mem` to trigger Horizontal Pod Autoscalers (HPA), and `/crash` / `/sleep` to test pod restart policies and event loop blocks.
+- **Realtime & Uploads**: `/ws` for long-lived WebSocket connections and `/upload` to test ingress body size limits.
+- **Observability**: `/metrics` exposes Prometheus-formatted application metrics automatically.
+
+## Tech Stack 🛠️
+
+- **Framework**: FastAPI + Uvicorn
+- **Configuration**: Pydantic Settings
+- **Testing**: Pytest, Pytest-Asyncio, HTTPX
+- **Containerization**: Podman + Containerfile (Multi-stage build)
+- **Deployment**: Kubernetes Manifests (Deployment, Service, Ingress, ConfigMap, Secret, PVC, HPA)
+
+---
+
+## Local Development & Testing 💻
+
+This project uses `uv` for fast Python dependency management.
+
+### Setup
+```bash
+# Sync dependencies (including dev dependencies)
+uv sync --group dev
+```
+
+### Running the App Locally
+```bash
+uv run uvicorn k3s_test_app.main:app --host 0.0.0.0 --port 8000 --reload
+```
+
+### Running Tests
+The project includes a full test suite built with `pytest` that mocks out infrastructure dependencies (filesystem, DNS, delays) to run instantly without a real cluster.
+```bash
+# Run the entire test suite
+uv run pytest -v
+
+# Run tests for a specific phase
+uv run pytest -v tests/test_phase5_resilience.py
+```
+> 📚 See [DOCS/TESTING.md](DOCS/TESTING.md) for more details on the testing architecture.
+
+---
+
+## Deployment to k3s 🚢
+
+All Kubernetes manifests are located in the `k8s/` directory.
+
+### 1. Build and Push the Image
+We use `podman` and a `Containerfile` to build the image.
+```bash
+podman build -t k3s-test-app:latest -f Containerfile .
+podman push docker.io/<username>/k3s-test-app:latest
+```
+*(Remember to update the `image:` reference in `k8s/deployment.yaml` if you push to a custom registry).*
+
+### 2. Ensure Metrics Server is Installed (HPA Requirement)
+The Horizontal Pod Autoscaler (HPA) requires the `metrics.k8s.io` API to scale based on CPU utilization.
+If your cluster does not have `metrics-server` installed (some k3s distributions include it by default, some don't), install it first:
+```bash
+kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/latest/download/components.yaml
+```
+
+### 3. Apply the Manifests
+```bash
+kubectl apply -f k8s/
+```
+This will deploy the application, configure Traefik Ingress (including Middlewares), mount secrets, provision local-path storage, and set up HPA rules.
+
+### 4. Access the Application
+The `Ingress` is configured to listen on `test.local`. You can test it locally by passing the `Host` header:
+```bash
+curl -H "Host: test.local" http://<k3s-node-ip>/healthz
+```
+
+---
+
+## API Reference 📖
+
+Once the application is running, you can view the auto-generated Swagger documentation at:
+- `http://<host>:8000/docs`
+
+### Key Endpoints:
+| Endpoint | Method | Purpose |
+|----------|--------|---------|
+| `/healthz` | GET | Liveness probe (Always returns 200) |
+| `/ready` | GET | Readiness probe (Fails if `READY=false`) |
+| `/version` | GET | Application version info |
+| `/config` | GET | Returns environment configuration |
+| `/secret` | GET | Reads from the mounted k8s Secret |
+| `/whoami` | GET | Returns Pod IP and hostname |
+| `/storage` | POST/GET | Writes/reads to the PVC |
+| `/stress/cpu` | GET | Spikes CPU utilization (tests HPA) |
+| `/metrics` | GET | Prometheus scraping endpoint |
+
+---
+
+## Documentation Links
+- [Architecture & Requirements](DOCS/ARCHITECTURE.md)
+- [Implementation Phases](DOCS/PHASES.md)
+- [Agent Memory/Changelog](.agent/MEMORY.md)

projects/k3s-test-app/k8s/configmap.yaml

@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: k3s-test-app-config
+data:
+  APP_NAME: "k3s-test-app"
+  LOG_LEVEL: "info"
+  CUSTOM_MESSAGE: "Hello from k3s ConfigMap!"

projects/k3s-test-app/k8s/deployment.yaml

@@ -0,0 +1,66 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: k3s-test-app
+  labels:
+    app: k3s-test-app
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: k3s-test-app
+  template:
+    metadata:
+      labels:
+        app: k3s-test-app
+    spec:
+      containers:
+        - name: k3s-test-app
+          image: gaurav0s/k3s-test-app:latest
+          imagePullPolicy: Always
+          ports:
+            - containerPort: 8000
+          livenessProbe:
+            httpGet:
+              path: /healthz
+              port: 8000
+            initialDelaySeconds: 10
+            periodSeconds: 15
+            timeoutSeconds: 3
+            failureThreshold: 3
+          readinessProbe:
+            httpGet:
+              path: /ready
+              port: 8000
+            initialDelaySeconds: 5
+            periodSeconds: 10
+            timeoutSeconds: 3
+            failureThreshold: 2
+          env:
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+          envFrom:
+            - configMapRef:
+                name: k3s-test-app-config
+          volumeMounts:
+            - name: test-secret-volume
+              mountPath: /etc/secrets/test-secret
+              readOnly: true
+            - name: storage-volume
+              mountPath: /data
+          resources:
+            requests:
+              cpu: 50m
+              memory: 64Mi
+            limits:
+              cpu: 500m
+              memory: 256Mi
+      volumes:
+        - name: test-secret-volume
+          secret:
+            secretName: k3s-test-app-secret
+        - name: storage-volume
+          persistentVolumeClaim:
+            claimName: k3s-test-app-pvc

projects/k3s-test-app/k8s/hpa.yaml

@@ -0,0 +1,18 @@
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: k3s-test-app-hpa
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: k3s-test-app
+  minReplicas: 1
+  maxReplicas: 2
+  metrics:
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: 60

projects/k3s-test-app/k8s/pvc.yaml

@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: k3s-test-app-pvc
+spec:
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 100Mi
+  # k3s uses local-path provisioner by default
+  storageClassName: local-path

projects/k3s-test-app/k8s/secret.yaml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: k3s-test-app-secret
+type: Opaque
+data:
+  # NOTE: This is a dummy non-sensitive test fixture to validate secret volume mounting. 
+  # Do not use this pattern for real production secrets.
+  api-key: c3VwZXItc2VjcmV0LWtleQ==  # 'super-secret-key' in base64

projects/k3s-test-app/k8s/service.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: k3s-test-app
+  labels:
+    app: k3s-test-app
+spec:
+  type: NodePort
+  selector:
+    app: k3s-test-app
+  ports:
+    - protocol: TCP
+      port: 80
+      targetPort: 8000
+      nodePort: 30080

projects/k3s-test-app/pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "k3s-test-app"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+authors = [
+    { name = "Gaurav5189", email = "gauravsethi4627@gmail.com" }
+]
+requires-python = ">=3.14"
+dependencies = [
+    "fastapi[standard]>=0.141.1",
+    "uvicorn[standard]>=0.34.0",
+    "pydantic-settings>=2.9.1",
+    "httpx>=0.28.1",
+    "prometheus-fastapi-instrumentator>=7.0.2",
+]
+
+[dependency-groups]
+dev = [
+    "pytest>=8.0",
+    "pytest-asyncio>=0.25",
+]
+
+[project.scripts]
+k3s-test-app = "k3s_test_app:main"
+
+[build-system]
+requires = ["uv_build>=0.12.5,<0.13.0"]
+build-backend = "uv_build"

projects/k3s-test-app/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+testpaths = tests

projects/k3s-test-app/src/k3s_test_app/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

projects/k3s-test-app/src/k3s_test_app/core/config.py

@@ -0,0 +1,18 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    app_name: str = "k3s-test-app"
+    app_version: str = "0.1.0"
+    log_level: str = "info"
+    storage_path: str = "/data"
+    ready: bool = True
+    custom_message: str = "default-message"
+    secret_path: str = "/etc/secrets/test-secret/api-key"
+
+    model_config = SettingsConfigDict(
+        env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+
+settings = Settings()

projects/k3s-test-app/src/k3s_test_app/core/logging.py

@@ -0,0 +1,40 @@
+import logging
+import json
+import sys
+from typing import Any, Dict
+
+
+class JSONFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        log_data: Dict[str, Any] = {
+            "level": record.levelname,
+            "message": record.getMessage(),
+            "name": record.name,
+            "time": self.formatTime(record, self.datefmt),
+        }
+        if record.exc_info:
+            log_data["exc_info"] = self.formatException(record.exc_info)
+        return json.dumps(log_data)
+
+
+def setup_logging(log_level_str: str) -> None:
+    level = getattr(logging, log_level_str.upper(), logging.INFO)
+    
+    logger = logging.getLogger("k3s_test_app")
+    logger.setLevel(level)
+    
+    # Avoid duplicate logs if setup_logging is called multiple times
+    logger.handlers.clear()
+    
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(JSONFormatter())
+    logger.addHandler(handler)
+    
+    # Configure uvicorn loggers to use the same formatter if they exist
+    for uvicorn_logger_name in ("uvicorn", "uvicorn.access", "uvicorn.error"):
+        uvicorn_logger = logging.getLogger(uvicorn_logger_name)
+        uvicorn_logger.handlers.clear()
+        uvicorn_logger.addHandler(handler)
+        
+        # Don't propagate to root logger
+        uvicorn_logger.propagate = False

projects/k3s-test-app/src/k3s_test_app/main.py

@@ -0,0 +1,51 @@
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from prometheus_fastapi_instrumentator import Instrumentator
+
+from k3s_test_app import __version__
+from k3s_test_app.core.config import settings
+from k3s_test_app.core.logging import setup_logging
+from k3s_test_app.routers import (
+    health, 
+    version, 
+    config, 
+    identity, 
+    network,
+    storage,
+    stress,
+    lifecycle,
+    upload,
+)
+
+logger = logging.getLogger("k3s_test_app")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    setup_logging(settings.log_level)
+    logger.info(f"Starting {settings.app_name} v{__version__}")
+    yield
+    logger.info(f"Shutting down {settings.app_name}")
+
+
+app = FastAPI(
+    title=settings.app_name,
+    version=__version__,
+    lifespan=lifespan,
+)
+
+# Instrument the app for Prometheus metrics
+Instrumentator().instrument(app).expose(app)
+
+# Include routers
+app.include_router(health.router, tags=["Health"])
+app.include_router(version.router, tags=["Info"])
+app.include_router(config.router, tags=["Config"])
+app.include_router(identity.router, tags=["Identity"])
+app.include_router(network.router, tags=["Network"])
+app.include_router(storage.router, tags=["Storage"])
+app.include_router(stress.router, tags=["Stress"])
+app.include_router(lifecycle.router, tags=["Lifecycle"])
+app.include_router(upload.router, tags=["Upload"])

projects/k3s-test-app/src/k3s_test_app/routers/__init__.py

@@ -0,0 +1 @@
+# Routers package

projects/k3s-test-app/src/k3s_test_app/routers/config.py

@@ -0,0 +1,38 @@
+import os
+from fastapi import APIRouter, HTTPException
+from k3s_test_app.core.config import settings
+import logging
+
+router = APIRouter()
+logger = logging.getLogger("k3s_test_app")
+
+
+@router.get("/config")
+async def get_config() -> dict:
+    """Return non-sensitive configuration values to verify ConfigMap injection."""
+    return {
+        "app_name": settings.app_name,
+        "log_level": settings.log_level,
+        "custom_message": settings.custom_message,
+    }
+
+
+@router.get("/secret")
+async def get_secret() -> dict:
+    """Read and return the mounted secret file to verify volume mounting."""
+    try:
+        with open(settings.secret_path, "r") as f:
+            f.read(1)
+        return {"secret_mounted": True}
+    except FileNotFoundError:
+        logger.error(f"Secret file not found at {settings.secret_path}")
+        raise HTTPException(
+            status_code=404,
+            detail="Secret file not found. Ensure it is mounted correctly."
+        )
+    except Exception as e:
+        logger.error(f"Error reading secret file: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Error reading secret file."
+        )

projects/k3s-test-app/src/k3s_test_app/routers/health.py

@@ -0,0 +1,21 @@
+from fastapi import APIRouter, Response, status
+from k3s_test_app.core.config import settings
+
+router = APIRouter()
+
+
+@router.get("/healthz", status_code=status.HTTP_200_OK)
+async def healthz() -> dict:
+    """Liveness probe endpoint."""
+    return {"status": "alive"}
+
+
+@router.get("/ready")
+async def ready(response: Response) -> dict:
+    """Readiness probe endpoint."""
+    if settings.ready:
+        response.status_code = status.HTTP_200_OK
+        return {"status": "ready"}
+    
+    response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
+    return {"status": "not_ready"}

projects/k3s-test-app/src/k3s_test_app/routers/identity.py

@@ -0,0 +1,23 @@
+import os
+import socket
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/whoami")
+async def whoami() -> dict:
+    """Return pod identity information (hostname and IP)."""
+    hostname = socket.gethostname()
+    # Try to get the IP from the Downward API first, fallback to socket resolution
+    pod_ip = os.getenv("POD_IP")
+    if not pod_ip:
+        try:
+            pod_ip = socket.gethostbyname(hostname)
+        except Exception:
+            pod_ip = "unknown"
+
+    return {
+        "hostname": hostname,
+        "pod_ip": pod_ip,
+    }

projects/k3s-test-app/src/k3s_test_app/routers/lifecycle.py

@@ -0,0 +1,22 @@
+import os
+import time
+from fastapi import APIRouter
+import logging
+
+router = APIRouter()
+logger = logging.getLogger("k3s_test_app")
+
+
+@router.get("/crash")
+async def crash() -> dict:
+    """Intentionally crash the pod to test liveness probe and restart policy."""
+    logger.critical("Intentional crash triggered! Exiting process.")
+    os._exit(1)
+
+
+@router.get("/sleep")
+async def sleep_block(seconds: int = 30) -> dict:
+    """Block the event loop using time.sleep to simulate a stuck process (tests liveness probe)."""
+    logger.warning(f"Blocking the main thread for {seconds} seconds.")
+    time.sleep(seconds)
+    return {"message": "Awoke from sleep."}

projects/k3s-test-app/src/k3s_test_app/routers/network.py

@@ -0,0 +1,46 @@
+import asyncio
+import httpx
+from fastapi import APIRouter, HTTPException, WebSocket, WebSocketDisconnect
+
+router = APIRouter()
+
+
+@router.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    """WebSocket endpoint that echoes messages back (tests long-lived connections)."""
+    await websocket.accept()
+    try:
+        while True:
+            data = await websocket.receive_text()
+            await websocket.send_text(f"Echo: {data}")
+    except WebSocketDisconnect:
+        pass
+
+
+@router.get("/delay/{seconds}")
+async def delay(seconds: int) -> dict:
+    """Simulate a delayed response for testing timeouts."""
+    await asyncio.sleep(seconds)
+    return {"message": f"Delayed for {seconds} seconds."}
+
+
+@router.get("/call-internal")
+async def call_internal() -> dict:
+    """Make an internal service-to-service call to validate cluster DNS."""
+    # This URL targets the service's internal cluster DNS name
+    url = "http://k3s-test-app.default.svc.cluster.local:80/healthz"
+    
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get(url)
+            response.raise_for_status()
+            return {
+                "status": "success",
+                "target_url": url,
+                "target_response": response.json(),
+            }
+    except httpx.HTTPError as e:
+        raise HTTPException(
+            status_code=502,
+            detail=f"Failed to communicate with internal service: {e}"
+        )

projects/k3s-test-app/src/k3s_test_app/routers/storage.py

@@ -0,0 +1,38 @@
+import os
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from k3s_test_app.core.config import settings
+
+router = APIRouter()
+
+
+class StorageData(BaseModel):
+    content: str
+
+
+@router.post("/storage")
+async def write_storage(data: StorageData) -> dict:
+    """Write data to the persistent volume."""
+    file_path = os.path.join(settings.storage_path, "test-file.txt")
+    try:
+        os.makedirs(settings.storage_path, exist_ok=True)
+        with open(file_path, "w") as f:
+            f.write(data.content)
+        return {"status": "success", "file": file_path}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to write to storage: {e}")
+
+
+@router.get("/storage")
+async def read_storage() -> dict:
+    """Read data from the persistent volume."""
+    file_path = os.path.join(settings.storage_path, "test-file.txt")
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="File not found in storage.")
+    
+    try:
+        with open(file_path, "r") as f:
+            content = f.read()
+        return {"content": content, "file": file_path}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to read from storage: {e}")

projects/k3s-test-app/src/k3s_test_app/routers/stress.py

@@ -0,0 +1,28 @@
+import time
+from fastapi import APIRouter
+import logging
+
+router = APIRouter()
+logger = logging.getLogger("k3s_test_app")
+
+
+@router.get("/stress/cpu")
+async def stress_cpu(seconds: int = 5) -> dict:
+    """Stress the CPU to trigger HPA scaling."""
+    logger.info(f"Stressing CPU for {seconds} seconds.")
+    end_time = time.time() + seconds
+    while time.time() < end_time:
+        pass
+    return {"message": f"Completed CPU stress for {seconds} seconds."}
+
+
+@router.get("/stress/mem")
+async def stress_mem(mb: int = 50) -> dict:
+    """Stress memory allocation. Do not allocate too much or OOMKilled will trigger."""
+    logger.info(f"Allocating {mb}MB of memory.")
+    # Allocate bytearray
+    data = bytearray(mb * 1024 * 1024)
+    # Give the system a second to register the memory usage before releasing
+    time.sleep(1)
+    del data
+    return {"message": f"Successfully allocated and freed {mb}MB."}

projects/k3s-test-app/src/k3s_test_app/routers/upload.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, UploadFile, File, HTTPException
+
+router = APIRouter()
+
+
+@router.post("/upload")
+async def upload_file(file: UploadFile = File(...)) -> dict:
+    """Accept a file upload and return its metadata to test body size limits."""
+    size_bytes = 0
+    max_upload_bytes = 10 * 1024 * 1024  # 10MB hard limit in the app
+    
+    while chunk := await file.read(1024 * 1024):
+        size_bytes += len(chunk)
+        if size_bytes > max_upload_bytes:
+            raise HTTPException(status_code=413, detail="Upload exceeds the 10MB app size limit.")
+    
+    return {
+        "filename": file.filename,
+        "content_type": file.content_type,
+        "size_bytes": size_bytes,
+        "message": f"Successfully received {size_bytes} bytes."
+    }

projects/k3s-test-app/src/k3s_test_app/routers/version.py

@@ -0,0 +1,15 @@
+import sys
+from fastapi import APIRouter
+from k3s_test_app.core.config import settings
+
+router = APIRouter()
+
+
+@router.get("/version")
+async def version() -> dict:
+    """Return version and environment information."""
+    return {
+        "app_name": settings.app_name,
+        "version": settings.app_version,
+        "python_version": sys.version,
+    }

projects/k3s-test-app/tests/__init__.py

@@ -0,0 +1 @@
+# Tests package

projects/k3s-test-app/tests/conftest.py

@@ -0,0 +1,12 @@
+"""Shared pytest fixtures for the k3s-test-app test suite."""
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient, ASGITransport
+from k3s_test_app.main import app
+
+
+@pytest_asyncio.fixture
+async def client() -> AsyncClient:
+    """Provide an async test client for the FastAPI app."""
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
+        yield ac

projects/k3s-test-app/tests/test_phase1_health.py

@@ -0,0 +1,41 @@
+"""Tests for Phase 1: health and version endpoints."""
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_healthz_returns_200(client: AsyncClient) -> None:
+    """Liveness probe should always return 200 with status=alive."""
+    response = await client.get("/healthz")
+    assert response.status_code == 200
+    assert response.json() == {"status": "alive"}
+
+
+@pytest.mark.asyncio
+async def test_ready_returns_200_by_default(client: AsyncClient) -> None:
+    """Readiness probe should return 200 when READY env var is true (default)."""
+    response = await client.get("/ready")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ready"}
+
+
+@pytest.mark.asyncio
+async def test_ready_returns_503_when_not_ready(client: AsyncClient, monkeypatch) -> None:
+    """Readiness probe should return 503 when settings.ready is False."""
+    from k3s_test_app.core import config as cfg_module
+    monkeypatch.setattr(cfg_module.settings, "ready", False)
+    response = await client.get("/ready")
+    assert response.status_code == 503
+    assert response.json() == {"status": "not_ready"}
+
+
+@pytest.mark.asyncio
+async def test_version_returns_expected_fields(client: AsyncClient) -> None:
+    """/version should return app_name, version, and python_version."""
+    response = await client.get("/version")
+    assert response.status_code == 200
+    data = response.json()
+    assert "app_name" in data
+    assert "version" in data
+    assert "python_version" in data
+    assert data["app_name"] == "k3s-test-app"

projects/k3s-test-app/tests/test_phase2_config.py

@@ -0,0 +1,54 @@
+"""Tests for Phase 2: config and secret endpoints."""
+import pytest
+import tempfile
+import os
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_config_returns_expected_fields(client: AsyncClient) -> None:
+    """/config should return app_name, log_level, and custom_message."""
+    response = await client.get("/config")
+    assert response.status_code == 200
+    data = response.json()
+    assert "app_name" in data
+    assert "log_level" in data
+    assert "custom_message" in data
+
+
+@pytest.mark.asyncio
+async def test_config_custom_message_is_overridable(client: AsyncClient, monkeypatch) -> None:
+    """CUSTOM_MESSAGE setting should be reflected in /config response."""
+    from k3s_test_app.core import config as cfg_module
+    monkeypatch.setattr(cfg_module.settings, "custom_message", "override-value")
+    response = await client.get("/config")
+    assert response.status_code == 200
+    assert response.json()["custom_message"] == "override-value"
+
+
+@pytest.mark.asyncio
+async def test_secret_returns_404_when_file_missing(client: AsyncClient, monkeypatch) -> None:
+    """/secret should return 404 when the secret file does not exist."""
+    from k3s_test_app.core import config as cfg_module
+    monkeypatch.setattr(cfg_module.settings, "secret_path", "/nonexistent/path/secret")
+    response = await client.get("/secret")
+    assert response.status_code == 404
+    assert "not found" in response.json()["detail"].lower()
+
+
+@pytest.mark.asyncio
+async def test_secret_returns_content_when_file_exists(client: AsyncClient, monkeypatch) -> None:
+    """/secret should return the file contents when the secret file exists."""
+    from k3s_test_app.core import config as cfg_module
+
+    with tempfile.NamedTemporaryFile(mode="w", suffix="-api-key", delete=False) as f:
+        f.write("my-test-secret")
+        tmp_path = f.name
+
+    try:
+        monkeypatch.setattr(cfg_module.settings, "secret_path", tmp_path)
+        response = await client.get("/secret")
+        assert response.status_code == 200
+        assert response.json() == {"secret_mounted": True}
+    finally:
+        os.unlink(tmp_path)

projects/k3s-test-app/tests/test_phase3_identity_network.py

@@ -0,0 +1,74 @@
+"""Tests for Phase 3: identity and network endpoints."""
+import pytest
+import socket
+from unittest.mock import AsyncMock, patch, MagicMock
+from httpx import AsyncClient
+import httpx
+
+
+@pytest.mark.asyncio
+async def test_whoami_returns_hostname_and_ip(client: AsyncClient) -> None:
+    """/whoami should return hostname and pod_ip fields."""
+    response = await client.get("/whoami")
+    assert response.status_code == 200
+    data = response.json()
+    assert "hostname" in data
+    assert "pod_ip" in data
+    # hostname should match the real system hostname
+    assert data["hostname"] == socket.gethostname()
+
+
+@pytest.mark.asyncio
+async def test_whoami_uses_pod_ip_env_var(client: AsyncClient, monkeypatch) -> None:
+    """/whoami should use the POD_IP env var when set (Downward API)."""
+    monkeypatch.setenv("POD_IP", "10.42.0.99")
+    response = await client.get("/whoami")
+    assert response.status_code == 200
+    assert response.json()["pod_ip"] == "10.42.0.99"
+
+
+@pytest.mark.asyncio
+async def test_delay_responds_after_sleep(client: AsyncClient) -> None:
+    """/delay/1 should return a success message (tests that it doesn't error)."""
+    # We patch asyncio.sleep to avoid actually waiting
+    with patch("k3s_test_app.routers.network.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+        response = await client.get("/delay/3")
+        mock_sleep.assert_awaited_once_with(3)
+    assert response.status_code == 200
+    assert response.json() == {"message": "Delayed for 3 seconds."}
+
+
+@pytest.mark.asyncio
+async def test_call_internal_returns_502_when_dns_fails(client: AsyncClient) -> None:
+    """/call-internal should return 502 when the internal service is unreachable."""
+    # In a local test environment, the cluster DNS won't resolve — this is expected
+    with patch("k3s_test_app.routers.network.httpx.AsyncClient") as mock_http:
+        mock_instance = AsyncMock()
+        mock_http.return_value.__aenter__ = AsyncMock(return_value=mock_instance)
+        mock_http.return_value.__aexit__ = AsyncMock(return_value=False)
+        mock_instance.get.side_effect = httpx.ConnectError("Name or service not known")
+
+        response = await client.get("/call-internal")
+        assert response.status_code == 502
+        assert "Failed to communicate with internal service" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_call_internal_returns_success_when_service_responds(client: AsyncClient) -> None:
+    """/call-internal should return 200 with target_response when the service is up."""
+    mock_response = MagicMock()
+    mock_response.raise_for_status = MagicMock()
+    mock_response.json.return_value = {"status": "alive"}
+
+    with patch("k3s_test_app.routers.network.httpx.AsyncClient") as mock_http:
+        mock_instance = AsyncMock()
+        mock_http.return_value.__aenter__ = AsyncMock(return_value=mock_instance)
+        mock_http.return_value.__aexit__ = AsyncMock(return_value=False)
+        mock_instance.get.return_value = mock_response
+
+        response = await client.get("/call-internal")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "success"
+        assert data["target_response"] == {"status": "alive"}
+        assert "k3s-test-app.default.svc.cluster.local" in data["target_url"]

projects/k3s-test-app/tests/test_phase4_storage.py

@@ -0,0 +1,39 @@
+"""Tests for Phase 4: storage endpoints."""
+import pytest
+import os
+import tempfile
+import shutil
+from httpx import AsyncClient
+
+
+@pytest.fixture
+def temp_storage_dir(monkeypatch):
+    """Fixture to mock settings.storage_path with a temporary directory."""
+    temp_dir = tempfile.mkdtemp()
+    from k3s_test_app.core import config as cfg_module
+    monkeypatch.setattr(cfg_module.settings, "storage_path", temp_dir)
+    yield temp_dir
+    shutil.rmtree(temp_dir)
+
+
+@pytest.mark.asyncio
+async def test_read_storage_not_found(client: AsyncClient, temp_storage_dir: str) -> None:
+    """GET /storage should return 404 if nothing has been written yet."""
+    response = await client.get("/storage")
+    assert response.status_code == 404
+    assert "not found" in response.json()["detail"].lower()
+
+
+@pytest.mark.asyncio
+async def test_write_and_read_storage(client: AsyncClient, temp_storage_dir: str) -> None:
+    """POST /storage should write data, and GET /storage should read it back."""
+    # Write data
+    post_response = await client.post("/storage", json={"content": "hello storage"})
+    assert post_response.status_code == 200
+    assert post_response.json()["status"] == "success"
+    
+    # Read data
+    get_response = await client.get("/storage")
+    assert get_response.status_code == 200
+    assert get_response.json()["content"] == "hello storage"
+    assert get_response.json()["file"].startswith(temp_storage_dir)

projects/k3s-test-app/tests/test_phase5_resilience.py

@@ -0,0 +1,51 @@
+"""Tests for Phase 5: load and resilience endpoints."""
+import pytest
+from httpx import AsyncClient
+from unittest.mock import patch
+
+
+def mock_time_gen():
+    t = 0
+    while True:
+        yield t
+        t += 10
+
+@pytest.mark.asyncio
+async def test_stress_cpu(client: AsyncClient) -> None:
+    """GET /stress/cpu should return 200 (mocking the time.time() to finish instantly)."""
+    with patch("k3s_test_app.routers.stress.time.time", side_effect=mock_time_gen()):
+        response = await client.get("/stress/cpu?seconds=5")
+        assert response.status_code == 200
+        assert "Completed CPU stress" in response.json()["message"]
+
+
+@pytest.mark.asyncio
+async def test_stress_mem(client: AsyncClient) -> None:
+    """GET /stress/mem should return 200."""
+    with patch("k3s_test_app.routers.stress.time.sleep"):
+        response = await client.get("/stress/mem?mb=1")
+        assert response.status_code == 200
+        assert "allocated and freed" in response.json()["message"]
+
+
+@pytest.mark.asyncio
+async def test_crash(client: AsyncClient) -> None:
+    """GET /crash should call os._exit(1)."""
+    class IntentionalCrash(Exception): pass
+    
+    with patch("k3s_test_app.routers.lifecycle.os._exit", side_effect=IntentionalCrash) as mock_exit:
+        try:
+            await client.get("/crash")
+        except IntentionalCrash:
+            pass # Expected
+        mock_exit.assert_called_once_with(1)
+
+
+@pytest.mark.asyncio
+async def test_sleep_block(client: AsyncClient) -> None:
+    """GET /sleep should call time.sleep and block the thread."""
+    with patch("k3s_test_app.routers.lifecycle.time.sleep") as mock_sleep:
+        response = await client.get("/sleep?seconds=2")
+        mock_sleep.assert_called_once_with(2)
+        assert response.status_code == 200
+        assert response.json()["message"] == "Awoke from sleep."

projects/k3s-test-app/tests/test_phase6_realtime_upload.py

@@ -0,0 +1,56 @@
+"""Tests for Phase 6: realtime and upload endpoints."""
+import pytest
+from httpx import AsyncClient
+import io
+
+
+@pytest.mark.asyncio
+async def test_websocket_echo() -> None:
+    """Test the WebSocket /ws endpoint."""
+    # We use a synchronous TestClient from fastapi.testclient for websockets
+    # because httpx AsyncClient doesn't have built-in WS support yet
+    from fastapi.testclient import TestClient
+    from k3s_test_app.main import app
+    
+    client = TestClient(app)
+    with client.websocket_connect("/ws") as websocket:
+        websocket.send_text("Hello k3s")
+        data = websocket.receive_text()
+        assert data == "Echo: Hello k3s"
+
+
+@pytest.mark.asyncio
+async def test_upload_file(client: AsyncClient) -> None:
+    """POST /upload should accept a file and return its size."""
+    file_content = b"This is a test file for upload."
+    files = {"file": ("test.txt", io.BytesIO(file_content), "text/plain")}
+    
+    response = await client.post("/upload", files=files)
+    assert response.status_code == 200
+    
+    data = response.json()
+    assert data["filename"] == "test.txt"
+    assert data["size_bytes"] == len(file_content)
+
+
+@pytest.mark.asyncio
+async def test_upload_file_exceeds_limit(client: AsyncClient) -> None:
+    """POST /upload should return 413 if the file exceeds the 10MB limit."""
+    # We can create a fake file-like object that returns data to simulate a large upload
+    # without allocating 10MB+ in memory all at once.
+    class LargeMockFile:
+        def __init__(self, size):
+            self.remaining = size
+        def read(self, size=-1):
+            if self.remaining <= 0:
+                return b""
+            chunk_size = size if size > 0 else self.remaining
+            chunk_size = min(chunk_size, self.remaining)
+            self.remaining -= chunk_size
+            return b"0" * chunk_size
+            
+    files = {"file": ("huge.txt", LargeMockFile(11 * 1024 * 1024), "text/plain")}
+    
+    response = await client.post("/upload", files=files)
+    assert response.status_code == 413
+    assert "exceeds" in response.json()["detail"].lower()

projects/k3s-test-app/tests/test_phase7_observability.py

@@ -0,0 +1,15 @@
+"""Tests for Phase 7: observability endpoints."""
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_metrics_endpoint_returns_prometheus_data(client: AsyncClient) -> None:
+    """GET /metrics should return Prometheus metrics in text format."""
+    response = await client.get("/metrics")
+    assert response.status_code == 200
+    
+    # Prometheus metrics are plain text
+    content = response.text
+    assert "TYPE" in content or "HELP" in content
+    assert "fastapi_" in content or "python_" in content or "process_" in content