mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-09 04:06:42 +00:00
kevinveenbirkenbach
17918992c7
# Conflicts: # plugins/lookup/compose_file_args.py # roles/web-app-dashboard/tasks/02_manager_ops.yml # roles/web-app-dashboard/vars/main.yml
255 lines
8.8 KiB
Python
255 lines
8.8 KiB
Python
import importlib.util
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from ansible.errors import AnsibleError
|
|
|
|
from . import PROJECT_ROOT
|
|
|
|
|
|
def _load_module(rel_path: str, name: str):
|
|
path = PROJECT_ROOT / rel_path
|
|
spec = importlib.util.spec_from_file_location(name, str(path))
|
|
mod = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
class _TlsResolveStub:
|
|
def __init__(self, enabled: bool, mode: str):
|
|
self._enabled = enabled
|
|
self._mode = mode
|
|
|
|
def run(self, terms, variables=None, **kwargs):
|
|
return [{"enabled": self._enabled, "mode": self._mode}]
|
|
|
|
|
|
class _DomainsResolveStub:
|
|
def run(self, terms, variables=None, **kwargs):
|
|
return [(variables or {}).get("domains", {})]
|
|
|
|
|
|
class _CaInjectedStub:
|
|
def __init__(self, injected: bool):
|
|
self._injected = injected
|
|
|
|
def run(self, terms, variables=None, **kwargs):
|
|
return [self._injected]
|
|
|
|
|
|
class TestComposeFArgs(unittest.TestCase):
|
|
def setUp(self):
|
|
self.m = _load_module(
|
|
"plugins/lookup/compose_file_args.py",
|
|
"compose_file_args_mod",
|
|
)
|
|
self.lookup = self.m.LookupModule()
|
|
|
|
self.lookup._loader = object()
|
|
self.lookup._templar = object()
|
|
|
|
self.vars = {
|
|
"DIR_COMPOSITIONS": "/x/",
|
|
"domains": {
|
|
"web-app-a": "example.invalid",
|
|
},
|
|
}
|
|
|
|
def _loader_dispatch(self, tls_stub, ca_injected=None):
|
|
verdict = tls_stub.run(None)[0]
|
|
injected = (
|
|
(verdict["enabled"] and verdict["mode"] == "self_signed")
|
|
if ca_injected is None
|
|
else ca_injected
|
|
)
|
|
|
|
def _get(name, *args, **kwargs):
|
|
if name == "domains":
|
|
return _DomainsResolveStub()
|
|
if name == "tls":
|
|
return tls_stub
|
|
if name == "ca_injected":
|
|
return _CaInjectedStub(injected)
|
|
raise AssertionError(f"unexpected lookup name: {name}")
|
|
|
|
return _get
|
|
|
|
def test_a_domainless_ca_client_still_gets_the_ca_override_file(self):
|
|
with (
|
|
patch.object(
|
|
self.m.lookup_loader,
|
|
"get",
|
|
side_effect=self._loader_dispatch(
|
|
_TlsResolveStub(False, "off"), ca_injected=True
|
|
),
|
|
),
|
|
patch.object(
|
|
self.m, "get_docker_paths", side_effect=self._stub_get_docker_paths
|
|
),
|
|
patch.object(self.m, "_role_provides_override", return_value=False),
|
|
):
|
|
out = self.lookup.run(["web-app-a"], variables={**self.vars, "domains": {}})
|
|
self.assertIn(
|
|
"/x/compose.ca.override.yml",
|
|
out[0],
|
|
"ca_injected grants a domainless ca_client the override and the "
|
|
"handler writes it; a domain gate here leaves that file on disk "
|
|
"while the container runs without a trust store",
|
|
)
|
|
|
|
def _stub_get_docker_paths(self, application_id: str, base_dir: str) -> dict:
|
|
self.assertEqual(application_id, "web-app-a")
|
|
self.assertEqual(base_dir, "/x/")
|
|
return {
|
|
"files": {
|
|
"compose": "/x/compose.yml",
|
|
"compose_override": "/x/compose.override.yml",
|
|
"compose_ca_override": "/x/compose.ca.override.yml",
|
|
}
|
|
}
|
|
|
|
def test_includes_base_and_override_when_role_provides_override_and_tls_off(self):
|
|
with (
|
|
patch.object(
|
|
self.m, "get_docker_paths", side_effect=self._stub_get_docker_paths
|
|
),
|
|
patch.object(self.m, "_role_provides_override", return_value=True),
|
|
patch.object(
|
|
self.m.lookup_loader,
|
|
"get",
|
|
side_effect=self._loader_dispatch(_TlsResolveStub(False, "off")),
|
|
),
|
|
):
|
|
out = self.lookup.run(["web-app-a"], variables=self.vars)[0]
|
|
|
|
self.assertEqual(out, "-f /x/compose.yml -f /x/compose.override.yml")
|
|
|
|
def test_flag_kwarg_emits_compose_file_for_stack_deploy(self):
|
|
with (
|
|
patch.object(
|
|
self.m, "get_docker_paths", side_effect=self._stub_get_docker_paths
|
|
),
|
|
patch.object(self.m, "_role_provides_override", return_value=True),
|
|
patch.object(
|
|
self.m.lookup_loader,
|
|
"get",
|
|
side_effect=self._loader_dispatch(_TlsResolveStub(False, "off")),
|
|
),
|
|
):
|
|
out = self.lookup.run(
|
|
["web-app-a"], variables=self.vars, flag="--compose-file"
|
|
)[0]
|
|
|
|
self.assertEqual(
|
|
out, "--compose-file /x/compose.yml --compose-file /x/compose.override.yml"
|
|
)
|
|
|
|
def test_invalid_flag_rejected(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self.lookup.run(["web-app-a"], variables=self.vars, flag="-bad")
|
|
|
|
def test_includes_ca_override_when_self_signed_and_domain_exists(self):
|
|
with (
|
|
patch.object(
|
|
self.m, "get_docker_paths", side_effect=self._stub_get_docker_paths
|
|
),
|
|
patch.object(self.m, "_role_provides_override", return_value=True),
|
|
patch.object(
|
|
self.m.lookup_loader,
|
|
"get",
|
|
side_effect=self._loader_dispatch(_TlsResolveStub(True, "self_signed")),
|
|
),
|
|
):
|
|
out = self.lookup.run(["web-app-a"], variables=self.vars)[0]
|
|
|
|
self.assertEqual(
|
|
out,
|
|
"-f /x/compose.yml -f /x/compose.override.yml -f /x/compose.ca.override.yml",
|
|
)
|
|
|
|
def test_fails_when_ca_override_missing_but_required(self):
|
|
def stub_missing_ca(application_id: str, base_dir: str) -> dict:
|
|
return {
|
|
"files": {
|
|
"compose": "/x/compose.yml",
|
|
"compose_override": "/x/compose.override.yml",
|
|
"compose_ca_override": "",
|
|
}
|
|
}
|
|
|
|
with (
|
|
patch.object(self.m, "get_docker_paths", side_effect=stub_missing_ca),
|
|
patch.object(self.m, "_role_provides_override", return_value=True),
|
|
patch.object(
|
|
self.m.lookup_loader,
|
|
"get",
|
|
side_effect=self._loader_dispatch(_TlsResolveStub(True, "self_signed")),
|
|
),
|
|
self.assertRaises(AnsibleError),
|
|
):
|
|
self.lookup.run(["web-app-a"], variables=self.vars)
|
|
|
|
def test_includes_only_base_when_role_does_not_provide_override(self):
|
|
with (
|
|
patch.object(
|
|
self.m, "get_docker_paths", side_effect=self._stub_get_docker_paths
|
|
),
|
|
patch.object(self.m, "_role_provides_override", return_value=False),
|
|
patch.object(
|
|
self.m.lookup_loader,
|
|
"get",
|
|
side_effect=self._loader_dispatch(_TlsResolveStub(False, "off")),
|
|
),
|
|
):
|
|
out = self.lookup.run(["web-app-a"], variables=self.vars)[0]
|
|
|
|
self.assertEqual(out, "-f /x/compose.yml")
|
|
|
|
def test_role_provides_override_reads_the_role_of_the_given_application(self):
|
|
with (
|
|
tempfile.TemporaryDirectory() as role,
|
|
patch.object(self.m, "abs_role_path_by_application_id", return_value=role),
|
|
):
|
|
self.assertFalse(self.m._role_provides_override(application_id="x"))
|
|
(Path(role) / "templates").mkdir()
|
|
(Path(role) / "templates" / "compose.override.yml.j2").write_text("")
|
|
self.assertTrue(self.m._role_provides_override(application_id="x"))
|
|
|
|
def test_requires_one_term(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self.lookup.run([], variables=self.vars)
|
|
with self.assertRaises(AnsibleError):
|
|
self.lookup.run(["a", "b"], variables=self.vars)
|
|
|
|
def test_requires_path_docker_compose_instances(self):
|
|
with self.assertRaises(AnsibleError):
|
|
self.lookup.run(
|
|
["web-app-a"], variables={"domains": {"web-app-a": "example.invalid"}}
|
|
)
|
|
|
|
def test_requires_docker_compose_structure_from_get_docker_paths(self):
|
|
with (
|
|
patch.object(self.m, "get_docker_paths", return_value="nope"),
|
|
self.assertRaises(AnsibleError),
|
|
):
|
|
self.lookup.run(["web-app-a"], variables=self.vars)
|
|
|
|
with (
|
|
patch.object(self.m, "get_docker_paths", return_value={}),
|
|
self.assertRaises(AnsibleError),
|
|
):
|
|
self.lookup.run(["web-app-a"], variables=self.vars)
|
|
|
|
with (
|
|
patch.object(self.m, "get_docker_paths", return_value={"files": "nope"}),
|
|
self.assertRaises(AnsibleError),
|
|
):
|
|
self.lookup.run(["web-app-a"], variables=self.vars)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|