access-engine/tests/test_outbox_replicas_chart.py
tegwick dc042022f7
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Allow stopped single-writer outbox maintenance through Helm
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
2026-10-16 20:25:05 +02:00

65 lines
3.1 KiB
Python

#!/usr/bin/env python3
"""Actual Helm stop/start renders and unsafe replica value rejection; offline only."""
import copy
import json
from pathlib import Path
import subprocess
import tempfile
import yaml
ROOT = Path(__file__).resolve().parents[1]
def render(chart, value, encoding="yaml", success=True):
with tempfile.TemporaryDirectory() as directory:
command = ["helm", "template", "flex-auth-secrets-engine", str(ROOT / "charts" / chart),
"--namespace", "flex-auth", "-f", str(ROOT / "values/secrets-engine.yaml"),
"--set", "outbox.enabled=true", "--set", "outbox.source=flex-auth.secrets-engine",
"--set", "envelopeSigning.enabled=true", "--set", "envelopeSigning.kid=test-kid",
"--set", "envelopeSigning.secretName=proposed-seed", "--set", "envelopeSigning.secretKey=seed"]
if encoding == "yaml":
overlay = Path(directory) / "values.yaml"
overlay.write_text(yaml.safe_dump({"replicaCount": value}))
command += ["-f", str(overlay)]
elif encoding == "json":
command += ["--set-json", "replicaCount=" + json.dumps(value)]
elif encoding == "string":
command += ["--set-string", "replicaCount=" + str(value)]
else:
command += ["--set", "replicaCount=" + str(value)]
result = subprocess.run(command, capture_output=True, text=True)
if not success:
assert result.returncode != 0, (chart, encoding, value)
assert "replicaCount" in result.stderr, result.stderr
return
assert result.returncode == 0, result.stderr
documents = [d for d in yaml.safe_load_all(result.stdout) if d]
deployment = next(d for d in documents if d["kind"] == "Deployment")
assert deployment["spec"]["strategy"] == {"type": "Recreate"}
assert type(deployment["spec"]["replicas"]) is int
assert deployment["spec"]["replicas"] == value
return documents
count = 0
for chart in ("flex-auth", "access-engine"):
for encoding in ("yaml", "json", "cli"):
stopped = render(chart, 0, encoding)
ready = render(chart, 1, encoding)
normalized = copy.deepcopy(ready)
next(d for d in normalized if d["kind"] == "Deployment")["spec"]["replicas"] = 0
assert stopped == normalized, "stop must change replicas only, including identical pod projection/PVC"
count += 2
for encoding in ("yaml", "json"):
for value in (-1, 2, 1.5, 0.5, True, False, None, "0", "1", [], {}):
render(chart, value, encoding, success=False)
count += 1
# Helm loses lexical 0 vs0.0 and1 vs1.0 before template evaluation.
for value in (0.0, 1.0):
render(chart, value, encoding)
count += 1
for value in ("-1", "2", "0.5", "1.5", "true", "false", "null", "0.0", "1.0"):
render(chart, value, "cli", success=False)
count += 1
for value in ("0", "1"):
render(chart, value, "string", success=False)
count += 1
print(f"stopped/single-writer chart checks passed: {count} actual Helm renders")