Files
cleveragents-core/robot/helper_k8s_helm_chart.py
brent.edwards b51df2ee0f
CI / docker (push) Blocked by required conditions
CI / status-check (push) Blocked by required conditions
CI / build (push) Successful in 20s
CI / helm (push) Successful in 23s
CI / lint (push) Successful in 3m19s
CI / typecheck (push) Successful in 3m57s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 4m5s
CI / integration_tests (push) Successful in 7m25s
CI / unit_tests (push) Failing after 13m18s
CI / quality (push) Failing after 13m19s
CI / e2e_tests (push) Failing after 18m16s
CI / coverage (push) Failing after 19m20s
CI / benchmark-publish (push) Failing after 28m16s
feat(server): add Kubernetes Helm chart for server deployment (#1085)
## Summary

This PR adds Kubernetes Helm deployment support for the CleverAgents server.

Closes #928

### What this PR includes

- Helm chart under `k8s/` with Deployment, Service, optional Ingress, ConfigMap,
  ServiceAccount, Secrets, NOTES, and optional Redis subchart configuration.
- Multi-stage `Dockerfile.server` for server runtime deployment.
- Deployment-focused docs in `k8s/README.md`.
- Behave + Robot + benchmark coverage for chart/deployment wiring.

### Review fixes applied (cycle 11 — hurui200320 review #2687)

**Critical fix:**

1. **CI SHA256 checksum verification fixed** — All 3 Helm install blocks (`unit_tests`, `integration_tests`, `helm` jobs) now save the tarball using its original filename (`helm-v3.16.4-linux-amd64.tar.gz`) instead of `helm.tgz`, so the `.sha256sum` file can correctly locate and verify it. This was causing all 3 CI Helm jobs to fail with "No such file or directory".

**Major fixes (test coverage gaps):**

2. **405 Allow header now tested** — The existing "POST to known path returns method not allowed" scenario now asserts that the `Allow: GET` header is present. Added a second 405 scenario testing `POST /live` for broader `_KNOWN_PATHS` coverage (review item #11).
3. **Security-hardening headers now tested** — New scenario "HTTP responses include security-hardening headers" verifies `content-length`, `x-content-type-options: nosniff`, and `cache-control: no-store` headers are present on HTTP responses.
4. **Lifespan warning logging now tested** — New scenario "Unrecognised lifespan message type logs warning and continues" queues `lifespan.startup` → `lifespan.bogus` → `lifespan.shutdown` and verifies: (a) the app completes the lifespan cycle cleanly, (b) a warning is logged mentioning the unrecognised type.

### Review fixes applied (cycle 10)

**Critical/Major fixes (from hurui200320's prior REQUEST_CHANGES):**

1. **Rebased branch onto master** — Removed merge commit per CONTRIBUTING.md rebase-only policy. Clean linear history restored.
2. **Fixed commit message body** — Replaced literal `\n` sequences with actual newlines. `ISSUES CLOSED: #928` footer is now on its own line after a blank separator.
3. **Moved uvicorn import to module top level** — `from uvicorn import run as uvicorn_run` is now at the top of `src/cleveragents/cli/commands/server.py` per Import Guidelines. Updated test mock target from `uvicorn.run` to `cleveragents.cli.commands.server.uvicorn_run`.
4. **Added SHA256 checksum verification** — All three Helm CLI install blocks in `.forgejo/workflows/ci.yml` now download and verify `helm.sha256sum` before extracting the binary.

**Minor fixes:**

5. **Added `Dockerfile.server` build to CI** — New "Build Docker image (Server)" step in the `docker` job validates the server Dockerfile.
6. **ASGI 405 Method Not Allowed** — Known paths (`/`, `/live`, `/ready`, `/health`) now return 405 with `Allow: GET` header for non-GET methods, per RFC 9110 §15.5.6. Added `_KNOWN_PATHS` frozenset.
7. **WebSocket close protocol fix** — App now calls `await receive()` to consume the `websocket.connect` event before closing. Changed close code from 1000 (Normal Closure) to 1008 (Policy Violation).
8. **Lifespan handler logging** — Unrecognised lifespan message types are now logged as warnings instead of silently consumed.
9. **Security-hardening headers** — `_send_response` now includes `content-length`, `x-content-type-options: nosniff`, and `cache-control: no-store` on all HTTP responses.
10. **`.dockerignore` credential patterns** — Added `*.pem`, `*.key`, `*.p12`, `*.pfx`, `credentials*.json`.
11. **`--log-level` validation** — Constrained to `click.Choice(["critical", "error", "warning", "info", "debug", "trace"])` for clean CLI validation errors.
12. **Reverted unrelated semgrep pre-commit change** — `pass_filenames` and `entry` restored to original values per atomic commit hygiene.
13. **Removed unused `ReceiveCallable` type alias** and `Callable`/`Awaitable` imports from `asgi_app_steps.py`.
14. **Fixed redundant `shutil.which("helm")` check** — `_skip_if_helm_missing` now returns `bool` to eliminate the duplicate check in `_render_chart`.
15. **Improved test deque error handling** — Lifespan test receive mock now raises descriptive `AssertionError` instead of opaque `IndexError`.
16. **Scope type dispatch** — Changed `if/if/if` to `if/elif/elif` for mutually exclusive ASGI scope types.
17. **Dockerfile.server base image** — Standardised to `python:3.13-slim` (floating minor) consistent with CLI Dockerfile.
18. **Dockerfile layer caching** — Split `uv pip install build` and `python -m build` into separate `RUN` instructions.
19. **Removed extraneous double blank line** in Dockerfile.server.

### Deferred items (acknowledged, not in scope)

- PodDisruptionBudget, HorizontalPodAutoscaler, NetworkPolicy — Follow-up for production hardening.
- `appVersion: "1.0.0"` placeholder — Needs tracking issue for release versioning alignment.
- Readiness probe with downstream dependency checks — Documented limitation.
- Cross-system test for probe paths matching ASGI routes — Test enhancement.
- Improved benchmarks (helm template timing vs PyYAML parsing) — Benchmark quality improvement.
- CI DRY violation (Helm install 3×) — Code quality improvement, consider composite action.
- File length limits exceeded (`k8s_helm_chart_steps.py` 551 lines, `helper_k8s_helm_chart.py` 678 lines) — Non-blocking, can be split in follow-up.
- `runAsGroup: 1000` in pod security context — Defense-in-depth improvement.
- HEAD method support on known paths — RFC compliance, does not affect K8s probes.
- `click.Choice` log-level validation via CLI runner test — Test gap.

### Scope note: status-check CI gate

The `status-check` job now includes `integration_tests`, `e2e_tests`, and `helm` in its `needs` list. The `helm` job is new in this PR. The `integration_tests` and `e2e_tests` additions fix previously-missing gate checks — included here since this PR modifies both of those jobs to install Helm.

### Quality gates

- `nox -e lint` 
- `nox -e typecheck` 
- `nox -e unit_tests`  (12,321 scenarios passed, 4 skipped)
- `nox -e integration_tests` — 3 pre-existing failures in unrelated areas (plan correction, resource types)
- `nox -e e2e_tests` — pre-existing failures (LLM API keys not available in local env)
- `nox -e coverage_report`  (**97.7%**)

Reviewed-on: #1085
Reviewed-by: Jeffrey Phillips Freeman <jeffrey.freeman@cleverthis.com>
Co-authored-by: Brent E. Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent E. Edwards <brent.edwards@cleverthis.com>
2026-03-27 23:53:22 +00:00

679 lines
28 KiB
Python

"""Helper script for Kubernetes Helm chart Robot Framework integration tests.
Provides validation functions for Helm chart structure and content,
focusing on cross-file consistency checks that verify interactions
between multiple chart files (templates, values, Dockerfile).
"""
from __future__ import annotations
import os
import re
import shutil
import subprocess
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any
import yaml
def _k8s_dir() -> Path:
"""Return the k8s/ directory path."""
return Path(__file__).resolve().parent.parent / "k8s"
def _render_assertions_required() -> bool:
"""Return whether rendered Helm assertions are required in this run."""
value = os.environ.get("CLEVERAGENTS_REQUIRE_HELM_RENDER_ASSERTIONS", "").lower()
return value in {"1", "true", "yes", "on"}
def _render_chart(set_overrides: list[str] | None = None) -> str | None:
"""Render the chart with ``helm template`` when Helm is available.
Returns rendered YAML text, or ``None`` when Helm is unavailable in the
current environment.
"""
if shutil.which("helm") is None:
if _render_assertions_required():
raise AssertionError(
"Helm CLI is required when "
"CLEVERAGENTS_REQUIRE_HELM_RENDER_ASSERTIONS=true"
)
return None
command: list[str] = [
"helm",
"template",
"cleveragents",
str(_k8s_dir()),
"--dependency-update",
"--set",
# Deployment template now requires explicit DB configuration.
"database.url=postgresql+asyncpg://user:pass@db:5432/cleveragents",
]
if set_overrides:
for item in set_overrides:
command.extend(["--set", item])
result = subprocess.run(
command,
cwd=str(_k8s_dir().parent),
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0, (
"helm template failed: "
f"command={' '.join(command)}\n"
f"stdout={result.stdout}\n"
f"stderr={result.stderr}"
)
return result.stdout
def _parse_rendered_resources(rendered: str) -> list[dict[str, Any]]:
"""Parse rendered multi-doc YAML into Kubernetes resource dictionaries."""
resources: list[dict[str, Any]] = []
for document in rendered.split("\n---\n"):
if not document.strip():
continue
loaded = yaml.safe_load(document)
if isinstance(loaded, dict):
resources.append(loaded)
return resources
def _find_resource(
resources: list[dict[str, Any]],
*,
kind: str,
name: str,
) -> dict[str, Any]:
for resource in resources:
metadata = resource.get("metadata", {})
if resource.get("kind") == kind and metadata.get("name") == name:
return resource
raise AssertionError(f"Rendered resource not found: kind={kind}, name={name}")
def _deployment_env_entries(resources: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return rendered deployment env entries for the main container."""
deployment = _find_resource(resources, kind="Deployment", name="cleveragents")
env_entries = deployment["spec"]["template"]["spec"]["containers"][0].get("env", [])
return [entry for entry in env_entries if isinstance(entry, dict)]
def _env_entry_by_name(
env_entries: list[dict[str, Any]], env_name: str
) -> dict[str, Any]:
"""Return a single environment variable entry by name."""
for entry in env_entries:
if entry.get("name") == env_name:
return entry
raise AssertionError(f"Rendered env entry not found: {env_name}")
def validate_cross_file_port_consistency() -> str:
"""Validate port values are consistent across templates and values.
Cross-file consistency check: verifies that the Deployment template
references server.port for containerPort, that values.yaml service.port
matches server.port, and that the Service template uses the named port.
"""
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
data: dict[str, Any] = yaml.safe_load(f)
service_port: int = data["service"]["port"]
server_port: int = data["server"]["port"]
assert service_port == server_port, (
f"service.port ({service_port}) != server.port ({server_port})"
)
# Verify deployment template uses server.port for containerPort
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert ".Values.server.port" in deployment_content, (
"Deployment containerPort should reference .Values.server.port"
)
# Cross-file: service.yaml must use the named port "http" as targetPort
service_path: Path = _k8s_dir() / "templates" / "service.yaml"
service_content: str = service_path.read_text(encoding="utf-8")
assert "targetPort: http" in service_content, (
"Service targetPort should reference the named port 'http'"
)
# Cross-file: Dockerfile EXPOSE port should match values default
dockerfile_path: Path = _k8s_dir().parent / "Dockerfile.server"
dockerfile_content: str = dockerfile_path.read_text(encoding="utf-8")
expose_pattern: re.Pattern[str] = re.compile(
rf"^EXPOSE\s+{server_port}$", re.MULTILINE
)
assert expose_pattern.search(dockerfile_content), (
f"Dockerfile EXPOSE should match server.port ({server_port})"
)
return "Port consistency OK"
def validate_secrets_structure() -> str:
"""Validate secrets template references match values.yaml structure.
Cross-file check: verifies the secrets template references the same
keys that values.yaml defines for database and Redis credentials, and
that the deployment template references the same secret names.
"""
secrets_path: Path = _k8s_dir() / "templates" / "secrets.yaml"
secrets_content: str = secrets_path.read_text(encoding="utf-8")
assert "database.url" in secrets_content, "Secrets should reference database.url"
assert "database.existingSecret" in secrets_content, (
"Secrets should check database.existingSecret"
)
assert "redis.auth.password" in secrets_content, (
"Secrets should reference redis.auth.password"
)
assert "redis.auth.existingSecret" in secrets_content, (
"Secrets should check redis.auth.existingSecret"
)
# Cross-file: deployment.yaml should reference the same secret keys
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert "database-url" in deployment_content, (
"Deployment should reference database-url secret key"
)
assert "redis-password" in deployment_content, (
"Deployment should reference redis-password secret key"
)
# Cross-file: values.yaml should define the keys the secrets reference
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
values: dict[str, Any] = yaml.safe_load(f)
assert "url" in values["database"], "values.yaml missing database.url"
assert "existingSecret" in values["database"], (
"values.yaml missing database.existingSecret"
)
assert "password" in values["redis"]["auth"], (
"values.yaml missing redis.auth.password"
)
assert "existingSecret" in values["redis"]["auth"], (
"values.yaml missing redis.auth.existingSecret"
)
return "Secrets structure OK"
def validate_deployment_command_args() -> str:
"""Validate Deployment template command/args match Dockerfile entrypoint.
Cross-file check: verifies the Deployment template passes server.host,
server.port, server.workers, and server.logLevel as CLI args to the
CleverAgents server command, and that the command matches Dockerfile.
"""
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert "server.host" in deployment_content, "Deployment should pass server.host"
assert "server.port" in deployment_content, "Deployment should pass server.port"
assert "server.workers" in deployment_content, (
"Deployment should pass server.workers"
)
assert "server.logLevel" in deployment_content, (
"Deployment should pass server.logLevel"
)
assert 'command: ["python", "-m", "cleveragents"]' in deployment_content, (
"Deployment command should invoke python -m cleveragents"
)
assert '"server"' in deployment_content, "Deployment args should include server"
assert '"serve"' in deployment_content, "Deployment args should include serve"
assert '"--app"' in deployment_content, "Deployment args should include --app"
# Cross-file: Dockerfile should also use cleveragents entrypoint
dockerfile_path: Path = _k8s_dir().parent / "Dockerfile.server"
dockerfile_content: str = dockerfile_path.read_text(encoding="utf-8")
assert 'ENTRYPOINT ["python", "-m", "cleveragents"]' in dockerfile_content, (
"Dockerfile ENTRYPOINT should use python -m cleveragents"
)
# Cross-file: both should reference the same ASGI app module
assert "cleveragents.a2a.asgi" in deployment_content, (
"Deployment should reference cleveragents.a2a.asgi"
)
assert "cleveragents.a2a.asgi" in dockerfile_content, (
"Dockerfile should reference cleveragents.a2a.asgi"
)
rendered = _render_chart()
if rendered is None:
return "SKIP: deployment-command-args-rendered"
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
values: dict[str, Any] = yaml.safe_load(f)
expected_args = [
"server",
"serve",
"--app",
"cleveragents.a2a.asgi:app",
"--host",
str(values["server"]["host"]),
"--port",
str(values["server"]["port"]),
"--workers",
str(values["server"]["workers"]),
"--log-level",
str(values["server"]["logLevel"]),
]
resources = _parse_rendered_resources(rendered)
deployment = _find_resource(resources, kind="Deployment", name="cleveragents")
containers = deployment["spec"]["template"]["spec"]["containers"]
assert containers, "Rendered deployment should include at least one container"
container = containers[0]
assert container.get("command") == ["python", "-m", "cleveragents"], (
"Rendered deployment command should be exactly ['python', '-m', 'cleveragents']"
)
assert container.get("args") == expected_args, (
"Rendered deployment args do not match expected command contract\n"
f"expected={expected_args}\n"
f"actual={container.get('args')}"
)
return "OK: deployment-command-args-rendered"
def validate_envfrom_wiring() -> str:
"""Validate Deployment envFrom references the correct ConfigMap name.
Cross-file check: verifies the Deployment template uses envFrom with
configMapRef, and that the ConfigMap template uses the matching name
suffix.
"""
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert "envFrom" in deployment_content, "Deployment should use envFrom"
assert "configMapRef" in deployment_content, "Deployment should reference ConfigMap"
# Cross-file: ConfigMap uses -config suffix; deployment should too
configmap_path: Path = _k8s_dir() / "templates" / "configmap.yaml"
configmap_content: str = configmap_path.read_text(encoding="utf-8")
assert "-config" in configmap_content, (
"ConfigMap name should include -config suffix"
)
assert "-config" in deployment_content, (
"Deployment configMapRef should reference -config suffix"
)
# Cross-file: ConfigMap env vars should match values.yaml keys
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
values: dict[str, Any] = yaml.safe_load(f)
assert "host" in values["server"], "values.yaml missing server.host for ConfigMap"
assert "port" in values["server"], "values.yaml missing server.port for ConfigMap"
assert "logLevel" in values["server"], "values.yaml missing server.logLevel"
assert "CLEVERAGENTS_SERVER_HOST" in configmap_content, (
"ConfigMap should map CLEVERAGENTS_SERVER_HOST"
)
assert "CLEVERAGENTS_SERVER_PORT" in configmap_content, (
"ConfigMap should map CLEVERAGENTS_SERVER_PORT"
)
assert "CLEVERAGENTS_LOG_LEVEL" in configmap_content, (
"ConfigMap should map CLEVERAGENTS_LOG_LEVEL"
)
rendered = _render_chart()
if rendered is None:
return "SKIP: envfrom-wiring-rendered"
resources = _parse_rendered_resources(rendered)
deployment = _find_resource(resources, kind="Deployment", name="cleveragents")
configmap = _find_resource(resources, kind="ConfigMap", name="cleveragents-config")
env_from = deployment["spec"]["template"]["spec"]["containers"][0].get(
"envFrom", []
)
assert env_from, "Rendered deployment should include envFrom"
ref_name = env_from[0]["configMapRef"]["name"]
assert ref_name == configmap["metadata"]["name"], (
"Rendered envFrom configMapRef should match rendered ConfigMap name"
)
return "OK: envfrom-wiring-rendered"
def validate_redis_disabled_rendering() -> str:
"""Validate rendered manifests omit Redis env when redis.enabled=false."""
rendered = _render_chart(["redis.enabled=false"])
if rendered is None:
return "SKIP: redis-disabled-rendering"
resources = _parse_rendered_resources(rendered)
deployment = _find_resource(resources, kind="Deployment", name="cleveragents")
env_entries = deployment["spec"]["template"]["spec"]["containers"][0].get("env", [])
env_names = {entry.get("name") for entry in env_entries if isinstance(entry, dict)}
assert "CLEVERAGENTS_REDIS_HOST" not in env_names, (
"Redis host env var must be absent when redis.enabled=false"
)
assert "CLEVERAGENTS_REDIS_PORT" not in env_names, (
"Redis port env var must be absent when redis.enabled=false"
)
assert "CLEVERAGENTS_REDIS_PASSWORD" not in env_names, (
"Redis password env var must be absent when redis.enabled=false"
)
return "OK: redis-disabled-rendering"
def validate_database_existing_secret_rendering() -> str:
"""Validate rendered DB env wiring uses database.existingSecret branch."""
rendered = _render_chart(
[
"database.existingSecret=custom-db-secret",
"database.existingSecretKey=db-url",
]
)
if rendered is None:
return "SKIP: database-existing-secret-rendering"
resources = _parse_rendered_resources(rendered)
env_entries = _deployment_env_entries(resources)
db_env = _env_entry_by_name(env_entries, "CLEVERAGENTS_DATABASE_URL")
secret_ref = db_env["valueFrom"]["secretKeyRef"]
assert secret_ref["name"] == "custom-db-secret", (
"Deployment should reference configured database.existingSecret"
)
assert secret_ref["key"] == "db-url", (
"Deployment should reference configured database.existingSecretKey"
)
return "OK: database-existing-secret-rendering"
def validate_redis_enabled_existing_secret_rendering() -> str:
"""Validate redis.enabled + redis.auth.existingSecret render wiring."""
rendered = _render_chart(
[
"redis.enabled=true",
"redis.auth.enabled=true",
"redis.auth.existingSecret=custom-redis-secret",
"redis.auth.existingSecretPasswordKey=custom-redis-key",
]
)
if rendered is None:
return "SKIP: redis-enabled-existing-secret-rendering"
resources = _parse_rendered_resources(rendered)
env_entries = _deployment_env_entries(resources)
password_env = _env_entry_by_name(env_entries, "CLEVERAGENTS_REDIS_PASSWORD")
secret_ref = password_env["valueFrom"]["secretKeyRef"]
assert secret_ref["name"] == "custom-redis-secret", (
"Redis password should come from redis.auth.existingSecret"
)
assert secret_ref["key"] == "custom-redis-key", (
"Redis password should use redis.auth.existingSecretPasswordKey"
)
return "OK: redis-enabled-existing-secret-rendering"
def validate_redis_enabled_inline_password_rendering() -> str:
"""Validate redis.enabled + redis.auth.password render wiring."""
rendered = _render_chart(
[
"redis.enabled=true",
"redis.auth.enabled=true",
"redis.auth.password=inline-secret",
]
)
if rendered is None:
return "SKIP: redis-enabled-inline-password-rendering"
resources = _parse_rendered_resources(rendered)
env_entries = _deployment_env_entries(resources)
password_env = _env_entry_by_name(env_entries, "CLEVERAGENTS_REDIS_PASSWORD")
secret_ref = password_env["valueFrom"]["secretKeyRef"]
assert secret_ref["name"] == "cleveragents-redis", (
"Inline password path should reference generated redis Secret"
)
assert secret_ref["key"] == "redis-password", (
"Inline password path should use redis-password key"
)
redis_secrets = [
resource
for resource in resources
if resource.get("kind") == "Secret"
and resource.get("metadata", {}).get("name") == "cleveragents-redis"
]
assert redis_secrets, (
"Expected at least one rendered Secret named cleveragents-redis"
)
assert any(
"redis-password" in (secret.get("stringData") or secret.get("data") or {})
for secret in redis_secrets
), "Expected rendered redis Secret data to include redis-password key"
return "OK: redis-enabled-inline-password-rendering"
def validate_redis_enabled_bitnami_fallback_rendering() -> str:
"""Validate redis.enabled fallback uses Bitnami-provisioned secret wiring."""
rendered = _render_chart(["redis.enabled=true", "redis.auth.enabled=true"])
if rendered is None:
return "SKIP: redis-enabled-bitnami-fallback-rendering"
resources = _parse_rendered_resources(rendered)
env_entries = _deployment_env_entries(resources)
password_env = _env_entry_by_name(env_entries, "CLEVERAGENTS_REDIS_PASSWORD")
secret_ref = password_env["valueFrom"]["secretKeyRef"]
assert secret_ref["name"] == "cleveragents-redis", (
"Fallback Redis path should reference Bitnami redis secret"
)
assert secret_ref["key"] == "redis-password", (
"Fallback Redis path should use redis-password key"
)
# Ensure dependency resources are rendered (proves dependency wiring).
_find_resource(resources, kind="Secret", name="cleveragents-redis")
return "OK: redis-enabled-bitnami-fallback-rendering"
def validate_ingress_tls_rendering_success() -> str:
"""Validate ingress render succeeds when TLS is configured."""
rendered = _render_chart(
[
"ingress.enabled=true",
"ingress.allowInsecure=false",
"ingress.tls[0].secretName=cleveragents-tls",
"ingress.tls[0].hosts[0]=cleveragents.example.com",
]
)
if rendered is None:
return "SKIP: ingress-tls-rendering-success"
resources = _parse_rendered_resources(rendered)
ingress = _find_resource(resources, kind="Ingress", name="cleveragents")
tls_entries = ingress.get("spec", {}).get("tls", [])
assert tls_entries, "Ingress spec.tls should be rendered when TLS is configured"
assert tls_entries[0].get("secretName") == "cleveragents-tls", (
"Ingress tls secretName should match configured value"
)
return "OK: ingress-tls-rendering-success"
def validate_redis_host_helper() -> str:
"""Validate Redis host helper uses Release.Name matching Bitnami naming.
Cross-file check: the Redis host helper should use .Release.Name
to match Bitnami subchart service naming, and the deployment template
should use this helper for the Redis host env var.
"""
helpers_path: Path = _k8s_dir() / "templates" / "_helpers.tpl"
helpers_content: str = helpers_path.read_text(encoding="utf-8")
assert "cleveragents.redisHost" in helpers_content, "Missing redisHost helper"
assert ".Release.Name" in helpers_content, "Redis host should use .Release.Name"
assert "redis-master" in helpers_content, (
"Redis host should follow Bitnami naming convention"
)
# Cross-file: deployment.yaml should use the helper
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert "cleveragents.redisHost" in deployment_content, (
"Deployment should use the redisHost helper"
)
# Cross-file: Chart.yaml should declare redis dependency
chart_path: Path = _k8s_dir() / "Chart.yaml"
with open(chart_path, encoding="utf-8") as f:
chart: dict[str, Any] = yaml.safe_load(f)
dep_names: list[str] = [d.get("name", "") for d in chart.get("dependencies", [])]
assert "redis" in dep_names, "Chart.yaml should declare redis dependency"
return "Redis host helper OK"
def validate_probe_config() -> str:
"""Validate probe configuration matches across values and deployment.
Cross-file check: verifies probes in values.yaml split /live and /ready,
and that the deployment template renders them via toYaml.
"""
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
data: dict[str, Any] = yaml.safe_load(f)
liveness: dict[str, Any] = data["livenessProbe"]
readiness: dict[str, Any] = data["readinessProbe"]
assert liveness["httpGet"]["path"] == "/live", "Liveness should probe /live"
assert readiness["httpGet"]["path"] == "/ready", "Readiness should probe /ready"
assert "initialDelaySeconds" in liveness, "Liveness missing initialDelaySeconds"
assert "initialDelaySeconds" in readiness, "Readiness missing initialDelaySeconds"
# Cross-file: deployment template should render these probes
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert "livenessProbe" in deployment_content, (
"Deployment should render livenessProbe"
)
assert "readinessProbe" in deployment_content, (
"Deployment should render readinessProbe"
)
return "Probe config OK"
def validate_image_defaults() -> str:
"""Validate image defaults are consistent across values and deployment.
Cross-file check: verifies image section in values.yaml has sensible
defaults, and that the deployment template correctly references them.
"""
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
data: dict[str, Any] = yaml.safe_load(f)
image: dict[str, Any] = data["image"]
assert "repository" in image, "Missing image.repository"
assert "pullPolicy" in image, "Missing image.pullPolicy"
assert image["pullPolicy"] == "IfNotPresent", (
f"Expected IfNotPresent, got {image['pullPolicy']}"
)
assert "tag" in image, "Missing image.tag"
# Cross-file: deployment template should reference image values
deployment_path: Path = _k8s_dir() / "templates" / "deployment.yaml"
deployment_content: str = deployment_path.read_text(encoding="utf-8")
assert ".Values.image.repository" in deployment_content, (
"Deployment should reference .Values.image.repository"
)
assert ".Values.image.pullPolicy" in deployment_content, (
"Deployment should reference .Values.image.pullPolicy"
)
assert ".Values.image.tag" in deployment_content, (
"Deployment should reference .Values.image.tag"
)
return "Image defaults OK"
def validate_dockerfile_multistage_nonroot() -> str:
"""Validate Dockerfile.server multi-stage build with non-root user.
Cross-file check: verifies the Dockerfile uses a multi-stage build
(builder + runtime stages), creates a non-root user, and the UID
matches the podSecurityContext.runAsUser in values.yaml.
"""
dockerfile_path: Path = _k8s_dir().parent / "Dockerfile.server"
assert dockerfile_path.exists(), f"Dockerfile.server not found at {dockerfile_path}"
dockerfile_content: str = dockerfile_path.read_text(encoding="utf-8")
# Multi-stage build check
builder_pattern: re.Pattern[str] = re.compile(
r"^FROM\s+.*\s+AS\s+builder", re.MULTILINE
)
assert builder_pattern.search(dockerfile_content), (
"Dockerfile.server missing builder stage (expected FROM ... AS builder)"
)
expose_pattern: re.Pattern[str] = re.compile(r"^EXPOSE\s+8000$", re.MULTILINE)
assert expose_pattern.search(dockerfile_content), (
"Dockerfile.server should EXPOSE 8000"
)
user_pattern: re.Pattern[str] = re.compile(r"^USER\s+appuser$", re.MULTILINE)
assert user_pattern.search(dockerfile_content), (
"Dockerfile.server should switch to appuser"
)
# Cross-file: verify non-root UID matches values.yaml podSecurityContext
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
values: dict[str, Any] = yaml.safe_load(f)
run_as_user: int = values["podSecurityContext"]["runAsUser"]
assert run_as_user == 1000, (
f"podSecurityContext.runAsUser should be 1000, got {run_as_user}"
)
uid_pattern: re.Pattern[str] = re.compile(
rf"useradd.*-u\s+{run_as_user}", re.MULTILINE
)
assert uid_pattern.search(dockerfile_content), (
f"Dockerfile.server UID should match values.yaml runAsUser ({run_as_user})"
)
return "Dockerfile multi-stage OK"
def validate_service_defaults() -> str:
"""Validate service defaults are consistent across values and template.
Cross-file check: verifies service section defaults in values.yaml
and that the service template references the correct values.
"""
values_path: Path = _k8s_dir() / "values.yaml"
with open(values_path, encoding="utf-8") as f:
data: dict[str, Any] = yaml.safe_load(f)
service: dict[str, Any] = data["service"]
assert service["type"] == "ClusterIP", f"Expected ClusterIP, got {service['type']}"
assert service["port"] == 8000, f"Expected port 8000, got {service['port']}"
# Cross-file: service template should reference values
service_path: Path = _k8s_dir() / "templates" / "service.yaml"
service_content: str = service_path.read_text(encoding="utf-8")
assert ".Values.service.type" in service_content, (
"Service template should reference .Values.service.type"
)
assert ".Values.service.port" in service_content, (
"Service template should reference .Values.service.port"
)
return "Service defaults OK"
# Typed dispatch dictionary for command-line invocation.
# Using an explicit typed dict instead of globals() dispatch to ensure
# Pyright can verify the return type of each function.
_COMMANDS: dict[str, Callable[[], str]] = {
"validate_cross_file_port_consistency": validate_cross_file_port_consistency,
"validate_secrets_structure": validate_secrets_structure,
"validate_deployment_command_args": validate_deployment_command_args,
"validate_envfrom_wiring": validate_envfrom_wiring,
"validate_redis_host_helper": validate_redis_host_helper,
"validate_probe_config": validate_probe_config,
"validate_image_defaults": validate_image_defaults,
"validate_dockerfile_multistage_nonroot": validate_dockerfile_multistage_nonroot,
"validate_service_defaults": validate_service_defaults,
"validate_redis_disabled_rendering": validate_redis_disabled_rendering,
"validate_database_existing_secret_rendering": (
validate_database_existing_secret_rendering
),
"validate_redis_enabled_existing_secret_rendering": (
validate_redis_enabled_existing_secret_rendering
),
"validate_redis_enabled_inline_password_rendering": (
validate_redis_enabled_inline_password_rendering
),
"validate_redis_enabled_bitnami_fallback_rendering": (
validate_redis_enabled_bitnami_fallback_rendering
),
"validate_ingress_tls_rendering_success": validate_ingress_tls_rendering_success,
}
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Usage: helper_k8s_helm_chart.py <{'|'.join(_COMMANDS)}>")
sys.exit(1)
func_name: str = sys.argv[1]
func = _COMMANDS.get(func_name)
if func is None:
print(f"Unknown function: {func_name}")
sys.exit(1)
result: str = func()
print(result)