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
## 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>
679 lines
28 KiB
Python
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)
|