"""Tests for 'setup' and 'clean-deploy', the commands that replace the external deploy script.""" import os import sys from pathlib import Path import click import pytest from click.testing import CliRunner sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import alakazam from alakazam import get_storage_traces, require_empty_environment, setup_environment class FakeServer: """Answers the abra calls of a setup run and records the order of the steps.""" def __init__(self, deployed=(), secrets=(), volumes=()): self.deployed = set(deployed) self.secrets = dict(secrets) self.volumes = set(volumes) self.steps = [] def install(self, monkeypatch, config=None): if config is None: config = { "authentik": {"app_domain": "login.a.org", "server": "a.org", "dependency": []}, "nextcloud": {"app_domain": "cloud.a.org", "server": "a.org", "dependency": ["authentik"], "readiness-hooks": [{"cmd": "app ready", "retries": 1}]}, } monkeypatch.setattr(alakazam, "INSTANCE_CONFIGS", {"a.org": config}) monkeypatch.setattr(alakazam, "abra", self.abra) monkeypatch.setattr(alakazam, "sleep", lambda seconds: None) monkeypatch.setattr(alakazam, "configure_apps", lambda recipes: self.steps.append("config")) monkeypatch.setattr(alakazam, "create_secrets", self.create_secrets) monkeypatch.setattr(alakazam, "deploy_apps", self.deploy_apps) monkeypatch.setattr(alakazam, "undeploy_apps", self.undeploy_apps) monkeypatch.setattr(alakazam, "purge_apps", self.purge_apps) monkeypatch.setattr(alakazam, "execute_cmds", lambda app_config, **kw: self.steps.append(f"init:{app_config['app_domain']}")) return self def create_secrets(self, recipes, variants=None, **kwargs): label = "conf" if variants == {"conf"} else "secrets" self.steps.append(f"{label}:{','.join(recipes) if recipes else 'all'}") def deploy_apps(self, instance_apps, **kwargs): for _, apps in instance_apps.items(): for app, domain, *_ in apps: self.steps.append(f"deploy:{app}") self.deployed.add(domain) def undeploy_apps(self, instance_apps): self.steps.append("undeploy") self.deployed.clear() def purge_apps(self, instance_apps): self.steps.append("purge") self.secrets.clear() self.volumes.clear() def abra(self, *args, machine_output=False, **kwargs): if args[:3] == ("app", "secret", "ls"): return [{"name": name, "created on server": "true"} for name in self.secrets.get(args[3], [])] if args[:3] == ("app", "volume", "ls"): return "NAME ON SERVER\ndata /var/lib\n" if args[4] in self.volumes else "" if args[:2] == ("app", "ls"): apps = [{"appName": d, "status": "deployed", "chaos": "false", "version": "1.0.0"} for d in sorted(self.deployed)] return {"a.org": {"apps": apps}} if args[:2] == ("app", "ps"): self.steps.append(f"ps:{args[-1]}") return {"app": {"service": "app", "status": "healthy", "version": "1.0.0", "chaos": "false", "state": "running", "image": "img"}} if args[:3] == ("app", "cmd", "-T"): self.steps.append(f"ready:{args[4]}") return "ok" raise AssertionError(f"unexpected abra call: {args}") class TestStorageTraces: def test_an_empty_environment_leaves_no_traces(self, monkeypatch): FakeServer().install(monkeypatch) assert get_storage_traces({"a.org": [["authentik", "login.a.org"]]}) == [] def test_leftover_secrets_are_found(self, monkeypatch): FakeServer(secrets={"login.a.org": ["email_pass", "admin_pass"]}).install(monkeypatch) traces = get_storage_traces({"a.org": [["authentik", "login.a.org"]]}) assert "2 secret(s)" in traces[0] assert "admin_pass, email_pass" in traces[0] def test_leftover_volumes_are_found(self, monkeypatch): FakeServer(volumes=["login.a.org"]).install(monkeypatch) traces = get_storage_traces({"a.org": [["authentik", "login.a.org"]]}) assert traces == ["login.a.org still has volumes on its server"] def test_an_unreadable_volume_list_aborts(self, monkeypatch): """Guessing that a failing read means "no volumes" would deploy on top of existing data.""" server = FakeServer().install(monkeypatch) original = server.abra def failing(*args, **kwargs): if args[:3] == ("app", "volume", "ls"): raise RuntimeError("abra -o app volume ls -d login.a.org\n STDERR: Error: no such server") return original(*args, **kwargs) monkeypatch.setattr(alakazam, "abra", failing) with pytest.raises(click.ClickException) as excinfo: get_storage_traces({"a.org": [["authentik", "login.a.org"]]}) assert "could not read the volumes of login.a.org" in excinfo.value.message def test_uncreated_secrets_do_not_count(self, monkeypatch): """A secret the recipe declares but the server does not hold is not a leftover.""" server = FakeServer().install(monkeypatch) monkeypatch.setattr(alakazam, "abra", lambda *a, **k: [{"name": "x", "created on server": "false"}] if a[:3] == ("app", "secret", "ls") else "") assert get_storage_traces({"a.org": [["authentik", "login.a.org"]]}) == [] class TestRequireEmptyEnvironment: def test_no_traces_passes(self): require_empty_environment([]) def test_traces_abort_with_a_hint(self): with pytest.raises(click.ClickException) as excinfo: require_empty_environment(["login.a.org is deployed"]) assert "login.a.org is deployed" in excinfo.value.message assert "clean-deploy" in excinfo.value.message assert excinfo.value.exit_code != 0 class TestSetupEnvironment: def test_runs_every_phase_per_dependency_level(self, monkeypatch): config = { "authentik": {"app_domain": "login.a.org", "server": "a.org", "dependency": []}, "traefik": {"app_domain": "a.org", "server": "a.org", "dependency": []}, "nextcloud": {"app_domain": "cloud.a.org", "server": "a.org", "dependency": ["authentik"], "readiness-hooks": [{"cmd": "app ready", "retries": 1}]}, "wekan": {"app_domain": "boards.a.org", "server": "a.org", "dependency": ["authentik"]}, } server = FakeServer().install(monkeypatch, config=config) setup_environment(()) assert server.steps[:8] == [ "config", "conf:all", "secrets:authentik,traefik", "deploy:authentik", "deploy:traefik", "secrets:nextcloud,wekan", "deploy:nextcloud", "deploy:wekan", ] def test_apps_of_one_level_are_deployed_before_that_level_is_awaited(self, monkeypatch): """Nothing in a level waits for a sibling, the readiness hooks close the level.""" server = FakeServer().install(monkeypatch) setup_environment(()) assert server.steps.index("deploy:nextcloud") < server.steps.index("ready:cloud.a.org") def test_configured_secrets_are_inserted_before_any_exchange(self, monkeypatch): """Otherwise an exchange generates a random value that shadows the configured one.""" server = FakeServer().install(monkeypatch) setup_environment(()) assert server.steps[1] == "conf:all" assert all(step != "conf:all" for step in server.steps[2:]) def test_initial_hooks_run_after_the_health_check(self, monkeypatch): server = FakeServer().install(monkeypatch) setup_environment(()) assert server.steps[-2:] == ["init:login.a.org", "init:cloud.a.org"] assert server.steps.index("ps:login.a.org") < server.steps.index("init:login.a.org") def test_a_deployed_app_aborts_before_deploying(self, monkeypatch): server = FakeServer(deployed=["login.a.org"]).install(monkeypatch) with pytest.raises(click.ClickException) as excinfo: setup_environment(()) assert "login.a.org is deployed" in excinfo.value.message assert server.steps == ["config"] def test_leftover_secrets_abort_the_run(self, monkeypatch): """An undeploy without a purge must not look like an empty environment.""" server = FakeServer(secrets={"login.a.org": ["email_pass"]}).install(monkeypatch) with pytest.raises(click.ClickException) as excinfo: setup_environment(()) assert "email_pass" in excinfo.value.message assert server.steps == ["config"] def test_leftover_volumes_abort_the_run(self, monkeypatch): server = FakeServer(volumes=["cloud.a.org"]).install(monkeypatch) with pytest.raises(click.ClickException): setup_environment(()) assert "deploy:authentik" not in server.steps def test_nothing_configured_is_a_failure(self, monkeypatch): FakeServer().install(monkeypatch, config={}) with pytest.raises(click.ClickException) as excinfo: setup_environment(()) assert "nothing to set up" in excinfo.value.message def test_a_failing_readiness_hook_stops_the_run(self, monkeypatch): server = FakeServer().install(monkeypatch) original = server.abra def failing(*args, **kwargs): if args[:3] == ("app", "cmd", "-T"): raise RuntimeError("FATA blueprints missing") return original(*args, **kwargs) monkeypatch.setattr(alakazam, "abra", failing) with pytest.raises(click.ClickException) as excinfo: setup_environment(()) assert "is not ready" in excinfo.value.message assert "ps:login.a.org" not in server.steps class TestCleanDeployCommand: def invoke(self, args, group_path, monkeypatch): monkeypatch.setattr(alakazam, "GROUP_PATH", Path(group_path)) return CliRunner().invoke(alakazam.clean_deploy, args, standalone_mode=False) def test_rebuilds_an_instance(self, monkeypatch, tmp_path): server = FakeServer(deployed=["login.a.org"], secrets={"login.a.org": ["email_pass"]}).install(monkeypatch) instance = tmp_path / "example.com.yml" instance.write_text("") result = self.invoke(["-n"], instance, monkeypatch) assert result.exit_code == 0, result.exception assert server.steps[:4] == ["config", "undeploy", "purge", "config"] def test_the_apps_are_configured_before_they_are_purged(self, monkeypatch, tmp_path): """purge_apps() skips an app without a local .env, so a fresh checkout would purge nothing.""" server = FakeServer(deployed=["login.a.org"]).install(monkeypatch) instance = tmp_path / "example.com.yml" instance.write_text("") self.invoke(["-n"], instance, monkeypatch) assert server.steps.index("config") < server.steps.index("purge") def test_the_apps_are_configured_again_after_the_purge(self, monkeypatch, tmp_path): """'abra app rm' deletes the local .env as well, so setup has to write it a second time.""" server = FakeServer(deployed=["login.a.org"]).install(monkeypatch) instance = tmp_path / "example.com.yml" instance.write_text("") self.invoke(["-n"], instance, monkeypatch) assert server.steps.count("config") == 2 assert server.steps.index("purge") < len(server.steps) - 1 - server.steps[::-1].index("config") def test_a_group_directory_is_rejected(self, monkeypatch, tmp_path): server = FakeServer().install(monkeypatch) result = self.invoke(["-n"], tmp_path, monkeypatch) assert isinstance(result.exception, click.UsageError) assert "single instance configuration file" in str(result.exception) assert server.steps == [] def test_the_confirmation_can_be_declined(self, monkeypatch, tmp_path): server = FakeServer().install(monkeypatch) instance = tmp_path / "example.com.yml" instance.write_text("") monkeypatch.setattr(alakazam, "GROUP_PATH", instance) result = CliRunner().invoke(alakazam.clean_deploy, [], input="no\n", standalone_mode=False) assert result.exit_code == 0 assert server.steps == []