import sys import unittest import unittest.mock as mock from unittest.mock import patch from ansible.errors import AnsibleError import utils.tls_common as _tls_common from plugins.lookup.tls import LookupModule sys.modules.setdefault("ansible.module_utils.tls_common", _tls_common) class TestTlsResolveLookup(unittest.TestCase): def setUp(self): self.lookup = LookupModule() self.lookup._loader = mock.MagicMock() self.domains = { "web-app-a": "a.example", "web-app-b": ["b.example", "b-alt.example"], "web-app-c": {"primary": "c.example", "api": "api.c.example"}, } self.applications = { "web-app-a": {}, "web-app-b": {"server": {"tls": {"flavor": "self_signed"}}}, "web-app-c": {"server": {"tls": {"enabled": False}}}, } self.base_vars = { "domains": self.domains, "applications": self.applications, "TLS_ENABLED": True, "TLS_MODE": "letsencrypt", } # Route the "domains"/"applications" lookups through # variables['domains'] / variables['applications'] so tests stay hermetic. def _get(name, *_a, **_k): def _run(_terms, variables=None, **__): return [(variables or {}).get(name, {})] return mock.MagicMock(run=_run) loader_patcher = patch("plugins.lookup.tls.lookup_loader") loader_mock = loader_patcher.start() loader_mock.get.side_effect = _get self.addCleanup(loader_patcher.stop) def test_domain_term_auto_mode(self): out = self.lookup.run(["a.example"], variables=self.base_vars)[0] self.assertEqual(out["application_id"], "web-app-a") self.assertEqual(out["domain"], "a.example") self.assertTrue(out["enabled"]) self.assertEqual(out["mode"], "letsencrypt") self.assertEqual(out["protocols"]["web"], "https") self.assertEqual(out["ports"]["web"], 443) self.assertEqual(out["url"]["base"], "https://a.example/") def test_app_term_forced_app(self): out = self.lookup.run(["web-app-a"], variables=self.base_vars, mode="app")[0] self.assertEqual(out["application_id"], "web-app-a") self.assertEqual(out["domain"], "a.example") def test_app_override_flavor(self): out = self.lookup.run(["web-app-b"], variables=self.base_vars, mode="app")[0] self.assertEqual(out["mode"], "self_signed") def test_app_override_enabled_false(self): out = self.lookup.run(["web-app-c"], variables=self.base_vars, mode="app")[0] self.assertFalse(out["enabled"]) self.assertEqual(out["mode"], "off") self.assertEqual(out["protocols"]["web"], "http") self.assertEqual(out["ports"]["web"], 80) def test_domains_all_contains_primary_first(self): out = self.lookup.run(["web-app-b"], variables=self.base_vars, mode="app")[0] self.assertEqual(out["domains"]["primary"], "b.example") self.assertEqual(out["domains"]["all"], ["b.example", "b-alt.example"]) def test_want_path_returns_scalar(self): val = self.lookup.run( ["web-app-a", "url.base"], variables=self.base_vars, mode="app" )[0] self.assertEqual(val, "https://a.example/") def test_url_origin_is_the_base_url_without_its_trailing_slash(self): origin, base = ( self.lookup.run(["web-app-a", want], variables=self.base_vars, mode="app")[ 0 ] for want in ("url.origin", "url.base") ) self.assertEqual(origin, "https://a.example") self.assertEqual(f"{origin}/", base) def test_protocols_web_path_is_http_for_an_onion_only_app(self): onion = dict(self.base_vars) onion["domains"] = {**self.domains, "web-app-onion": "abc123.onion"} onion["applications"] = {**self.applications, "web-app-onion": {}} val = self.lookup.run( ["web-app-onion", "protocols.web"], variables=onion, mode="app" )[0] self.assertEqual(val, "http") def test_protocols_web_path_is_https_for_a_clearnet_app(self): val = self.lookup.run( ["web-app-a", "protocols.web"], variables=self.base_vars, mode="app" )[0] self.assertEqual(val, "https") def test_missing_required_vars(self): with self.assertRaises(AnsibleError): self.lookup.run(["web-app-a"], variables={}) def test_invalid_tls_mode_default(self): bad = dict(self.base_vars) bad["TLS_MODE"] = "invalid" with self.assertRaises(AnsibleError): self.lookup.run(["web-app-a"], variables=bad, mode="app") def test_invalid_forced_mode(self): with self.assertRaises(AnsibleError): self.lookup.run(["web-app-a"], variables=self.base_vars, mode="nope") def test_www_prefix_falls_back_to_bare_domain(self): out = self.lookup.run(["www.a.example"], variables=self.base_vars)[0] self.assertEqual(out["application_id"], "web-app-a") self.assertEqual(out["domain"], "a.example") def test_www_prefix_with_unknown_bare_still_raises(self): with self.assertRaises(AnsibleError) as ctx: self.lookup.run(["www.unknown.example"], variables=self.base_vars) self.assertIn("not found", str(ctx.exception)) def test_non_www_unregistered_still_raises(self): with self.assertRaises(AnsibleError) as ctx: self.lookup.run(["unknown.example"], variables=self.base_vars) self.assertIn("not found", str(ctx.exception)) def test_app_term_family_aligned_for_clearnet_consumer(self): v = dict(self.base_vars) v["domains"] = { **self.domains, "web-svc-cdn": ["cdn.abc.onion", "cdn.example"], } v["application_id"] = "web-app-a" out = self.lookup.run(["web-svc-cdn"], variables=v, mode="app")[0] self.assertEqual(out["domain"], "cdn.example") self.assertTrue(out["enabled"]) self.assertEqual(out["url"]["base"], "https://cdn.example/") def test_app_term_onion_consumer_stays_onion(self): v = dict(self.base_vars) v["domains"] = { **self.domains, "web-svc-cdn": ["cdn.abc.onion", "cdn.example"], "web-app-dashboard": ["dash.abc.onion"], } v["application_id"] = "web-app-dashboard" out = self.lookup.run(["web-svc-cdn"], variables=v, mode="app")[0] self.assertEqual(out["domain"], "cdn.abc.onion") self.assertFalse(out["enabled"]) self.assertEqual(out["url"]["base"], "http://cdn.abc.onion/") def test_domain_term_keeps_explicit_onion(self): v = dict(self.base_vars) v["domains"] = { **self.domains, "web-svc-cdn": ["cdn.abc.onion", "cdn.example"], } v["application_id"] = "web-app-a" out = self.lookup.run(["cdn.abc.onion"], variables=v)[0] self.assertEqual(out["domain"], "cdn.abc.onion") self.assertFalse(out["enabled"]) def test_app_term_onion_only_target_falls_back(self): v = dict(self.base_vars) v["domains"] = { **self.domains, "web-svc-cdn": ["cdn.abc.onion"], } v["application_id"] = "web-app-a" out = self.lookup.run(["web-svc-cdn"], variables=v, mode="app")[0] self.assertEqual(out["domain"], "cdn.abc.onion") self.assertFalse(out["enabled"]) def test_term_with_unresolved_jinja_is_templated(self): class _FakeTemplar: def __init__(self, mapping): self._mapping = mapping self.calls = [] def template(self, value): self.calls.append(value) return self._mapping.get(value, value) fake = _FakeTemplar({"{{ MY_DOMAIN }}": "a.example"}) self.lookup._templar = fake out = self.lookup.run(["{{ MY_DOMAIN }}"], variables=self.base_vars)[0] self.assertEqual(out["application_id"], "web-app-a") self.assertEqual(out["domain"], "a.example") self.assertIn("{{ MY_DOMAIN }}", fake.calls) if __name__ == "__main__": unittest.main()