import argparse import base64 import copy import hashlib import importlib.util import json from pathlib import Path import re import tempfile import unittest from unittest.mock import patch PACKAGE = Path(__file__).resolve().parents[1] def load_module(name, filename): spec = importlib.util.spec_from_file_location(name, PACKAGE / filename) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module ns = load_module("notification", "notification.py") storage = load_module("prepare_storage", "prepare-storage.py") def fixture(): config = ns.read_json(PACKAGE / "config.example.json") artifacts = ns.read_json(PACKAGE / "artifacts.example.json") artifacts["sourceCommit"] = "a" * 40 artifacts["sourceArchiveSha256"] = "b" * 64 for key, value in artifacts["images"].items(): value["image"] = "registry.hrynco.com/test/" + key + "@sha256:" + "c" * 64 if key in ("web", "worker", "migrator"): value["sourceArchiveSha256"] = artifacts["sourceArchiveSha256"] return config, artifacts class FakeCluster: def __init__(self, history=None, worker=0, worker_error=False, consumer_count=1, postgres_ready=True, postgres_exists=True, storage_error=False): self.schema = list(ns.MIGRATIONS if history is None else history) self.replicas = {"web": 1, "worker": worker, "postgres": 1} if not postgres_exists: self.replicas.pop("postgres") elif not postgres_ready: self.replicas["postgres"] = 0 self.events = [] self.worker_error = worker_error self.consumer_counts = list(consumer_count) if isinstance(consumer_count, list) else [consumer_count] self.postgres_ready = postgres_ready self.storage_error = storage_error self.applications = [] def get(self, kind, name=None, namespace=None): if kind in ("namespace", "secret"): return {"metadata": {"labels": ns.labels()}} if kind == "deployment" and name in self.replicas: value = {"metadata": {"generation": 1}, "spec": {"replicas": self.replicas[name]}} if name == "postgres": value["status"] = {"readyReplicas": int(self.postgres_ready), "updatedReplicas": int(self.postgres_ready), "observedGeneration": 1} return value return None def history(self): if not self.postgres_ready or self.replicas.get("postgres", 0) != 1: raise RuntimeError("cannot query stopped database") self.events.append(("history",)) return list(self.schema) def scale(self, component, replicas): self.replicas[component] = replicas self.events.append(("scale", component, replicas)) def apply(self, resources): self.events.append(("apply",)) self.applications.append(copy.deepcopy(resources)) for item in resources: if item["kind"] == "Deployment": self.replicas[item["metadata"]["name"]] = item["spec"]["replicas"] def rollout(self, component): self.events.append(("rollout", component)) if component == "postgres": if self.storage_error: raise RuntimeError("postgres rollout failed") self.postgres_ready = True if component == "worker" and self.worker_error: raise RuntimeError("worker rollout failed") def run(self, *args, **kwargs): if "psql" in args: return b"t\n" if "messages_ready" in args: return b"notification.send-email\t0\t0\n" self.events.append(("broker-consumer-check",)) count = self.consumer_counts.pop(0) if len(self.consumer_counts) > 1 else self.consumer_counts[0] return ("notification.send-email\t" + str(count) + "\n").encode() class ContractTests(unittest.TestCase): def test_example_requires_real_pinned_artifacts(self): with self.assertRaises(ValueError): ns.manifest(ns.read_json(PACKAGE / "config.example.json"), ns.read_json(PACKAGE / "artifacts.example.json")) def test_rejects_mutable_image_and_different_migrator_source(self): config, artifacts = fixture() artifacts["images"]["worker"]["image"] = "registry.hrynco.com/test:latest" with self.assertRaises(ValueError): ns.validate(config, artifacts) config, artifacts = fixture() artifacts["images"]["migrator"]["sourceArchiveSha256"] = "d" * 64 with self.assertRaises(ValueError): ns.validate(config, artifacts) def test_retained_prebound_storage_and_nonroot_exact_guard(self): config, artifacts = fixture() resources = ns.manifest(config, artifacts) pvs = [x for x in resources if x["kind"] == "PersistentVolume"] self.assertEqual(len(pvs), 2) for pv in pvs: self.assertEqual(pv["spec"]["persistentVolumeReclaimPolicy"], "Retain") self.assertEqual(pv["spec"]["claimRef"]["namespace"], ns.NAMESPACE) self.assertTrue(pv["spec"]["local"]["path"].startswith("/mnt/shared/notification-service/")) for resource in resources: if resource["kind"] != "Deployment": continue pod = resource["spec"]["template"]["spec"] self.assertFalse(pod["automountServiceAccountToken"]) self.assertTrue(pod["containers"][0]["securityContext"]["runAsNonRoot"]) if resource["metadata"]["name"] in ("postgres", "rabbitmq"): self.assertNotIn("fsGroup", pod.get("securityContext", {})) guard = pod["initContainers"][0] values = {x["name"]: x["value"] for x in guard["env"]} self.assertEqual(values["EXPECTED_UUID"], config["storage"]["uuid"]) self.assertEqual(values["EXPECTED_ROLE"], ns.NAMESPACE + "/" + resource["metadata"]["name"]) if resource["metadata"]["name"] == "postgres": self.assertFalse(any(m["mountPath"] == "/var/run/postgresql" for m in pod["containers"][0]["volumeMounts"])) def test_rabbit_probes_allow_observed_slow_diagnostics_without_rapid_restarts(self): config, artifacts = fixture() broker = next(x for x in ns.manifest(config, artifacts) if x["kind"] == "Deployment" and x["metadata"]["name"] == "rabbitmq") container = broker["spec"]["template"]["spec"]["containers"][0] for name in ("startupProbe", "readinessProbe", "livenessProbe"): probe = container[name] # The constrained live node measured seven seconds for a healthy CLI call. self.assertGreaterEqual(probe["timeoutSeconds"], 15) self.assertGreater(probe["periodSeconds"], probe["timeoutSeconds"]) self.assertEqual(probe["exec"]["command"][:2], ["rabbitmq-diagnostics", "-q"]) startup = container["startupProbe"] self.assertEqual(startup["periodSeconds"] * startup["failureThreshold"], 180) self.assertEqual(container["readinessProbe"]["exec"]["command"][-1], "check_running") self.assertGreaterEqual(container["livenessProbe"]["periodSeconds"], 30) def test_no_public_ports_or_internet_allowance(self): config, artifacts = fixture() resources = ns.manifest(config, artifacts) for item in resources: self.assertNotEqual(item["kind"], "Ingress") if item["kind"] == "Service": self.assertNotIn("type", item["spec"]) if item["kind"] == "NetworkPolicy": for rule in item["spec"].get("egress", []): for target in rule.get("to", []): self.assertNotIn("ipBlock", target) self.assertIn("namespaceSelector", target) self.assertIn("podSelector", target) worker = next(x for x in resources if x["kind"] == "Deployment" and x["metadata"]["name"] == "worker") self.assertEqual(worker["spec"]["replicas"], 0) def test_broker_hashes_and_scope_are_deterministic(self): config, _ = fixture() credentials = {} for key in ("rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword"): credentials[key] = "secret-test-value" credentials[key + "Salt"] = base64.b64encode(b"salt").decode() first = ns.rabbit_definitions(config, credentials) self.assertEqual(first, ns.rabbit_definitions(config, credentials)) self.assertNotIn("secret-test-value", json.dumps(first)) worker = next(x for x in first["permissions"] if x["user"] == "notification_worker") self.assertEqual(worker["read"], r"^notification\.send\-email$") self.assertNotIn(".*", worker["write"]) for user in ("notification_worker", "invemory_staging"): permissions = next(x for x in first["permissions"] if x["user"] == user) self.assertIsNotNone(re.fullmatch(permissions["write"], "amq.default")) self.assertIsNone(re.fullmatch(permissions["write"], "unreviewed.exchange")) def test_production_boot_definitions_keep_result_publish_permission(self): config, _ = fixture() credentials = {} for key in ("rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword"): credentials[key] = "staging-test-value" credentials[key + "Salt"] = base64.b64encode(b"salt").decode() password = "production-test-value" password_hash = base64.b64encode(b"prod" + hashlib.sha256(b"prod" + password.encode()).digest()).decode() definitions = ns.rabbit_definitions(config, credentials, password_hash) self.assertEqual([q["name"] for q in definitions["queues"]], ["notification.send-email", "invemory.staging.notification-results", "invemory.production.notification-results"]) production = next(x for x in definitions["users"] if x["name"] == "invemory_production") self.assertEqual(production["password_hash"], password_hash) self.assertNotIn(password, json.dumps(definitions)) worker = next(x for x in definitions["permissions"] if x["user"] == "notification_worker") self.assertIsNotNone(re.fullmatch(worker["configure"], "invemory.production.notification-results")) self.assertIsNotNone(re.fullmatch(worker["configure"], "invemory.staging.notification-results")) self.assertIsNotNone(re.fullmatch(worker["write"], "amq.default")) self.assertEqual(worker["read"], r"^notification\.send\-email$") def test_secrets_not_present_in_rendered_resources(self): config, artifacts = fixture() self.assertFalse(any(x["kind"] == "Secret" for x in ns.manifest(config, artifacts))) def test_storage_provision_rejects_wrong_disk_before_creating(self): with tempfile.TemporaryDirectory() as folder: target = Path(folder) / "notification-service" result = argparse.Namespace(stdout=json.dumps({"filesystems": [{"target": "/mnt/files", "fstype": "ext4", "uuid": "wrong"}]}).encode()) with patch.object(storage.sys, "platform", "linux"), patch.object(storage.os, "geteuid", return_value=0, create=True), patch.object(storage, "BASE", target), patch.object(storage.Path, "is_dir", return_value=True), patch.object(storage.Path, "is_symlink", return_value=False), patch.object(storage.subprocess, "run", return_value=result): with self.assertRaises(RuntimeError): storage.main() self.assertFalse(target.exists()) class ClusterQueries(unittest.TestCase): def cluster(self): return ns.Cluster(argparse.Namespace(kubectl="kubectl", kubeconfig="test-config")) def test_empty_pod_listing_uses_json_list_without_ignore_not_found(self): cluster = self.cluster() pod_list = {"apiVersion": "v1", "kind": "PodList", "items": []} with patch.object(cluster, "run", return_value=json.dumps(pod_list).encode()) as run: self.assertEqual(cluster.get("pods"), pod_list) self.assertEqual(run.call_args.args, ("get", "pods", "-n", ns.NAMESPACE, "-o", "json")) def test_missing_named_resource_retains_ignore_not_found_semantics(self): cluster = self.cluster() with patch.object(cluster, "run", return_value=b"") as run: self.assertIsNone(cluster.get("deployment", "worker")) self.assertIn("--ignore-not-found", run.call_args.args) def test_empty_non_json_listing_fails_with_explicit_gate(self): cluster = self.cluster() with patch.object(cluster, "run", return_value=b""): with self.assertRaisesRegex(RuntimeError, "no JSON for resource listing"): cluster.get("pods") def test_stopping_worker_with_no_pods_succeeds_without_wait(self): cluster = self.cluster() calls = [] def run(*args, **kwargs): calls.append(args) if args[:3] == ("get", "deployment", "worker"): return json.dumps({"spec": {"replicas": 0}}).encode() if args[:2] == ("get", "pods"): return json.dumps({"apiVersion": "v1", "kind": "PodList", "items": []}).encode() if args[:2] == ("scale", "deployment/worker"): return b"deployment.apps/worker scaled\n" raise AssertionError("unexpected operation during empty-pod scale") with patch.object(cluster, "run", side_effect=run): cluster.scale("worker", 0) self.assertTrue(any(call[0] == "scale" for call in calls)) self.assertFalse(any(call[0] == "wait" for call in calls)) self.assertNotIn("--ignore-not-found", next(call for call in calls if call[:2] == ("get", "pods"))) class DeploymentGates(unittest.TestCase): def execute(self, cluster, folder, migrate=False, enable=False, verified=False, backup_error=False, job_error=False): config, artifacts = fixture() args = argparse.Namespace(private_dir=folder, migrate=migrate, enable_worker=enable, sandbox_verified=verified) def save_backup(client, private_dir): client.events.append(("backup",)) if backup_error: raise RuntimeError("backup failed") def run_job(client, value): client.events.append(("job", value["metadata"]["name"])) if job_error: raise RuntimeError("migration failed") client.schema = list(ns.MIGRATIONS) elapsed = [0] def sleep(seconds): elapsed[0] += seconds with patch.object(ns, "preflight"), patch.object(ns, "host_storage_check"), patch.object(ns, "backup", side_effect=save_backup), patch.object(ns, "run_job", side_effect=run_job), patch.object(ns.time, "monotonic", side_effect=lambda: elapsed[0]), patch.object(ns.time, "sleep", side_effect=sleep): ns.deploy(cluster, config, artifacts, args) def test_ordinary_repeat_preserves_stopped_worker(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=0) self.execute(cluster, folder) self.assertEqual(cluster.replicas["worker"], 0) self.assertNotIn(("backup",), cluster.events) self.assertNotIn(("rollout", "worker"), cluster.events) def test_ordinary_repeat_waits_for_running_worker_and_checks_consumer(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1) self.execute(cluster, folder) self.assertIn(("rollout", "worker"), cluster.events) self.assertIn(("broker-consumer-check",), cluster.events) def test_ordinary_repeat_rejects_failed_worker_rollout(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1, worker_error=True) with self.assertRaisesRegex(RuntimeError, "worker rollout failed"): self.execute(cluster, folder) self.assertNotIn(("broker-consumer-check",), cluster.events) self.assertEqual(cluster.replicas["worker"], 0) def test_ordinary_repeat_stops_worker_when_consumer_is_missing(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1, consumer_count=0) with self.assertRaisesRegex(RuntimeError, "exactly one"): self.execute(cluster, folder) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(cluster.events.count(("broker-consumer-check",)), 30) def test_enabling_worker_waits_for_delayed_consumer(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(consumer_count=[0, 0, 1]) self.execute(cluster, folder, enable=True, verified=True) self.assertEqual(cluster.replicas["worker"], 1) self.assertEqual(cluster.events.count(("broker-consumer-check",)), 3) self.assertLess(cluster.events.index(("rollout", "worker")), cluster.events.index(("broker-consumer-check",))) def test_ordinary_repeat_waits_for_delayed_consumer(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1, consumer_count=[0, 1]) self.execute(cluster, folder) self.assertEqual(cluster.replicas["worker"], 1) self.assertEqual(cluster.events.count(("broker-consumer-check",)), 2) def test_extra_consumer_stops_worker_immediately(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(consumer_count=[2, 1]) with self.assertRaisesRegex(RuntimeError, "unexpected broker count"): self.execute(cluster, folder, enable=True, verified=True) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(cluster.events.count(("broker-consumer-check",)), 1) def test_consumer_gate_zero_timeout_performs_no_broker_query(self): config, _ = fixture() cluster = FakeCluster() with self.assertRaisesRegex(RuntimeError, "before timeout"): ns.wait_for_worker_consumer(cluster, config, timeout=0) self.assertEqual(cluster.events, []) def test_consumer_query_has_bounded_timeout_and_redacts_failure(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1) with patch.object(cluster, "run", side_effect=RuntimeError("private broker detail")) as run: with self.assertRaisesRegex(RuntimeError, "could not verify") as error: self.execute(cluster, folder) self.assertNotIn("private broker detail", str(error.exception)) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(run.call_args.kwargs, {"timeout": 20, "sensitive": True}) def test_consumer_query_cannot_succeed_after_deadline(self): config, _ = fixture() with patch.object(ns.time, "monotonic", side_effect=[0, 0, 61]): with self.assertRaisesRegex(RuntimeError, "before timeout"): ns.wait_for_worker_consumer(FakeCluster(), config) def test_backup_failure_leaves_both_writers_stopped(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1) with self.assertRaises(RuntimeError): self.execute(cluster, folder, migrate=True, backup_error=True) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(cluster.replicas["web"], 0) self.assertNotIn(("apply",), cluster.events) def test_migration_failure_leaves_both_writers_stopped(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1) with self.assertRaises(RuntimeError): self.execute(cluster, folder, migrate=True, job_error=True) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(cluster.replicas["web"], 0) self.assertLess(cluster.events.index(("backup",)), cluster.events.index(("job", "notification-migrator"))) def test_noop_migration_resumes_web_after_grants_and_keeps_worker_stopped(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1) self.execute(cluster, folder, migrate=True) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(cluster.replicas["web"], 1) self.assertLess(cluster.events.index(("job", "notification-grants")), cluster.events.index(("scale", "web", 1))) self.assertLess(cluster.events.index(("job", "notification-sandbox")), cluster.events.index(("scale", "web", 1))) def test_unknown_history_stops_before_mutation(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(history=["unknown"]) with self.assertRaises(RuntimeError): self.execute(cluster, folder, migrate=True) self.assertEqual(cluster.events, [("history",)]) def test_interrupted_first_install_recovers_storage_before_history_then_backs_up(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(history=[], worker=0, postgres_ready=False) self.execute(cluster, folder, migrate=True) self.assertLess(cluster.events.index(("rollout", "postgres")), cluster.events.index(("history",))) self.assertLess(cluster.events.index(("history",)), cluster.events.index(("backup",))) self.assertLess(cluster.events.index(("backup",)), cluster.events.index(("job", "notification-migrator"))) first_deployments = {v["metadata"]["name"] for v in cluster.applications[0] if v["kind"] == "Deployment"} self.assertEqual(first_deployments, {"postgres", "rabbitmq", "mailpit"}) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(cluster.replicas["web"], 1) def test_missing_deployment_does_not_skip_backup_of_retained_database(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(postgres_exists=False, postgres_ready=False) self.execute(cluster, folder, migrate=True) self.assertIn(("backup",), cluster.events) self.assertLess(cluster.events.index(("backup",)), cluster.events.index(("job", "notification-migrator"))) def test_recovery_backup_failure_keeps_writers_stopped_and_never_migrates(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1, postgres_ready=False) with self.assertRaisesRegex(RuntimeError, "backup failed"): self.execute(cluster, folder, migrate=True, backup_error=True) self.assertEqual(cluster.replicas["web"], 0) self.assertEqual(cluster.replicas["worker"], 0) self.assertEqual(len(cluster.applications), 1) # Foundation only. self.assertFalse(any(event[0] == "job" for event in cluster.events)) def test_recovery_storage_failure_does_not_query_history_or_start_writers(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(worker=1, postgres_ready=False, storage_error=True) with self.assertRaisesRegex(RuntimeError, "postgres rollout failed"): self.execute(cluster, folder, migrate=True) self.assertNotIn(("history",), cluster.events) self.assertNotIn(("backup",), cluster.events) self.assertEqual(cluster.replicas["web"], 0) self.assertEqual(cluster.replicas["worker"], 0) def test_recovery_unknown_history_keeps_writers_stopped_and_never_migrates(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(history=["unknown"], worker=1, postgres_ready=False) with self.assertRaisesRegex(RuntimeError, "unknown or non-prefix"): self.execute(cluster, folder, migrate=True) self.assertEqual(cluster.replicas["web"], 0) self.assertEqual(cluster.replicas["worker"], 0) self.assertFalse(any(event[0] == "job" for event in cluster.events)) def test_ordinary_deploy_of_stopped_database_requires_recovery_without_mutation(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster(postgres_ready=False) with self.assertRaisesRegex(RuntimeError, "recovery requires --migrate"): self.execute(cluster, folder) self.assertEqual(cluster.events, []) def test_enabling_worker_without_inspection_fails_before_mutation(self): with tempfile.TemporaryDirectory() as folder: cluster = FakeCluster() with self.assertRaises(RuntimeError): self.execute(cluster, folder, enable=True) self.assertEqual(cluster.events, []) if __name__ == "__main__": unittest.main()