227 lines
10 KiB
Python
227 lines
10 KiB
Python
"""Tests for the configuration preflight that runs before any abra call."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
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 check_config_readable, get_relevant_config_paths, preflight_configs
|
|
|
|
# a git-crypt encrypted file starts with this marker and is not valid UTF-8
|
|
GIT_CRYPT_HEADER = b"\x00GITCRYPT\x00\xeb\x1c\x9a\x7f\x3d\x00\xff\xfe"
|
|
|
|
|
|
@pytest.fixture
|
|
def config_root(tmp_path):
|
|
"""A root path holding one group with one instance, mirroring the documented layout."""
|
|
root = tmp_path / "root"
|
|
group = root / "group"
|
|
group.mkdir(parents=True)
|
|
(root / "alaka.yml").write_text("authentik:\n version: 1.0.0\n")
|
|
(root / "config-sets.yml").write_text("bbb:\n authentik:\n env:\n A: b\n")
|
|
(group / "alaka-versions.yml").write_text("authentik:\n version: 2.0.0\n")
|
|
(group / "example.com.yml").write_text("authentik:\n server: example.com\n")
|
|
return root
|
|
|
|
|
|
class TestCheckConfigReadable:
|
|
def test_valid_config(self, tmp_path):
|
|
path = tmp_path / "alaka.yml"
|
|
path.write_text("authentik:\n version: 1.0.0\n")
|
|
assert check_config_readable(path) is None
|
|
|
|
def test_missing_file_is_not_a_failure(self, tmp_path):
|
|
"""A configuration that simply is not there is a legitimate state."""
|
|
assert check_config_readable(tmp_path / "absent.yml") is None
|
|
|
|
def test_empty_file_is_not_a_failure(self, tmp_path):
|
|
path = tmp_path / "alaka.yml"
|
|
path.write_text("")
|
|
assert check_config_readable(path) is None
|
|
|
|
def test_encrypted_file(self, tmp_path):
|
|
path = tmp_path / "alaka-secrets.yml"
|
|
path.write_bytes(GIT_CRYPT_HEADER)
|
|
assert "codec can't decode" in check_config_readable(path)
|
|
|
|
def test_malformed_yaml(self, tmp_path):
|
|
path = tmp_path / "alaka.yml"
|
|
path.write_text("authentik:\n - version: 1.0.0\n broken: [\n")
|
|
assert check_config_readable(path)
|
|
|
|
def test_reason_is_a_single_line(self, tmp_path):
|
|
"""Multi-line parser output would break up the failure report."""
|
|
path = tmp_path / "alaka.yml"
|
|
path.write_text("authentik:\n - version: 1.0.0\n broken: [\n")
|
|
assert "\n" not in check_config_readable(path)
|
|
|
|
|
|
class TestReadConfigSeverity:
|
|
"""An unreadable file that reaches read_config is out of scope by construction."""
|
|
|
|
def test_an_unreadable_config_warns_rather_than_errors(self, config_root, caplog):
|
|
import logging
|
|
broken = config_root / "foreign" / "alaka-secrets.yml"
|
|
broken.parent.mkdir()
|
|
broken.write_bytes(GIT_CRYPT_HEADER)
|
|
with caplog.at_level(logging.DEBUG):
|
|
assert alakazam.read_config(str(broken)) == {}
|
|
assert [r.levelname for r in caplog.records] == ["WARNING"]
|
|
assert "is skipped" in caplog.text
|
|
|
|
|
|
class TestPreflightConfigs:
|
|
def test_passes_for_readable_configs(self, config_root):
|
|
preflight_configs([config_root / "alaka.yml", config_root / "config-sets.yml"])
|
|
|
|
def test_aborts_on_an_unreadable_config(self, config_root):
|
|
broken = config_root / "alaka-secrets.yml"
|
|
broken.write_bytes(GIT_CRYPT_HEADER)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
preflight_configs([config_root / "alaka.yml", broken])
|
|
assert str(broken) in excinfo.value.message
|
|
assert "git crypt unlock" in excinfo.value.message
|
|
|
|
def test_reports_every_failure_at_once(self, config_root):
|
|
"""A CI run should see all broken files, not just the first one."""
|
|
for name in ("alaka-a.yml", "alaka-b.yml"):
|
|
(config_root / name).write_bytes(GIT_CRYPT_HEADER)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
preflight_configs([config_root / "alaka-a.yml", config_root / "alaka-b.yml"])
|
|
assert "alaka-a.yml" in excinfo.value.message
|
|
assert "alaka-b.yml" in excinfo.value.message
|
|
assert excinfo.value.message.startswith("2 ")
|
|
|
|
def test_exit_code_is_non_zero(self):
|
|
assert click.ClickException("").exit_code != 0
|
|
|
|
|
|
class TestRelevantConfigPaths:
|
|
def test_covers_the_inheritance_chain_and_the_instances(self, config_root):
|
|
paths = get_relevant_config_paths(config_root, config_root / "group", [])
|
|
assert config_root / "alaka.yml" in paths
|
|
assert config_root / "config-sets.yml" in paths
|
|
assert config_root / "group" / "alaka-versions.yml" in paths
|
|
assert config_root / "group" / "example.com.yml" in paths
|
|
|
|
def test_covers_the_connection_configuration(self, config_root):
|
|
paths = get_relevant_config_paths(config_root, config_root / "group", [])
|
|
assert alakazam.Path(alakazam.COMBINE_PATH) in paths
|
|
|
|
def test_a_single_instance_file_pulls_in_its_ancestors(self, config_root):
|
|
paths = get_relevant_config_paths(config_root, config_root / "group" / "example.com.yml", [])
|
|
assert config_root / "alaka.yml" in paths
|
|
assert config_root / "group" / "alaka-versions.yml" in paths
|
|
assert config_root / "group" / "example.com.yml" in paths
|
|
|
|
def test_leaves_out_foreign_groups(self, config_root):
|
|
"""A group that cannot change this run's result stays out of the strict check."""
|
|
foreign = config_root / "foreign"
|
|
foreign.mkdir()
|
|
(foreign / "alaka.yml").write_text("nextcloud:\n version: 1.0.0\n")
|
|
(foreign / "other.com.yml").write_text("nextcloud:\n server: other.com\n")
|
|
paths = get_relevant_config_paths(config_root, config_root / "group", [])
|
|
assert foreign / "alaka.yml" not in paths
|
|
assert foreign / "other.com.yml" not in paths
|
|
|
|
def test_leaves_out_paths_above_the_root(self, config_root):
|
|
(config_root.parent / "alaka.yml").write_text("authentik:\n version: 0.0.1\n")
|
|
paths = get_relevant_config_paths(config_root, config_root / "group", [])
|
|
assert config_root.parent / "alaka.yml" not in paths
|
|
|
|
def test_honours_exclude_paths(self, config_root):
|
|
excluded = config_root / "group" / "excluded"
|
|
excluded.mkdir()
|
|
(excluded / "skip.com.yml").write_text("authentik:\n server: skip.com\n")
|
|
paths = get_relevant_config_paths(config_root, config_root / "group", [str(excluded)])
|
|
assert excluded / "skip.com.yml" not in paths
|
|
|
|
def test_has_no_duplicates(self, config_root):
|
|
paths = get_relevant_config_paths(config_root, config_root / "group", [])
|
|
assert len(paths) == len(set(paths))
|
|
|
|
|
|
class TestPreflightInCli:
|
|
"""The preflight has to abort before the first abra call, not halfway through a run."""
|
|
|
|
@pytest.fixture
|
|
def cli_env(self, config_root, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(alakazam, "abra", lambda *a, **k: calls.append(a) or "")
|
|
monkeypatch.setattr(alakazam, "fetch_recipes", lambda *a, **k: None)
|
|
monkeypatch.setattr(alakazam, "get_settings_path", lambda: str(config_root / "alakazam.yml"))
|
|
monkeypatch.setattr(alakazam, "get_abra_dir", lambda: config_root / ".abra")
|
|
(config_root / "alakazam.yml").write_text(f"root: {config_root}\n")
|
|
return calls
|
|
|
|
def test_clean_configs_run_through(self, config_root, cli_env):
|
|
result = CliRunner().invoke(alakazam.cli, [str(config_root / "group"), "ls"])
|
|
assert result.exit_code == 0, result.output
|
|
|
|
def test_encrypted_config_aborts_without_calling_abra(self, config_root, cli_env):
|
|
(config_root / "group" / "alaka-secrets.yml").write_bytes(GIT_CRYPT_HEADER)
|
|
result = CliRunner().invoke(alakazam.cli, [str(config_root / "group"), "ls"])
|
|
assert result.exit_code != 0
|
|
assert "alaka-secrets.yml" in result.output
|
|
assert "git crypt unlock" in result.output
|
|
assert cli_env == []
|
|
|
|
def test_encrypted_config_of_a_foreign_group_does_not_abort(self, config_root, cli_env):
|
|
foreign = config_root / "foreign"
|
|
foreign.mkdir()
|
|
(foreign / "alaka-secrets.yml").write_bytes(GIT_CRYPT_HEADER)
|
|
result = CliRunner().invoke(alakazam.cli, [str(config_root / "group"), "ls"])
|
|
assert result.exit_code == 0, result.output
|
|
|
|
def test_broken_settings_file_aborts(self, config_root, cli_env):
|
|
(config_root / "alakazam.yml").write_bytes(GIT_CRYPT_HEADER)
|
|
result = CliRunner().invoke(alakazam.cli, [str(config_root / "group"), "ls"])
|
|
assert result.exit_code != 0
|
|
assert "alakazam.yml" in result.output
|
|
|
|
|
|
class TestLogLevel:
|
|
"""-l has to win over the handler that an early logging.warning() installs by itself."""
|
|
|
|
@pytest.fixture
|
|
def probe(self, config_root, monkeypatch):
|
|
import logging
|
|
monkeypatch.setattr(alakazam, "abra", lambda *a, **k: "")
|
|
monkeypatch.setattr(alakazam, "fetch_recipes", lambda *a, **k: None)
|
|
monkeypatch.setattr(alakazam, "get_settings_path", lambda: str(config_root / "alakazam.yml"))
|
|
monkeypatch.setattr(alakazam, "get_abra_dir", lambda: config_root / ".abra")
|
|
(config_root / "alakazam.yml").write_text(f"root: {config_root}\n")
|
|
# a missing config-sets.yml makes read_config warn before cli() is through
|
|
(config_root / "config-sets.yml").unlink()
|
|
levels = []
|
|
|
|
@alakazam.cli.command()
|
|
def probe():
|
|
levels.append(logging.getLogger().getEffectiveLevel())
|
|
|
|
monkeypatch.setattr(alakazam, "GROUP_PATH", None)
|
|
return levels
|
|
|
|
def run(self, config_root, level):
|
|
return CliRunner().invoke(alakazam.cli, ["-l", level, str(config_root / "group"), "probe"])
|
|
|
|
def test_debug_reaches_the_root_logger(self, config_root, probe):
|
|
import logging
|
|
result = self.run(config_root, "DEBUG")
|
|
assert result.exit_code == 0, result.output
|
|
assert probe == [logging.DEBUG]
|
|
|
|
def test_info_reaches_the_root_logger(self, config_root, probe):
|
|
import logging
|
|
self.run(config_root, "INFO")
|
|
assert probe == [logging.INFO]
|
|
|
|
def test_an_invalid_level_is_rejected(self, config_root, probe):
|
|
result = self.run(config_root, "LOUD")
|
|
assert result.exit_code != 0
|