from __future__ import annotations from contextlib import redirect_stderr, redirect_stdout import io import json from pathlib import Path import stat import subprocess import sys import tempfile import unittest from unittest.mock import MagicMock, patch META_ROOT = Path(__file__).resolve().parents[1] DEPLOYMENT_TOOLS = META_ROOT / "tools" / "deployment" if str(DEPLOYMENT_TOOLS) not in sys.path: sys.path.insert(0, str(DEPLOYMENT_TOOLS)) from govoplan_deploy.bundle import ( # noqa: E402 atomic_write, bundle_paths, canonical_json, environment_fingerprint, initial_secrets, read_env, reconcile_runtime_environment, render_compose, write_env, ) from govoplan_deploy.cli import main # noqa: E402 import govoplan_deploy.cli as deployment_cli # noqa: E402 from govoplan_deploy.model import ( # noqa: E402 SpecError, default_spec, parse_spec, ) from govoplan_deploy.planning import _endpoint_check, build_plan # noqa: E402 import govoplan_deploy.planning as deployment_planning # noqa: E402 def run_cli(arguments: list[str]) -> tuple[int, str, str]: stdout = io.StringIO() stderr = io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): result = main(arguments) return result, stdout.getvalue(), stderr.getvalue() class DeploymentInstallerTests(unittest.TestCase): def test_default_bundle_has_base_modules_and_managed_dependencies(self) -> None: spec = default_spec() compose = render_compose(spec) self.assertEqual("evaluation", spec.profile) self.assertIn("access", spec.enabled_modules) self.assertIn("postgres", compose["services"]) self.assertIn("redis", compose["services"]) self.assertIn("worker", compose["services"]) self.assertNotIn("test-mail", compose["services"]) self.assertEqual( ["127.0.0.1:8080:8080"], compose["services"]["web"]["ports"], ) def test_disabled_redis_removes_workers_and_sets_single_process_acknowledgement( self, ) -> None: spec = default_spec(redis_mode="disabled") values = initial_secrets(spec) compose = render_compose(spec) self.assertNotIn("redis", compose["services"]) self.assertNotIn("worker", compose["services"]) self.assertNotIn("scheduler", compose["services"]) self.assertEqual("false", values["CELERY_ENABLED"]) self.assertEqual( "true", values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] ) def test_self_hosted_rejects_insecure_or_test_only_choices(self) -> None: with self.assertRaisesRegex(SpecError, "must use HTTPS"): default_spec(profile="self-hosted") with self.assertRaisesRegex(SpecError, "require managed or external Redis"): default_spec( profile="self-hosted", public_url="https://govoplan.example.test", redis_mode="disabled", ) with self.assertRaisesRegex(SpecError, "test-mail"): default_spec( profile="self-hosted", public_url="https://govoplan.example.test", mail_mode="test-mail", ) def test_spec_rejects_unknown_fields_and_duplicate_modules(self) -> None: raw = default_spec().to_dict() raw["unexpected"] = True with self.assertRaisesRegex(SpecError, "unknown fields"): parse_spec(raw) raw = default_spec().to_dict() raw["enabled_modules"].append(raw["enabled_modules"][0]) with self.assertRaisesRegex(SpecError, "duplicate module"): parse_spec(raw) raw = default_spec().to_dict() raw["network_subnet"] = "127.0.0.0/24" with self.assertRaisesRegex(SpecError, "RFC1918"): parse_spec(raw) def test_generated_secrets_are_stable_across_reconfiguration(self) -> None: first = default_spec() values = initial_secrets(first) master_key = values["MASTER_KEY_B64"] postgres_password = values["POSTGRES_PASSWORD"] redis_password = values["REDIS_PASSWORD"] second = default_spec(mail_mode="test-mail", module_set="full") reconciled = reconcile_runtime_environment(second, values) self.assertEqual(master_key, reconciled["MASTER_KEY_B64"]) self.assertEqual(postgres_password, reconciled["POSTGRES_PASSWORD"]) self.assertEqual(redis_password, reconciled["REDIS_PASSWORD"]) self.assertEqual( ",".join(second.enabled_modules), reconciled["ENABLED_MODULES"] ) def test_external_services_require_explicit_valid_urls(self) -> None: spec = default_spec(postgres_mode="external", redis_mode="external") with self.assertRaisesRegex(ValueError, "external PostgreSQL"): initial_secrets(spec) with self.assertRaisesRegex(ValueError, "external Redis"): initial_secrets( spec, supplied={ "DATABASE_URL": ( "postgresql+psycopg://user:secret@db.example.test/govoplan" ) }, ) values = initial_secrets( spec, supplied={ "DATABASE_URL": ( "postgresql+psycopg://user:secret@db.example.test/govoplan" ), "REDIS_URL": "rediss://:secret@redis.example.test/0", }, ) self.assertEqual( "rediss://:secret@redis.example.test/0", values["REDIS_URL"] ) def test_s3_requires_complete_private_configuration_and_https_in_production( self, ) -> None: evaluation = default_spec(storage_mode="s3") with self.assertRaisesRegex(ValueError, "S3 storage requires"): initial_secrets(evaluation) supplied = { "FILE_STORAGE_S3_ENDPOINT_URL": "http://s3.example.test", "FILE_STORAGE_S3_REGION": "eu-test-1", "FILE_STORAGE_S3_ACCESS_KEY_ID": "key", "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "secret", "FILE_STORAGE_S3_BUCKET": "govoplan", } self.assertEqual( "s3", initial_secrets(evaluation, supplied=supplied)[ "FILE_STORAGE_BACKEND" ], ) production = default_spec( profile="self-hosted", public_url="https://govoplan.example.test", storage_mode="s3", ) with self.assertRaisesRegex(ValueError, "must use HTTPS"): initial_secrets(production, supplied=supplied) def test_compose_contains_no_secret_values(self) -> None: spec = default_spec() values = initial_secrets(spec) rendered = json.dumps(render_compose(spec), sort_keys=True) for key in ( "MASTER_KEY_B64", "POSTGRES_PASSWORD", "REDIS_PASSWORD", ): self.assertNotIn(values[key], rendered) self.assertNotIn("secrets.env", rendered) self.assertNotIn("env_file", rendered) self.assertNotIn("./:/etc/govoplan", rendered) self.assertNotIn("installer", render_compose(spec)["services"]) postgres_environment = render_compose(spec)["services"]["postgres"][ "environment" ] redis_environment = render_compose(spec)["services"]["redis"][ "environment" ] self.assertEqual( {"POSTGRES_DB", "POSTGRES_USER", "POSTGRES_PASSWORD"}, set(postgres_environment), ) self.assertEqual({"REDIS_PASSWORD"}, set(redis_environment)) self.assertNotIn("MASTER_KEY_B64", postgres_environment) self.assertNotIn("MASTER_KEY_B64", redis_environment) def test_secret_environment_round_trips_literal_special_characters(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: path = Path(directory) / "secrets.env" values = { "PASSWORD": r"dollar$ quote' slash\\ hash# space ", "EMPTY_VALUE": "", } write_env(path, values) self.assertEqual(values, read_env(path)) def test_first_plan_creates_services_and_applied_receipt_becomes_noop( self, ) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: paths = bundle_paths(Path(directory)) paths.root.chmod(0o700) spec = default_spec() write_env(paths.env, initial_secrets(spec)) first = build_plan(spec, paths, include_host_checks=False) self.assertIn( ("create", "installation"), {(action.action, action.target) for action in first.actions}, ) self.assertIn( ("start", "api"), {(action.action, action.target) for action in first.actions}, ) receipt = { "spec_sha256": first.desired_spec_sha256, "compose_sha256": first.desired_compose_sha256, "environment_fingerprint": ( first.desired_environment_fingerprint ), "services": list(render_compose(spec)["services"]), } atomic_write(paths.receipt, canonical_json(receipt), mode=0o600) second = build_plan(spec, paths, include_host_checks=False) self.assertEqual( [("noop", "installation")], [(action.action, action.target) for action in second.actions], ) def test_reconfiguration_plans_removed_component_containers(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: paths = bundle_paths(Path(directory)) paths.root.chmod(0o700) first_spec = default_spec(mail_mode="test-mail") first_environment = initial_secrets(first_spec) write_env(paths.env, first_environment) first_plan = build_plan( first_spec, paths, include_host_checks=False ) atomic_write( paths.receipt, canonical_json( { "spec_sha256": first_plan.desired_spec_sha256, "compose_sha256": first_plan.desired_compose_sha256, "environment_fingerprint": ( first_plan.desired_environment_fingerprint ), "services": list(render_compose(first_spec)["services"]), } ), mode=0o600, ) second_spec = default_spec(redis_mode="disabled", mail_mode="disabled") write_env( paths.env, reconcile_runtime_environment(second_spec, first_environment), ) second_plan = build_plan( second_spec, paths, include_host_checks=False ) removed = { action.target for action in second_plan.actions if action.action == "remove" } self.assertEqual( {"redis", "worker", "scheduler", "test-mail"}, removed, ) def test_secret_change_is_planned_without_exposing_secret_values(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: paths = bundle_paths(Path(directory)) paths.root.chmod(0o700) spec = default_spec() values = initial_secrets(spec) write_env(paths.env, values) first = build_plan(spec, paths, include_host_checks=False) atomic_write( paths.receipt, canonical_json( { "spec_sha256": first.desired_spec_sha256, "compose_sha256": first.desired_compose_sha256, "environment_fingerprint": ( first.desired_environment_fingerprint ), "services": list(render_compose(spec)["services"]), } ), mode=0o600, ) rotated = dict(values) rotated["REDIS_PASSWORD"] = "rotated-high-entropy-value" write_env(paths.env, rotated) second = build_plan(spec, paths, include_host_checks=False) plan_json = json.dumps(second.to_dict()) self.assertIn( ("reconfigure", "environment"), {(action.action, action.target) for action in second.actions}, ) self.assertNotIn(rotated["REDIS_PASSWORD"], plan_json) self.assertEqual( environment_fingerprint(rotated), second.desired_environment_fingerprint, ) def test_external_endpoint_preflight_reports_reachability(self) -> None: connection = MagicMock() connection.__enter__.return_value = connection with patch.object( deployment_planning.socket, "create_connection", return_value=connection, ) as connect: check = _endpoint_check( "external.redis", "External Redis", "rediss://redis.example.test/0", default_ports={"redis": 6379, "rediss": 6380}, ) self.assertEqual("ok", check.level) connect.assert_called_once_with( ("redis.example.test", 6380), timeout=1.5, ) def test_cli_init_writes_private_idempotent_bundle(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: root = Path(directory) / "installation" result, _stdout, _stderr = run_cli( [ "init", "--non-interactive", "--directory", str(root), "--redis", "disabled", "--mail", "test-mail", ] ) self.assertEqual(0, result) self.assertEqual(0o700, stat.S_IMODE(root.stat().st_mode)) for name in ( "installation.json", "secrets.env", "compose.json", "plan.json", ): self.assertEqual( 0o600, stat.S_IMODE((root / name).stat().st_mode) ) values = read_env(root / "secrets.env") self.assertTrue(values["MASTER_KEY_B64"]) self.assertNotIn( values["POSTGRES_PASSWORD"], (root / "compose.json").read_text(encoding="utf-8"), ) self.assertEqual( 1, run_cli( [ "init", "--non-interactive", "--directory", str(root), ] )[0], ) def test_cli_requires_external_url_when_switching_from_managed(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: root = Path(directory) / "installation" self.assertEqual( 0, run_cli( [ "init", "--non-interactive", "--directory", str(root), ] )[0], ) self.assertEqual( 1, run_cli( [ "configure", "--directory", str(root), "--postgres", "external", ] )[0], ) self.assertEqual( "managed", json.loads( (root / "installation.json").read_text(encoding="utf-8") )["components"]["postgres"]["mode"], ) self.assertEqual( 1, run_cli( [ "configure", "--directory", str(root), "--installation-id", "renamed-installation", ] )[0], ) def test_deployer_builds_and_runs_as_one_zipapp(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: output = Path(directory) / "govoplan-deploy.pyz" build = subprocess.run( [ sys.executable, str(DEPLOYMENT_TOOLS / "build-deployer-zipapp.py"), "--output", str(output), ], cwd=META_ROOT, check=False, capture_output=True, text=True, ) help_result = subprocess.run( [sys.executable, str(output), "--help"], cwd=META_ROOT, check=False, capture_output=True, text=True, ) install_root = Path(directory) / "install" init_result = subprocess.run( [ sys.executable, str(output), "init", "--non-interactive", "--directory", str(install_root), ], cwd=META_ROOT, check=False, capture_output=True, text=True, ) render_result = subprocess.run( [ sys.executable, str(output), "render", "--directory", str(install_root), "--json", ], cwd=META_ROOT, check=False, capture_output=True, text=True, ) self.assertEqual(0, build.returncode, build.stderr) self.assertTrue(output.exists()) self.assertEqual(0o755, stat.S_IMODE(output.stat().st_mode)) self.assertEqual(0, help_result.returncode, help_result.stderr) self.assertIn("govoplan-deploy", help_result.stdout) self.assertEqual(0, init_result.returncode, init_result.stderr) self.assertEqual(1, render_result.returncode, render_result.stderr) self.assertTrue(json.loads(render_result.stdout)["blocked"]) def test_generated_environment_passes_core_startup_validation_when_available( self, ) -> None: try: from govoplan_core.core.install_config import ( validate_runtime_configuration, ) except ModuleNotFoundError: self.skipTest("govoplan-core is not installed in this test environment") for profile, public_url in ( ("evaluation", "http://127.0.0.1:8080"), ("self-hosted", "https://govoplan.example.test"), ): with self.subTest(profile=profile): environment = initial_secrets( default_spec(profile=profile, public_url=public_url) ) validation = validate_runtime_configuration(environment) self.assertEqual( (), validation.errors, validation.to_text(), ) def test_apply_orders_dependencies_migration_and_runtime_before_receipt( self, ) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: root = Path(directory) / "installation" self.assertEqual( 0, run_cli( [ "init", "--non-interactive", "--directory", str(root), "--api-image", "local/govoplan-api:test", "--web-image", "local/govoplan-web:test", ] )[0], ) commands: list[list[str]] = [] def record_command(argv, *, cwd): self.assertEqual(root, cwd) commands.append(list(argv)) compose_version = subprocess.CompletedProcess( args=["docker", "compose", "version"], returncode=0, stdout="2.30.0\n", stderr="", ) with ( patch.object( deployment_cli.shutil, "which", return_value="/usr/bin/docker", ), patch.object( deployment_planning.shutil, "which", return_value="/usr/bin/docker", ), patch.object( deployment_planning, "_run_command", return_value=compose_version, ), patch.object(deployment_cli, "_run", side_effect=record_command), patch.object(deployment_cli, "_wait_for_health") as wait_for_health, ): result, _stdout, _stderr = run_cli( [ "apply", "--directory", str(root), "--skip-pull", "--allow-unverified-images", ] ) self.assertEqual(0, result) migrate_index = next( index for index, command in enumerate(commands) if command[-3:] == ["run", "--rm", "migrate"] ) runtime_index = next( index for index, command in enumerate(commands) if "--remove-orphans" in command ) self.assertLess(migrate_index, runtime_index) self.assertIn("postgres", commands[0]) self.assertIn("redis", commands[0]) wait_for_health.assert_called_once_with( "http://127.0.0.1:8080/health", timeout_seconds=120.0, ) receipt = json.loads( (root / "receipt.json").read_text(encoding="utf-8") ) self.assertEqual("govoplan-local", receipt["installation_id"]) self.assertEqual( {"address": "127.0.0.1", "port": 8080}, receipt["listen"], ) self.assertNotIn("installer", receipt["services"]) def test_installation_root_symlink_is_rejected(self) -> None: if not hasattr(Path, "symlink_to"): self.skipTest("symbolic links are unavailable") with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory: base = Path(directory) target = base / "real" target.mkdir() link = base / "linked" link.symlink_to(target, target_is_directory=True) with self.assertRaisesRegex(ValueError, "symbolic link"): bundle_paths(link) if __name__ == "__main__": unittest.main()