tegwick
dd15d8cf3f
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
69 lines
4.7 KiB
Python
69 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Source render integration tests; never contacts the cluster."""
|
|
import copy
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import subprocess
|
|
import unittest
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("twins", ROOT / "tools/gen_runtime_twins.py")
|
|
twins = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(twins)
|
|
|
|
def render(chart, values, namespace, name):
|
|
output = subprocess.check_output(["helm", "template", name, str(ROOT / "charts" / chart), "-f", str(ROOT / "values" / values), "--namespace", namespace], text=True)
|
|
return {obj["kind"]: obj for obj in yaml.safe_load_all(output) if obj}
|
|
|
|
def after(args, flag):
|
|
return [args[i + 1] for i, item in enumerate(args[:-1]) if item == flag]
|
|
|
|
class RuntimeTwinTests(unittest.TestCase):
|
|
def test_actual_nine_twin_render_security_and_content(self):
|
|
for pin in twins.PINS:
|
|
with self.subTest(pin=pin):
|
|
source = yaml.safe_load((ROOT / "values" / (pin + ".yaml")).read_text())
|
|
new = yaml.safe_load((ROOT / "values/access-engine" / (pin + ".yaml")).read_text())
|
|
self.assertEqual(new, twins.twin(source))
|
|
old_objects = render("flex-auth", pin + ".yaml", "flex-auth", source["name"])
|
|
objects = render("access-engine", "access-engine/" + pin + ".yaml", "access-engine", new["name"])
|
|
self.assertEqual(set(objects), set(old_objects))
|
|
for kind in ("Deployment", "Service", "ServiceAccount", "NetworkPolicy"):
|
|
self.assertEqual(objects[kind]["metadata"]["namespace"], "access-engine")
|
|
deployment = objects["Deployment"]
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
container = pod["containers"][0]
|
|
old_container = old_objects["Deployment"]["spec"]["template"]["spec"]["containers"][0]
|
|
self.assertEqual(container["image"], old_container["image"])
|
|
self.assertEqual(container["args"][:len(source["args"])], source["args"])
|
|
self.assertEqual(after(container["args"], "--caller-audience"), ["access-engine"])
|
|
self.assertEqual(after(container["args"], "--caller-binding"), after(old_container["args"], "--caller-binding"))
|
|
self.assertEqual(after(container["args"], "--readback-caller"), after(old_container["args"], "--readback-caller"))
|
|
self.assertEqual(pod["serviceAccountName"], new["name"])
|
|
mount = next(m["mountPath"] for m in container["volumeMounts"] if m["name"] == "access-engine-reviewer")
|
|
self.assertEqual(after(container["args"], "--caller-reviewer-token-file"), [mount + "/token"])
|
|
self.assertEqual(after(container["args"], "--caller-ca-file"), [mount + "/ca.crt"])
|
|
self.assertFalse(pod["automountServiceAccountToken"])
|
|
self.assertEqual(objects["ClusterRole"]["rules"], old_objects["ClusterRole"]["rules"])
|
|
binding = objects["ClusterRoleBinding"]
|
|
self.assertEqual(binding["subjects"], [{"kind": "ServiceAccount", "name": new["name"], "namespace": "access-engine"}])
|
|
self.assertEqual(binding["roleRef"]["name"], new["name"] + "-tokenreviewer")
|
|
self.assertEqual(objects["NetworkPolicy"]["spec"]["ingress"], old_objects["NetworkPolicy"]["spec"]["ingress"])
|
|
self.assertEqual(objects["NetworkPolicy"]["spec"]["egress"], old_objects["NetworkPolicy"]["spec"]["egress"])
|
|
self.assertEqual(objects["Service"]["spec"]["selector"], deployment["spec"]["selector"]["matchLabels"])
|
|
self.assertEqual(deployment["spec"]["selector"]["matchLabels"], {"app.kubernetes.io/name": new["name"]})
|
|
self.assertTrue(container["securityContext"]["readOnlyRootFilesystem"])
|
|
self.assertFalse(container["securityContext"]["allowPrivilegeEscalation"])
|
|
self.assertEqual(container["livenessProbe"], old_container["livenessProbe"])
|
|
self.assertEqual(container["readinessProbe"], old_container["readinessProbe"])
|
|
|
|
def test_refuse_unpinned_unenforced_and_shared_writer(self):
|
|
source = yaml.safe_load((ROOT / "values/user-engine.yaml").read_text())
|
|
for field, value in (("image", {"digest": "latest"}), ("callerAuth", {"mode": "disabled"}), ("outbox", {"enabled": True}), ("auditDelivery", {"enabled": True}), ("name", "other-pin")):
|
|
with self.subTest(field=field):
|
|
bad = copy.deepcopy(source); bad[field] = value
|
|
with self.assertRaises(ValueError): twins.twin(bad)
|
|
self.assertNotIn("namespace", source)
|
|
|
|
if __name__ == "__main__": unittest.main()
|