192 lines
9.2 KiB
Python
192 lines
9.2 KiB
Python
"""Tests for config-sets, in particular the ones that apply by themselves."""
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import alakazam
|
|
from alakazam import get_config_set_app_configs, merge_instance_configs, read_config_sets
|
|
|
|
INTEGRATION = {
|
|
"integrate-authentik-nextcloud": {
|
|
"apply-when": ["authentik", "nextcloud"],
|
|
"authentik": {"uncomment": ["compose.nextcloud.yml"]},
|
|
"nextcloud": {"uncomment": ["compose.authentik.yml"],
|
|
"shared_secrets": {"authentik": {"nextcloud_secret": "authentik_secret"}}},
|
|
}
|
|
}
|
|
|
|
|
|
def apply(config_sets, instance_apps, active=None):
|
|
"""The effective configuration, with the automatic sets below the explicit ones."""
|
|
automatic, explicit = get_config_set_app_configs(active or {}, config_sets, instance_apps)
|
|
return alakazam.merge_dict(automatic, explicit)
|
|
|
|
|
|
class TestApplyWhen:
|
|
def test_it_applies_when_every_app_is_there(self):
|
|
result = apply(INTEGRATION, ["authentik", "nextcloud", "traefik"])
|
|
assert result["authentik"]["uncomment"] == ["compose.nextcloud.yml"]
|
|
assert result["nextcloud"]["shared_secrets"] == {"authentik": {"nextcloud_secret": "authentik_secret"}}
|
|
|
|
def test_it_stays_out_when_one_app_is_missing(self):
|
|
assert apply(INTEGRATION, ["authentik", "traefik"]) == {}
|
|
|
|
def test_apply_when_is_not_passed_on_as_an_app(self):
|
|
"""It is a condition, not configuration, and would end up in an .env otherwise."""
|
|
assert "apply-when" not in apply(INTEGRATION, ["authentik", "nextcloud"])
|
|
|
|
def test_an_instance_can_switch_it_off(self):
|
|
assert apply(INTEGRATION, ["authentik", "nextcloud"],
|
|
active={"integrate-authentik-nextcloud": False}) == {}
|
|
|
|
def test_a_one_sided_integration_configures_only_one_app(self):
|
|
"""4 of the 17 pairs in the shipped combine.yml were one-sided."""
|
|
sets = {"integrate-matrix-synapse-traefik": {
|
|
"apply-when": ["matrix-synapse", "traefik"],
|
|
"traefik": {"uncomment": ["compose.matrix.yml"]}}}
|
|
assert list(apply(sets, ["matrix-synapse", "traefik"])) == ["traefik"]
|
|
assert apply(sets, ["traefik"]) == {}
|
|
|
|
|
|
class TestExplicitSets:
|
|
SET = {"bbb": {"authentik": {"env": {"A": "b"}}}}
|
|
|
|
def test_an_enabled_set_applies(self):
|
|
assert apply(self.SET, ["authentik"], active={"bbb": True})["authentik"]["env"] == {"A": "b"}
|
|
|
|
def test_a_set_that_is_not_enabled_does_not_apply(self):
|
|
assert apply(self.SET, ["authentik"]) == {}
|
|
|
|
def test_an_unknown_name_is_reported(self, caplog):
|
|
with caplog.at_level(logging.WARNING):
|
|
apply(self.SET, ["authentik"], active={"typo": True})
|
|
assert "'typo' is enabled but not defined" in caplog.text
|
|
|
|
def test_an_explicit_set_refines_an_automatic_one(self):
|
|
"""Automatic integrations are the base, an explicit set may still override them."""
|
|
sets = dict(INTEGRATION)
|
|
sets["override"] = {"authentik": {"env": {"SOURCE": "explicit"}}}
|
|
result = apply(sets, ["authentik", "nextcloud"], active={"override": True})
|
|
assert result["authentik"]["env"] == {"SOURCE": "explicit"}
|
|
assert result["authentik"]["uncomment"] == ["compose.nextcloud.yml"]
|
|
|
|
|
|
class TestReadConfigSets:
|
|
def write(self, root, name, text):
|
|
(root / name).write_text(text)
|
|
|
|
def test_the_base_file_is_read(self, tmp_path):
|
|
self.write(tmp_path, "config-sets.yml", "bbb:\n authentik:\n env:\n A: b\n")
|
|
assert read_config_sets(tmp_path)["bbb"]["authentik"]["env"] == {"A": "b"}
|
|
|
|
def test_further_files_are_merged(self, tmp_path):
|
|
self.write(tmp_path, "config-sets.yml", "bbb:\n authentik: {}\n")
|
|
self.write(tmp_path, "config-sets-authentik.yml", "integrate:\n apply-when: [a, b]\n")
|
|
assert sorted(read_config_sets(tmp_path)) == ["bbb", "integrate"]
|
|
|
|
def test_the_base_file_is_merged_first(self, tmp_path):
|
|
self.write(tmp_path, "config-sets.yml", "bbb:\n authentik:\n env:\n A: base\n")
|
|
self.write(tmp_path, "config-sets-later.yml", "bbb:\n authentik:\n env:\n A: later\n")
|
|
assert read_config_sets(tmp_path)["bbb"]["authentik"]["env"]["A"] == "later"
|
|
|
|
def test_a_root_without_config_sets(self, tmp_path):
|
|
assert read_config_sets(tmp_path) == {}
|
|
|
|
def test_a_key_defined_twice_is_reported(self, tmp_path, caplog):
|
|
self.write(tmp_path, "config-sets.yml", "bbb:\n authentik:\n env:\n A: base\n")
|
|
self.write(tmp_path, "config-sets-later.yml", "bbb:\n authentik:\n env:\n A: later\n")
|
|
with caplog.at_level(logging.WARNING):
|
|
read_config_sets(tmp_path)
|
|
assert "bbb.authentik.env.A" in caplog.text
|
|
|
|
|
|
class TestIntegrationReachesTheMergedConfig:
|
|
"""The instance config is what every command reads, the integration has to arrive there."""
|
|
|
|
def merge(self, apps):
|
|
return merge_instance_configs({}, "example.com", {app: None for app in apps}, INTEGRATION)
|
|
|
|
def test_both_apps_get_their_side(self):
|
|
merged = self.merge(["authentik", "nextcloud"])
|
|
assert merged["authentik"]["uncomment"] == ["compose.nextcloud.yml"]
|
|
assert merged["nextcloud"]["uncomment"] == ["compose.authentik.yml"]
|
|
|
|
def test_a_lone_app_gets_nothing(self):
|
|
assert "uncomment" not in self.merge(["authentik"])["authentik"]
|
|
|
|
|
|
class TestUnknownAppKeys:
|
|
"""A misspelt app key drops its configuration, which is only visible if it is reported."""
|
|
|
|
def test_a_typo_in_an_automatic_set_is_reported(self, caplog):
|
|
sets = {"integrate": {"apply-when": ["authentik", "nextcloud"],
|
|
"nextcoud": {"env": {"A": "b"}}}}
|
|
with caplog.at_level(logging.WARNING):
|
|
merge_instance_configs({}, "example.com", {"authentik": None, "nextcloud": None}, sets)
|
|
assert "'nextcoud' is not listed" in caplog.text
|
|
|
|
def test_a_typo_in_an_explicit_set_is_reported(self, caplog):
|
|
sets = {"bbb": {"nextcoud": {"env": {"A": "b"}}}}
|
|
with caplog.at_level(logging.WARNING):
|
|
merge_instance_configs({}, "example.com",
|
|
{"nextcloud": None, "CONFIG-SETS": {"bbb": True}}, sets)
|
|
assert "'nextcoud' is not listed" in caplog.text
|
|
|
|
def test_a_correct_app_key_is_not_reported(self, caplog):
|
|
sets = {"integrate": {"apply-when": ["authentik", "nextcloud"],
|
|
"nextcloud": {"env": {"A": "b"}}}}
|
|
with caplog.at_level(logging.WARNING):
|
|
merge_instance_configs({}, "example.com", {"authentik": None, "nextcloud": None}, sets)
|
|
assert caplog.text == ""
|
|
|
|
|
|
class TestPrecedence:
|
|
"""An integration is the lowest layer, as combine.yml was before it."""
|
|
|
|
def test_the_group_configuration_beats_an_integration(self):
|
|
group = {"nextcloud": {"env": {"SOURCE": "group"}}}
|
|
sets = {"integrate": {"apply-when": ["authentik", "nextcloud"],
|
|
"nextcloud": {"env": {"SOURCE": "integration"}}}}
|
|
merged = merge_instance_configs(group, "example.com", {"authentik": None, "nextcloud": None}, sets)
|
|
assert merged["nextcloud"]["env"]["SOURCE"] == "group"
|
|
|
|
def test_the_instance_beats_everything(self):
|
|
group = {"nextcloud": {"env": {"SOURCE": "group"}}}
|
|
sets = {"integrate": {"apply-when": ["authentik", "nextcloud"],
|
|
"nextcloud": {"env": {"SOURCE": "integration"}}}}
|
|
merged = merge_instance_configs(
|
|
group, "example.com", {"authentik": None, "nextcloud": {"env": {"SOURCE": "instance"}}}, sets)
|
|
assert merged["nextcloud"]["env"]["SOURCE"] == "instance"
|
|
|
|
def test_an_integration_still_adds_what_nobody_else_sets(self):
|
|
group = {"nextcloud": {"env": {"OTHER": "group"}}}
|
|
sets = {"integrate": {"apply-when": ["authentik", "nextcloud"],
|
|
"nextcloud": {"env": {"SOURCE": "integration"}}}}
|
|
merged = merge_instance_configs(group, "example.com", {"authentik": None, "nextcloud": None}, sets)
|
|
assert merged["nextcloud"]["env"] == {"SOURCE": "integration", "OTHER": "group"}
|
|
|
|
|
|
class TestHookOrder:
|
|
"""An integration adds to an app that already configures itself, and must come after it."""
|
|
|
|
SETS = {"integrate-authentik-wordpress": {
|
|
"apply-when": ["authentik", "wordpress"],
|
|
"wordpress": {"initial-hooks": ["app set_authentik"]}}}
|
|
|
|
def test_the_integration_hook_runs_last(self):
|
|
"""set_authentik configures SSO in a WordPress that core_install has to create first."""
|
|
group = {"wordpress": {"initial-hooks": ["app core_install", "app enable_auto_updates"]}}
|
|
merged = merge_instance_configs(group, "example.com", {"authentik": None, "wordpress": None}, self.SETS)
|
|
assert merged["wordpress"]["initial-hooks"] == [
|
|
"app core_install", "app enable_auto_updates", "app set_authentik"]
|
|
|
|
def test_an_instance_hook_also_comes_first(self):
|
|
instance = {"authentik": None, "wordpress": {"initial-hooks": ["app from_instance"]}}
|
|
merged = merge_instance_configs({}, "example.com", instance, self.SETS)
|
|
assert merged["wordpress"]["initial-hooks"] == ["app from_instance", "app set_authentik"]
|