mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 19:56:53 +00:00
kevinveenbirkenbachandClaude Opus 5
b6bf0b0bc1
sys-svc-compose writes compose.ca.override.yml for exactly lookup('ca_injected', application_id), which since 1254e27c2d also answers true for a service that declares ca_client and only calls the self-signed endpoint. compose_file_args rebuilt that predicate instead of asking it, and its copy still demanded a domain, so for a domainless ca_client the file was written and never reached --compose-file. Swarm takes its file list from that lookup; compose finds the file on disk by itself, which is why only the swarm jobs of run 36064589578 failed while the compose job passed.
The broker's container in that run carries no SSL_CERT_FILE and no CA mount, while openwebui, same job and same node, carries seven CA env vars and four binds. Every outbound call to Keycloak then fails the handshake, which the CLI test reports as 503 and Open WebUI relays as HTTP 400.
Two corrections in ca_injected itself. Its templar.template() call was dead: as_str strips the TrustedAsTemplate tag and ansible-core 2.19+ refuses to render an untrusted string, so the line could not resolve what it advertised, and tls.py and cert.py never template TLS_MODE either. And the ca_client branch short-circuited on the global TLS_MODE, skipping resolve_mode's per-app server.tls.mode override; it now applies only where there is no domain and so no per-app answer to get.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
243 lines
8.3 KiB
Python
243 lines
8.3 KiB
Python
import importlib.util
|
|
import unittest
|
|
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_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()
|