from __future__ import annotations import json import shutil import tempfile import unittest from pathlib import Path import deploy as l0 class L0DeployTests(unittest.TestCase): def make_transport_state(self, root: Path) -> None: for relative, content in ( ("secrets/tls/server.crt", "certificate"), ("secrets/tls/server.key", "private-key"), ): path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) def make_state(self, root: Path) -> None: self.make_transport_state(root) files = { "config/runtime.env": "NEXUS_ENV=test\n", # The deployer deliberately treats this as opaque and does not # acquire an L1/L2 metadata grammar. "config/serving.json": "catalog-owned-opaque-activation", "secrets/workload-identities.json": "{}", "secrets/signing-keys.json": "{}", } for role in l0.WORKER_ROLES: files[f"config/workers/{role}.json"] = json.dumps( {"work_family": f"runtime.{role}.v1"} ) files[f"secrets/workers/{role}.token"] = "token" for relative, content in files.items(): path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) for relative in ("data/catalog", "data/files", "data/uploads"): (root / relative).mkdir(parents=True) def write_database(self, root: Path, *, remote: bool) -> Path: path = root / ("remote.env" if remote else "local.env") if remote: content = "\n".join( ( "TIDB_HOST=cluster.example.com", "TIDB_PORT=4000", "TIDB_USER=runtime", "TIDB_PASSWORD=secret", "TIDB_DATABASE=nexuscrm", ) ) else: content = "\n".join( ( "TIDB_HOST=host.docker.internal", "TIDB_PORT=4000", "TIDB_USER=root", "TIDB_PASSWORD=", "TIDB_DATABASE=nexuscrm", "TIDB_TLS_DISABLE=1", ) ) path.write_text(content + "\n") return path def test_database_modes_fail_closed(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) local = self.write_database(root, remote=False) remote = self.write_database(root, remote=True) _, local_values = l0.validate_database(local, "local") _, remote_values = l0.validate_database(remote, "remote") self.assertEqual("host.docker.internal", local_values["TIDB_HOST"]) self.assertEqual("cluster.example.com", remote_values["TIDB_HOST"]) with self.assertRaisesRegex(l0.DeployError, "forbids"): l0.validate_database(local, "remote") with self.assertRaisesRegex(l0.DeployError, "requires TIDB_TLS_DISABLE"): l0.validate_database(remote, "local") local.write_text(local.read_text().replace("host.docker.internal", "127.0.0.1")) with self.assertRaisesRegex(l0.DeployError, "host.docker.internal"): l0.validate_database(local, "local") def test_remote_mode_explicitly_clears_inherited_tls_disable(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) self.make_state(root) database, _ = l0.validate_database(self.write_database(root, remote=True), "remote") environment = l0.compose_environment( l0.validate_serving_state(root), database, "remote", "nexuscrm-l0-test", "runtime-api:test", "runtime-worker:test", 19090, ) self.assertEqual("", environment["NEXUSCRM_L0_TIDB_TLS_DISABLE"]) def test_remote_init_requires_l0_identity_secrets(self) -> None: with self.assertRaisesRegex(l0.DeployError, "controlplane-env-file"): l0.validate_controlplane_env(None, "remote") with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "controlplane.env" path.write_text("JWT_SECRET=\nNEXUS_ADMIN_PASSWORD=\n") with self.assertRaisesRegex(l0.DeployError, "JWT_SECRET"): l0.validate_controlplane_env(path, "remote") path.write_text("JWT_SECRET=jwt\nNEXUS_ADMIN_PASSWORD=admin\n") self.assertEqual(path.resolve(), l0.validate_controlplane_env(path, "remote")) def test_deploy_requires_only_l0_transport_and_activation_requires_catalog(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) self.make_state(root) (root / "config/serving.json").unlink() self.assertEqual(root.resolve(), l0.validate_transport_state(root)) with self.assertRaisesRegex(l0.DeployError, "serving artifact"): l0.validate_serving_state(root) @unittest.skipUnless(shutil.which("docker"), "Docker is required for Compose validation") def test_compose_is_one_api_and_seven_worker_roles(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) self.make_state(root) database, _ = l0.validate_database(self.write_database(root, remote=False), "local") state = l0.validate_serving_state(root) environment = l0.compose_environment( state, database, "local", "nexuscrm-l0-test", "runtime-api:test", "runtime-worker:test", 19090, ) rendered = l0.run( l0.compose_command( "nexuscrm-l0-test", "--profile", "bootstrap", "--profile", "serving", "config", "--format", "json", ), environment=environment, capture=True, ) compose = json.loads(rendered.stdout) services = compose["services"] expected_services = { "runtime-bootstrap", "runtime-api", *(f"worker-{role}" for role in l0.WORKER_ROLES), } self.assertEqual(expected_services, set(services)) bootstrap = services["runtime-bootstrap"] self.assertEqual("runtime-api:test", bootstrap["image"]) self.assertEqual(["bootstrap"], bootstrap["command"]) self.assertNotIn("env_file", bootstrap) self.assertEqual(2, len(bootstrap["volumes"])) self.assertIn("/livez", " ".join(bootstrap["healthcheck"]["test"])) self.assertEqual("runtime-api:test", services["runtime-api"]["image"]) self.assertEqual(1, len(services["runtime-api"]["ports"])) published = services["runtime-api"]["ports"][0] self.assertEqual("127.0.0.1", published["host_ip"]) self.assertEqual("19090", str(published["published"])) self.assertEqual(9090, published["target"]) for role in l0.WORKER_ROLES: worker = services[f"worker-{role}"] self.assertEqual("runtime-worker:test", worker["image"]) self.assertFalse(worker.get("ports")) self.assertEqual("1", worker["environment"]["TIDB_TLS_DISABLE"]) for forbidden in ("tidb", "sfcrm", "integration", "gateway", "frontend", "ingress"): self.assertNotIn(forbidden, services) bootstrap_services = l0.run( l0.compose_command( "nexuscrm-l0-test", "--profile", "bootstrap", "config", "--services" ), environment=environment, capture=True, ).stdout.split() serving_services = l0.run( l0.compose_command( "nexuscrm-l0-test", "--profile", "serving", "config", "--services" ), environment=environment, capture=True, ).stdout.split() self.assertEqual(["runtime-bootstrap"], bootstrap_services) self.assertNotIn("runtime-bootstrap", serving_services) self.assertEqual(8, len(serving_services)) if __name__ == "__main__": unittest.main()