257 lines
12 KiB
Python
257 lines
12 KiB
Python
"""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 == []
|