freemo
e5398e3cc7
CI / lint (push) Successful in 17s
CI / build (push) Successful in 16s
CI / security (push) Has been cancelled
CI / docker (push) Has been cancelled
CI / typecheck (push) Has been cancelled
CI / unit_tests (push) Has been cancelled
CI / coverage (push) Has been cancelled
CI / integration_tests (push) Has been cancelled
CI / e2e_tests (push) Has been cancelled
CI / status-check (push) Has been cancelled
CI / benchmark-regression (push) Has been cancelled
CI / benchmark-publish (push) Has been cancelled
CI / quality (push) Has been cancelled
CI / helm (push) Has been cancelled
The `cleveragents server serve` CLI subcommand already exists in server.py and Dockerfile.server already uses `python -m cleveragents` as its ENTRYPOINT with `server serve` as the CMD. Add BDD scenarios to k8s_helm_chart.feature that explicitly verify: - Dockerfile.server ENTRYPOINT uses `python -m cleveragents` - Dockerfile.server CMD includes the `server serve` subcommand tokens Add corresponding step definitions to k8s_helm_chart_steps.py. Closes #1088 Co-authored-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me> Co-committed-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
581 lines
22 KiB
Python
581 lines
22 KiB
Python
"""Step definitions for the Kubernetes Helm Chart feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
|
|
def _project_root() -> Path:
|
|
"""Return the project root directory.
|
|
|
|
Walks up from this file until we find pyproject.toml.
|
|
"""
|
|
current: Path = Path(__file__).resolve()
|
|
for parent in current.parents:
|
|
if (parent / "pyproject.toml").exists():
|
|
return parent
|
|
raise RuntimeError("Could not find project root (no pyproject.toml found)")
|
|
|
|
|
|
def _skip_if_helm_missing(context: Context, *, reason: str) -> bool:
|
|
"""Skip the current scenario when Helm CLI is unavailable.
|
|
|
|
Returns ``True`` when Helm is missing (scenario was skipped).
|
|
"""
|
|
if shutil.which("helm") is None:
|
|
context.scenario.skip(reason)
|
|
return True
|
|
return False
|
|
|
|
|
|
def _render_chart(context: Context, *, set_overrides: list[str] | None = None) -> None:
|
|
"""Run ``helm template`` and store the subprocess result on context."""
|
|
if _skip_if_helm_missing(
|
|
context,
|
|
reason="Helm CLI is required for rendered Helm behavior scenarios",
|
|
):
|
|
context.helm_render_result = None
|
|
return
|
|
|
|
command: list[str] = [
|
|
"helm",
|
|
"template",
|
|
"cleveragents",
|
|
str(context.project_root / "k8s"),
|
|
"--dependency-update",
|
|
]
|
|
for item in set_overrides or []:
|
|
command.extend(["--set", item])
|
|
|
|
context.helm_render_result = subprocess.run(
|
|
command,
|
|
cwd=str(context.project_root),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
@given("the project root directory")
|
|
def step_given_project_root(context: Context) -> None:
|
|
context.project_root = _project_root()
|
|
|
|
|
|
@when("I read the Helm chart metadata")
|
|
def step_read_chart_metadata(context: Context) -> None:
|
|
chart_path: Path = context.project_root / "k8s" / "Chart.yaml"
|
|
assert chart_path.exists(), f"Chart.yaml not found at {chart_path}"
|
|
with open(chart_path, encoding="utf-8") as f:
|
|
context.chart_metadata = yaml.safe_load(f)
|
|
|
|
|
|
@then('the chart API version should be "{version}"')
|
|
def step_chart_api_version(context: Context, version: str) -> None:
|
|
assert context.chart_metadata.get("apiVersion") == version
|
|
|
|
|
|
@then('the chart name should be "{name}"')
|
|
def step_chart_name(context: Context, name: str) -> None:
|
|
assert context.chart_metadata.get("name") == name
|
|
|
|
|
|
@then('the chart type should be "{chart_type}"')
|
|
def step_chart_type(context: Context, chart_type: str) -> None:
|
|
assert context.chart_metadata.get("type") == chart_type
|
|
|
|
|
|
@then("the chart version should be set")
|
|
def step_chart_version_set(context: Context) -> None:
|
|
assert "version" in context.chart_metadata
|
|
assert context.chart_metadata.get("version")
|
|
|
|
|
|
@then('the chart should have a dependency named "{dep_name}"')
|
|
def step_chart_has_dependency(context: Context, dep_name: str) -> None:
|
|
deps: list[dict[str, str]] = context.chart_metadata.get("dependencies", [])
|
|
dep_names: list[str] = [d.get("name", "") for d in deps]
|
|
assert dep_name in dep_names, f"Dependency '{dep_name}' not found in {dep_names}"
|
|
for dep in deps:
|
|
if dep.get("name") == dep_name:
|
|
context.matched_dependency = dep
|
|
break
|
|
|
|
|
|
@then('the Redis dependency should be conditional on "{condition}"')
|
|
def step_redis_dependency_condition(context: Context, condition: str) -> None:
|
|
dep: dict[str, str] = context.matched_dependency
|
|
assert dep.get("condition") == condition, (
|
|
f"Expected condition '{condition}', got '{dep.get('condition')}'"
|
|
)
|
|
|
|
|
|
@when("I read the Helm chart values")
|
|
def step_read_chart_values(context: Context) -> None:
|
|
values_path: Path = context.project_root / "k8s" / "values.yaml"
|
|
assert values_path.exists(), f"values.yaml not found at {values_path}"
|
|
with open(values_path, encoding="utf-8") as f:
|
|
context.chart_values = yaml.safe_load(f)
|
|
|
|
|
|
@then('the values should contain key "{key}"')
|
|
def step_values_contains_key(context: Context, key: str) -> None:
|
|
assert key in context.chart_values, (
|
|
f"Key '{key}' not found in values.yaml. "
|
|
f"Available keys: {list(context.chart_values.keys())}"
|
|
)
|
|
|
|
|
|
@then("the default replica count should be {count:d}")
|
|
def step_default_replica_count(context: Context, count: int) -> None:
|
|
assert context.chart_values.get("replicaCount") == count
|
|
|
|
|
|
@then("the resources should have CPU and memory limits")
|
|
def step_resources_have_limits(context: Context) -> None:
|
|
resources: dict[str, Any] = context.chart_values.get("resources", {})
|
|
assert "limits" in resources, "Missing 'limits' in resources"
|
|
limits: dict[str, str] = resources.get("limits", {})
|
|
assert "cpu" in limits, "Missing 'cpu' in resource limits"
|
|
assert "memory" in limits, "Missing 'memory' in resource limits"
|
|
|
|
|
|
@then("the resources should have CPU and memory requests")
|
|
def step_resources_have_requests(context: Context) -> None:
|
|
resources: dict[str, Any] = context.chart_values.get("resources", {})
|
|
assert "requests" in resources, "Missing 'requests' in resources"
|
|
reqs: dict[str, str] = resources.get("requests", {})
|
|
assert "cpu" in reqs, "Missing 'cpu' in resource requests"
|
|
assert "memory" in reqs, "Missing 'memory' in resource requests"
|
|
|
|
|
|
@then("the ingress should have an enabled toggle")
|
|
def step_ingress_has_enabled(context: Context) -> None:
|
|
ingress: dict[str, Any] = context.chart_values.get("ingress", {})
|
|
assert "enabled" in ingress, "Missing 'enabled' toggle in ingress"
|
|
assert isinstance(ingress.get("enabled"), bool), "'enabled' should be a boolean"
|
|
|
|
|
|
@then("the ingress should have a TLS configuration section")
|
|
def step_ingress_has_tls(context: Context) -> None:
|
|
ingress: dict[str, Any] = context.chart_values.get("ingress", {})
|
|
assert "tls" in ingress, "Missing 'tls' section in ingress"
|
|
|
|
|
|
@then("the Redis configuration should have an enabled toggle")
|
|
def step_redis_has_enabled(context: Context) -> None:
|
|
redis: dict[str, Any] = context.chart_values.get("redis", {})
|
|
assert "enabled" in redis, "Missing 'enabled' toggle in redis"
|
|
assert isinstance(redis.get("enabled"), bool), "'enabled' should be a boolean"
|
|
|
|
|
|
@then("the Redis configuration should have authentication settings")
|
|
def step_redis_has_auth(context: Context) -> None:
|
|
redis: dict[str, Any] = context.chart_values.get("redis", {})
|
|
assert "auth" in redis, "Missing 'auth' section in redis"
|
|
auth: dict[str, Any] = redis.get("auth", {})
|
|
assert "enabled" in auth, "Missing 'enabled' in redis auth"
|
|
|
|
|
|
@then("the server config should have host setting")
|
|
def step_server_has_host(context: Context) -> None:
|
|
server: dict[str, Any] = context.chart_values.get("server", {})
|
|
assert "host" in server, "Missing 'host' in server config"
|
|
|
|
|
|
@then("the server config should have port setting")
|
|
def step_server_has_port(context: Context) -> None:
|
|
server: dict[str, Any] = context.chart_values.get("server", {})
|
|
assert "port" in server, "Missing 'port' in server config"
|
|
|
|
|
|
@then("the server config should have workers setting")
|
|
def step_server_has_workers(context: Context) -> None:
|
|
server: dict[str, Any] = context.chart_values.get("server", {})
|
|
assert "workers" in server, "Missing 'workers' in server config"
|
|
|
|
|
|
@then("the database config should support existing secrets")
|
|
def step_database_supports_secrets(context: Context) -> None:
|
|
database: dict[str, Any] = context.chart_values.get("database", {})
|
|
assert "existingSecret" in database, "Missing 'existingSecret' in database config"
|
|
assert "existingSecretKey" in database, (
|
|
"Missing 'existingSecretKey' in database config"
|
|
)
|
|
|
|
|
|
@then("the pod security context should enforce non-root execution")
|
|
def step_pod_security_nonroot(context: Context) -> None:
|
|
pod_sc: dict[str, Any] = context.chart_values.get("podSecurityContext", {})
|
|
assert pod_sc.get("runAsNonRoot") is True, (
|
|
"podSecurityContext.runAsNonRoot should be true"
|
|
)
|
|
assert pod_sc.get("runAsUser") == 1000, (
|
|
"podSecurityContext.runAsUser should be 1000"
|
|
)
|
|
|
|
|
|
@then("the container security context should drop all capabilities")
|
|
def step_container_security_caps(context: Context) -> None:
|
|
sc: dict[str, Any] = context.chart_values.get("securityContext", {})
|
|
caps: dict[str, list[str]] = sc.get("capabilities", {})
|
|
assert "ALL" in caps.get("drop", []), "securityContext should drop ALL capabilities"
|
|
|
|
|
|
@then("the container security context should have a seccomp profile")
|
|
def step_container_security_seccomp(context: Context) -> None:
|
|
sc: dict[str, Any] = context.chart_values.get("securityContext", {})
|
|
seccomp: dict[str, str] = sc.get("seccompProfile", {})
|
|
assert seccomp.get("type") == "RuntimeDefault", (
|
|
f"Expected seccompProfile type 'RuntimeDefault', got '{seccomp.get('type')}'"
|
|
)
|
|
|
|
|
|
@then("the container security context should set readOnlyRootFilesystem to true")
|
|
def step_container_security_readonly_root(context: Context) -> None:
|
|
sc: dict[str, Any] = context.chart_values.get("securityContext", {})
|
|
assert sc.get("readOnlyRootFilesystem") is True, (
|
|
"securityContext.readOnlyRootFilesystem should be true"
|
|
)
|
|
|
|
|
|
@then("the container security context should set allowPrivilegeEscalation to false")
|
|
def step_container_security_no_privilege_escalation(context: Context) -> None:
|
|
sc: dict[str, Any] = context.chart_values.get("securityContext", {})
|
|
assert sc.get("allowPrivilegeEscalation") is False, (
|
|
"securityContext.allowPrivilegeEscalation should be false"
|
|
)
|
|
|
|
|
|
@then('the chart file "{filepath}" should exist')
|
|
def step_chart_file_should_exist(context: Context, filepath: str) -> None:
|
|
full_path: Path = context.project_root / filepath
|
|
assert full_path.exists(), f"File not found: {full_path}"
|
|
|
|
|
|
@when('I read the template "{filepath}"')
|
|
def step_read_template(context: Context, filepath: str) -> None:
|
|
template_path: Path = context.project_root / filepath
|
|
assert template_path.exists(), f"Template not found at {template_path}"
|
|
context.template_content = template_path.read_text(encoding="utf-8")
|
|
|
|
|
|
@then('the template should contain "{text}"')
|
|
def step_template_should_contain(context: Context, text: str) -> None:
|
|
pattern: re.Pattern[str] = re.compile(re.escape(text), re.MULTILINE)
|
|
assert pattern.search(context.template_content), (
|
|
f"Template does not contain '{text}'"
|
|
)
|
|
|
|
|
|
@then('the template should match pattern "{pattern}"')
|
|
def step_template_should_match_pattern(context: Context, pattern: str) -> None:
|
|
regex: re.Pattern[str] = re.compile(pattern, re.MULTILINE)
|
|
assert regex.search(context.template_content), (
|
|
f"Template does not match pattern '{pattern}'"
|
|
)
|
|
|
|
|
|
@when("I read the server Dockerfile")
|
|
def step_read_server_dockerfile(context: Context) -> None:
|
|
dockerfile_path: Path = context.project_root / "Dockerfile.server"
|
|
assert dockerfile_path.exists(), f"Dockerfile.server not found at {dockerfile_path}"
|
|
context.dockerfile_content = dockerfile_path.read_text(encoding="utf-8")
|
|
|
|
|
|
@then("it should have a builder stage")
|
|
def step_dockerfile_has_builder_stage(context: Context) -> None:
|
|
pattern: re.Pattern[str] = re.compile(r"^FROM\s+.*\s+AS\s+builder", re.MULTILINE)
|
|
assert pattern.search(context.dockerfile_content), (
|
|
"Dockerfile.server missing builder stage (expected FROM ... AS builder)"
|
|
)
|
|
|
|
|
|
@then("it should have a runtime stage")
|
|
def step_dockerfile_has_runtime_stage(context: Context) -> None:
|
|
lines: list[str] = context.dockerfile_content.splitlines()
|
|
from_count: int = sum(1 for line in lines if re.match(r"^FROM\s+", line))
|
|
assert from_count >= 2, (
|
|
f"Expected at least 2 FROM statements for multi-stage build, found {from_count}"
|
|
)
|
|
|
|
|
|
@then("it should create a non-root user with uid 1000")
|
|
def step_dockerfile_nonroot_user(context: Context) -> None:
|
|
pattern: re.Pattern[str] = re.compile(r"useradd.*-u\s+1000")
|
|
assert pattern.search(context.dockerfile_content), (
|
|
"Dockerfile.server should create a user with uid 1000 via useradd"
|
|
)
|
|
assert "USER appuser" in context.dockerfile_content, (
|
|
"Dockerfile.server should switch to appuser"
|
|
)
|
|
|
|
|
|
@then("it should expose port 8000")
|
|
def step_dockerfile_expose_port(context: Context) -> None:
|
|
pattern: re.Pattern[str] = re.compile(r"^EXPOSE\s+8000", re.MULTILINE)
|
|
assert pattern.search(context.dockerfile_content), (
|
|
"Dockerfile.server should expose port 8000"
|
|
)
|
|
|
|
|
|
@then("it should not use unpinned uv latest tag")
|
|
def step_dockerfile_pinned_uv(context: Context) -> None:
|
|
assert "uv:latest" not in context.dockerfile_content, (
|
|
"Dockerfile.server should pin the uv image version, not use :latest"
|
|
)
|
|
|
|
|
|
@then("the uv image reference should use a specific semver tag")
|
|
def step_dockerfile_uv_semver(context: Context) -> None:
|
|
pattern: re.Pattern[str] = re.compile(r"uv:\d+\.\d+\.\d+", re.MULTILINE)
|
|
assert pattern.search(context.dockerfile_content), (
|
|
"Dockerfile.server should pin uv to a specific semver version "
|
|
"(e.g., uv:0.8.0), not a floating tag"
|
|
)
|
|
|
|
|
|
@then("the Dockerfile should remove uv after installation")
|
|
def step_dockerfile_removes_uv(context: Context) -> None:
|
|
pattern: re.Pattern[str] = re.compile(
|
|
r"rm\s+.*(/usr/local/bin/uv|uv)", re.MULTILINE
|
|
)
|
|
assert pattern.search(context.dockerfile_content), (
|
|
"Dockerfile.server should remove uv from runtime image after "
|
|
"package installation to reduce attack surface"
|
|
)
|
|
|
|
|
|
@then('the ENTRYPOINT should use "{entrypoint}"')
|
|
def step_dockerfile_entrypoint(context: Context, entrypoint: str) -> None:
|
|
"""Assert that the Dockerfile ENTRYPOINT contains the expected command."""
|
|
pattern: re.Pattern[str] = re.compile(r"^ENTRYPOINT\s+\[.*\]", re.MULTILINE)
|
|
match = pattern.search(context.dockerfile_content)
|
|
assert match is not None, "Dockerfile.server is missing an ENTRYPOINT instruction"
|
|
entrypoint_line = match.group(0)
|
|
# Verify each token of the expected entrypoint appears in the ENTRYPOINT line
|
|
for token in entrypoint.split():
|
|
assert token in entrypoint_line, (
|
|
f"Expected ENTRYPOINT to contain '{token}', but got: {entrypoint_line!r}"
|
|
)
|
|
|
|
|
|
@then('the CMD should include "{cmd_a}" and "{cmd_b}"')
|
|
def step_dockerfile_cmd_includes(context: Context, cmd_a: str, cmd_b: str) -> None:
|
|
"""Assert that the Dockerfile CMD contains the expected subcommand tokens."""
|
|
pattern: re.Pattern[str] = re.compile(r"^CMD\s+\[.*\]", re.MULTILINE)
|
|
match = pattern.search(context.dockerfile_content)
|
|
assert match is not None, "Dockerfile.server is missing a CMD instruction"
|
|
cmd_line = match.group(0)
|
|
assert cmd_a in cmd_line, (
|
|
f"Expected CMD to contain '{cmd_a}', but got: {cmd_line!r}"
|
|
)
|
|
assert cmd_b in cmd_line, (
|
|
f"Expected CMD to contain '{cmd_b}', but got: {cmd_line!r}"
|
|
)
|
|
|
|
|
|
@when("I read the deployment README")
|
|
def step_read_deployment_readme(context: Context) -> None:
|
|
readme_path: Path = context.project_root / "k8s" / "README.md"
|
|
assert readme_path.exists(), f"k8s/README.md not found at {readme_path}"
|
|
context.readme_content = readme_path.read_text(encoding="utf-8")
|
|
|
|
|
|
@then("it should contain quick start instructions")
|
|
def step_readme_has_quickstart(context: Context) -> None:
|
|
assert "Quick Start" in context.readme_content, (
|
|
"README should contain 'Quick Start' section"
|
|
)
|
|
|
|
|
|
@then("it should contain TLS configuration documentation")
|
|
def step_readme_has_tls_docs(context: Context) -> None:
|
|
assert "TLS" in context.readme_content, "README should document TLS configuration"
|
|
|
|
|
|
@then("it should contain Redis configuration documentation")
|
|
def step_readme_has_redis_docs(context: Context) -> None:
|
|
assert "Redis" in context.readme_content, (
|
|
"README should document Redis configuration"
|
|
)
|
|
|
|
|
|
@then("it should contain scaling instructions")
|
|
def step_readme_has_scaling_docs(context: Context) -> None:
|
|
assert (
|
|
"Scaling" in context.readme_content or "replicaCount" in context.readme_content
|
|
), "README should document scaling instructions"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Probes validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the liveness probe should target the live endpoint")
|
|
def step_liveness_probe_live(context: Context) -> None:
|
|
probe: dict[str, Any] = context.chart_values.get("livenessProbe", {})
|
|
http_get: dict[str, Any] = probe.get("httpGet", {})
|
|
assert http_get.get("path") == "/live", (
|
|
f"Expected liveness probe path /live, got {http_get.get('path')}"
|
|
)
|
|
|
|
|
|
@then("the readiness probe should target the ready endpoint")
|
|
def step_readiness_probe_ready(context: Context) -> None:
|
|
probe: dict[str, Any] = context.chart_values.get("readinessProbe", {})
|
|
http_get: dict[str, Any] = probe.get("httpGet", {})
|
|
assert http_get.get("path") == "/ready", (
|
|
f"Expected readiness probe path /ready, got {http_get.get('path')}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Image defaults validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the image pull policy should be "{policy}"')
|
|
def step_image_pull_policy(context: Context, policy: str) -> None:
|
|
image: dict[str, Any] = context.chart_values.get("image", {})
|
|
assert image.get("pullPolicy") == policy, (
|
|
f"Expected pullPolicy '{policy}', got '{image.get('pullPolicy')}'"
|
|
)
|
|
|
|
|
|
@then("the image tag should default to empty for appVersion fallback")
|
|
def step_image_tag_default_empty(context: Context) -> None:
|
|
image: dict[str, Any] = context.chart_values.get("image", {})
|
|
assert image.get("tag") == "", (
|
|
f"Expected empty tag for appVersion fallback, got '{image.get('tag')}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Service defaults validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the service type should be "{svc_type}"')
|
|
def step_service_type(context: Context, svc_type: str) -> None:
|
|
service: dict[str, Any] = context.chart_values.get("service", {})
|
|
assert service.get("type") == svc_type, (
|
|
f"Expected service type '{svc_type}', got '{service.get('type')}'"
|
|
)
|
|
|
|
|
|
@then("the service port should be {port:d}")
|
|
def step_service_port(context: Context, port: int) -> None:
|
|
service: dict[str, Any] = context.chart_values.get("service", {})
|
|
assert service.get("port") == port, (
|
|
f"Expected service port {port}, got {service.get('port')}"
|
|
)
|
|
|
|
|
|
@when("I render the Helm chart without database configuration")
|
|
def step_render_without_database_configuration(context: Context) -> None:
|
|
_render_chart(context)
|
|
|
|
|
|
@then("the render should fail with required database configuration error")
|
|
def step_render_fails_database_required(context: Context) -> None:
|
|
result = context.helm_render_result
|
|
if result is None:
|
|
return
|
|
assert result.returncode != 0, "Expected helm template to fail without DB config"
|
|
combined_output = f"{result.stdout}\n{result.stderr}"
|
|
assert "database configuration is required" in combined_output, (
|
|
"Expected database required fail-fast error in Helm output"
|
|
)
|
|
|
|
|
|
@when("I render the Helm chart with ingress enabled and no TLS")
|
|
def step_render_with_ingress_enabled_no_tls(context: Context) -> None:
|
|
_render_chart(
|
|
context,
|
|
set_overrides=[
|
|
"database.url=postgresql+asyncpg://user:pass@db:5432/cleveragents",
|
|
"ingress.enabled=true",
|
|
"ingress.allowInsecure=false",
|
|
],
|
|
)
|
|
|
|
|
|
@then("the render should fail with required ingress TLS configuration error")
|
|
def step_render_fails_ingress_tls_required(context: Context) -> None:
|
|
result = context.helm_render_result
|
|
if result is None:
|
|
return
|
|
assert result.returncode != 0, (
|
|
"Expected helm template to fail when ingress is enabled without TLS"
|
|
)
|
|
combined_output = f"{result.stdout}\n{result.stderr}"
|
|
assert "ingress.tls is required" in combined_output, (
|
|
"Expected ingress TLS required fail-fast error in Helm output"
|
|
)
|
|
|
|
|
|
@when("I render the Helm chart with ingress enabled and insecure opt-out enabled")
|
|
def step_render_with_ingress_insecure_opt_out(context: Context) -> None:
|
|
_render_chart(
|
|
context,
|
|
set_overrides=[
|
|
"database.url=postgresql+asyncpg://user:pass@db:5432/cleveragents",
|
|
"ingress.enabled=true",
|
|
"ingress.allowInsecure=true",
|
|
],
|
|
)
|
|
|
|
|
|
@then("the ingress render should succeed in insecure dev mode")
|
|
def step_render_succeeds_insecure_ingress_opt_out(context: Context) -> None:
|
|
result = context.helm_render_result
|
|
if result is None:
|
|
return
|
|
assert result.returncode == 0, (
|
|
"Expected helm template to succeed when ingress.allowInsecure=true"
|
|
)
|
|
combined_output = f"{result.stdout}\n{result.stderr}"
|
|
assert "kind: Ingress" in combined_output, (
|
|
"Expected rendered output to include an Ingress resource"
|
|
)
|
|
|
|
|
|
@when("I render the Helm chart with ingress enabled and TLS configured")
|
|
def step_render_with_ingress_tls_configured(context: Context) -> None:
|
|
_render_chart(
|
|
context,
|
|
set_overrides=[
|
|
"database.url=postgresql+asyncpg://user:pass@db:5432/cleveragents",
|
|
"ingress.enabled=true",
|
|
"ingress.allowInsecure=false",
|
|
"ingress.tls[0].secretName=cleveragents-tls",
|
|
"ingress.tls[0].hosts[0]=cleveragents.example.com",
|
|
],
|
|
)
|
|
|
|
|
|
@then("the ingress render should succeed with TLS enabled")
|
|
def step_render_succeeds_with_ingress_tls(context: Context) -> None:
|
|
result = context.helm_render_result
|
|
if result is None:
|
|
return
|
|
assert result.returncode == 0, (
|
|
"Expected helm template to succeed when ingress TLS is configured"
|
|
)
|
|
combined_output = f"{result.stdout}\n{result.stderr}"
|
|
assert "kind: Ingress" in combined_output, (
|
|
"Expected rendered output to include an Ingress resource"
|
|
)
|
|
assert "secretName: cleveragents-tls" in combined_output, (
|
|
"Expected rendered Ingress to include configured TLS secret"
|
|
)
|