"""Offline chart contract tests: python3 -m unittest discover -s helm/tests -v. Requires Helm on PATH and PyYAML. Never connects to a cluster. """ import json from pathlib import Path import subprocess import unittest import yaml CHART = Path(__file__).resolve().parents[1] def render(values): return subprocess.run( ["helm", "template", "test", str(CHART), "--namespace", "retl-test", "--set", "projectRoot=/tmp/jitsu", "-f", "-"], input=yaml.safe_dump(values), text=True, capture_output=True, check=False, ) def enabled(**overrides): return {"reverseEtl": { "enabled": True, "runnerImage": "retl:test", "runtimeSecret": "retl-runtime", **overrides, }} class ReverseEtlChartTest(unittest.TestCase): def manifests(self, values): result = render(values) self.assertEqual(result.returncode, 0, result.stderr) return {(doc["kind"], doc["metadata"]["name"]): doc for doc in yaml.safe_load_all(result.stdout) if doc} def controller_env(self, docs): env = docs[("Deployment", "syncctl")]["spec"]["template"]["spec"]["containers"][0]["env"] self.assertEqual(len(env), len({item["name"] for item in env})) return {item["name"]: item.get("value") for item in env} def test_disabled_by_default(self): docs = self.manifests({}) self.assertEqual(self.controller_env(docs)["SYNCCTL_REVERSE_ENABLED"], "false") self.assertNotIn(("ServiceAccount", "test-retl-runner"), docs) self.assertNotIn(("RoleBinding", "test-retl-runner"), docs) def test_controller_override_is_preserved_when_restacked(self): docs = self.manifests({**enabled(controllerSecret="retl-controller"), "syncctlProjectRoot": "/tmp/retl-worktree"}) controller = docs[("Deployment", "syncctl")]["spec"]["template"]["spec"] self.assertEqual(controller["containers"][0]["envFrom"][-1], {"secretRef": {"name": "retl-controller"}}) project = next(volume for volume in controller["volumes"] if volume["name"] == "project") self.assertEqual(project["hostPath"]["path"], "/tmp/retl-worktree") for (kind, name), doc in docs.items(): if kind == "Deployment" and name != "syncctl": for container in doc["spec"]["template"]["spec"]["containers"]: self.assertNotIn({"secretRef": {"name": "retl-controller"}}, container.get("envFrom", [])) def test_syncctl_control_plane_stays_cluster_internal(self): for values in [{}, enabled()]: with self.subTest(values=values): service = self.manifests(values)[("Service", "syncctl")]["spec"] self.assertEqual(service["type"], "ClusterIP") self.assertNotIn("externalIPs", service) self.assertNotIn("loadBalancerIP", service) self.assertTrue(all("nodePort" not in port for port in service["ports"])) http = next(port for port in service["ports"] if port["name"] == "http") self.assertEqual(http["port"], 3043) self.assertEqual(http["targetPort"], "http") def test_extra_ca_is_opt_in_and_shared_with_runner(self): for ca in ["", "portless-ca"]: docs = self.manifests({"syncctlExtraCAConfigMap": ca}) pod = docs[("Deployment", "syncctl")]["spec"]["template"]["spec"] env = self.controller_env(docs) self.assertEqual(env["SYNCCTL_EXTRA_CA_CONFIG_MAP"], ca) volumes = [v for v in pod["volumes"] if v["name"] == "extra-ca"] mounts = [v for v in pod["containers"][0]["volumeMounts"] if v["name"] == "extra-ca"] if ca: self.assertEqual(volumes[0]["configMap"], { "name": ca, "items": [{"key": "ca.crt", "path": "ca.crt"}]}) self.assertTrue(mounts[0]["readOnly"]) self.assertEqual(env["SSL_CERT_FILE"], mounts[0]["mountPath"] + "/ca.crt") else: self.assertEqual(volumes, []) self.assertEqual(mounts, []) self.assertNotIn("SSL_CERT_FILE", env) def test_enabled_resources_and_identity(self): annotations = {"iam.gke.io/gcp-service-account": "runner@example.iam.gserviceaccount.com"} docs = self.manifests(enabled(serviceAccount={"annotations": annotations}, resources={"limits": {"ephemeral-storage": "8Gi"}}, scratchSizeLimit="6Gi")) env = self.controller_env(docs) self.assertEqual(env["SYNCCTL_REVERSE_ENABLED"], "true") self.assertEqual(env["SYNCCTL_REVERSE_RUNNER_IMAGE"], "retl:test") self.assertEqual(env["SYNCCTL_REVERSE_RUNTIME_SECRET"], "retl-runtime") self.assertEqual(env["SYNCCTL_REVERSE_SERVICE_ACCOUNT"], "test-retl-runner") self.assertEqual(env["SYNCCTL_REVERSE_SCRATCH_SIZE_LIMIT"], "6Gi") resources = json.loads(env["SYNCCTL_REVERSE_RUNNER_RESOURCES"]) self.assertEqual(resources["limits"]["ephemeral-storage"], "8Gi") self.assertEqual(resources["requests"]["memory"], "256Mi") self.assertEqual(docs[("ServiceAccount", "test-retl-runner")]["metadata"]["annotations"], annotations) self.assertNotIn("annotations", docs[("ServiceAccount", "sync-pod")]["metadata"]) self.assertEqual(docs[("Role", "test-retl-runner")]["rules"], [{ "apiGroups": ["coordination.k8s.io"], "resources": ["leases"], "verbs": ["get", "create", "update", "delete"], }]) self.assertEqual(docs[("RoleBinding", "test-retl-runner")]["subjects"], [{ "kind": "ServiceAccount", "name": "test-retl-runner", "namespace": "retl-test", }]) # The runtime Secret is referenced by name; credentials never enter Helm values/manifests. self.assertNotIn(("Secret", "retl-runtime"), docs) def test_external_service_account(self): docs = self.manifests(enabled(serviceAccount={"create": False, "name": "external-runner"})) self.assertNotIn(("ServiceAccount", "external-runner"), docs) self.assertEqual(self.controller_env(docs)["SYNCCTL_REVERSE_SERVICE_ACCOUNT"], "external-runner") self.assertEqual(docs[("RoleBinding", "test-retl-runner")]["subjects"][0]["name"], "external-runner") def test_extra_controller_secret_is_scoped_and_ordered(self): docs = self.manifests({**enabled(), "syncctlExtraEnvFrom": [ {"secretRef": {"name": "jitsu-retl-controller"}}, ]}) for (kind, name), doc in docs.items(): if kind != "Deployment": continue for container in doc["spec"]["template"]["spec"]["containers"]: sources = container.get("envFrom", []) refs = [source.get("secretRef", {}).get("name") for source in sources] if name == "syncctl": self.assertEqual(refs, ["jitsu-secrets", "jitsu-deps-urls", "jitsu-retl-controller"]) else: self.assertNotIn("jitsu-retl-controller", refs) def test_eks_annotation(self): annotations = {"eks.amazonaws.com/role-arn": "arn:aws:iam::123456789012:role/retl"} docs = self.manifests(enabled(serviceAccount={"name": "aws-runner", "annotations": annotations})) self.assertEqual(docs[("ServiceAccount", "aws-runner")]["metadata"]["annotations"], annotations) def test_invalid_settings_fail_render(self): for values, message in [ (enabled(runnerImage=""), "runnerImage is required"), (enabled(runtimeSecret=""), "runtimeSecret is required"), (enabled(serviceAccount={"create": False}), "name is required"), (enabled(serviceAccount={"create": False, "name": "existing", "annotations": {"foo": "bar"}}), "annotate the existing"), (enabled(serviceAccount={"name": "sync-pod"}), "dedicated service account"), ({**enabled(), "scaling": {"syncctl": {"replicas": 0}}}, "requires scaling.syncctl"), (enabled(enabled="true"), "boolean"), (enabled(scratchSizeLimit="many"), "scratchSizeLimit"), (enabled(resources={"limits": {"gpu": "1"}}), "gpu"), (enabled(serviceAccount={"annotations": {"foo": True}}), "string"), ]: with self.subTest(values=values): result = render(values) self.assertNotEqual(result.returncode, 0) self.assertIn(message, result.stderr) def test_env_overrides_cannot_bypass_validation(self): for scope in ["common", "syncctl"]: for prefix in ["", "SYNCCTL_"]: result = render({"env": {scope: {prefix + "REVERSE_ENABLED": True}}}) self.assertNotEqual(result.returncode, 0) self.assertIn("via reverseEtl", result.stderr) def test_schema_hook_never_accepts_data_loss(self): docs = self.manifests({}) hooks = [doc for (kind, _), doc in docs.items() if kind == "Job"] commands = "\n".join(" ".join(container["command"]) for job in hooks for container in job["spec"]["template"]["spec"]["containers"]) self.assertIn("prisma db push", commands) self.assertNotIn("--accept-data-loss", commands) if __name__ == "__main__": unittest.main()