Files
computer-playbook/tests/unit/python/plugins/lookup/test_roles_with_service.py
T
kevinveenbirkenbachandClaude Opus 5 7162b7bf4c Merge fork/feature/svc-net-tor into feature/ai-agent-employees
834 commits from the Tor onion-service branch against 299 of ours, from merge
base 7b646bdece: 127 conflicted files, 3506 files changed.

The two sides are largely orthogonal. Theirs adds svc-net-tor and a `tor` service
across the role tree, plus two sweeping refactors: shipped scripts sorted by
language (files/ruby/, files/python/, files/shell/, scripts/lint/*.sh) and the
unit suite moved under tests/unit/python/. Ours is requirement 036, the native AI
gateway, plus the MCP role work. Most conflicts were both sides appending to the
same service map, resolved as unions.

Resolutions that were not unions:

Credentials moved namespace. Theirs renamed meta/schema.yml to meta/secrets.yml
and re-addressed credentials as secrets.credentials.<name>. Git carried our
litellm_api_key blocks through the rename, but every consumer we added still read
'credentials.<name>', which no longer resolves. All 45 call sites are migrated.

plugins/filter/container_volumes.py is deleted by theirs and relocated to
plugins/lookup/container_volumes.py, where it carries the same tmpfs defect our
f85996b43 had just fixed: a hard-coded {"size": ...} that silently drops a per
mount `mode`. The fix is re-applied in the relocated file, as that commit's merge
note anticipated.

compose.yml keeps only INFINITO_PLAYWRIGHT_REPORTS_BASE_DIR. Theirs also forwards
INFINITO_PLAYWRIGHT_STAGE_BASE_DIR; that line is deliberately dropped, per the
analysis recorded in 45804e9ad -- STAGE has a drift-proof role-vars fallback,
while forwarding it promotes a setdefault-pinnable .env value above that SPOT.

tasks/stages/01_constructor.yml takes their include, ./tasks/utils/setup/group.yml,
because their refactor removed tasks/groups/*-roles.yml, leaving our include
target nonexistent. Our task name is kept, since svc-virt is in the loop.

Makefile swarm-zombie is a union: their validate plus teardown, then our
lab_subnet cleanup, which their side dropped while leaving lab_subnet.sh as the
only orphaned reference in the file.

01_bootstrap.sh keeps our deterministic --ip pinning, which distinguishes an
already-attached node from a real failure, over their `|| true`.

Their language-sorted layout orphaned references our side still held: zammad's
two scripts/apply/*.rb, jenkins' java-with-ca-trust.sh, and the notify topics
their handlers renamed to hyphenated form. All corrected.

roles/sys-svc-compose/files/shell/swarm/stack_ready.sh gains our latched-update
block: paused, rollback_paused, rollback_started and rollback_completed exit 2
rather than polling a state that cannot resolve without another deploy.

config_from_run is unioned to (title, logged, jobs); the production caller in
trigger/__main__.py now passes jobs, without which our suite suppression would
have been inert.

READMEs are regenerated rather than hand-merged: 15 LiteLLM rows and 91 Tor rows
now derive from the merged metadata, which is itself evidence the meta merge is
coherent.

Verified: 5161 unit tests pass. The single failure, test_query.py TestSortSpec,
also fails on fork/feature/svc-net-tor at b51d2f9919, measured in a baseline
worktree -- so this merge introduces no new unit-test failure. Two modules are
excluded from that run because `baudolo`, a dependency their branch introduces,
is not installed on this host.
Not verified: no deploy has run against the merged tree, and make quality has not
been run end to end here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-24 16:32:30 +02:00

566 lines
21 KiB
Python

"""Unit tests for ``plugins/lookup/roles_with_service.py``.
The lookup enumerates consumer roles for a given service by walking the
merged applications mapping. The applications loader is mocked so the
tests stay hermetic — no filesystem access into ``roles/``.
"""
import importlib.util
import unittest
import unittest.mock as mock
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 _DummyTemplar:
def __init__(self, available_variables):
self.available_variables = available_variables
class RolesWithServiceLookupTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.mod = _load_module(
"plugins/lookup/roles_with_service.py",
"lookup_roles_with_service",
)
def _make_lookup(self, available_vars: dict | None = None):
lm = self.mod.LookupModule()
lm._templar = _DummyTemplar(available_vars or {})
lm._loader = mock.MagicMock()
return lm
def _run(self, terms, applications: dict, vars_: dict | None = None, **kwargs):
"""Run the lookup with the ``applications`` and ``tls`` sub-lookups
patched to stay hermetic. The stubbed tls resolves
``[role_id, "url.base"]`` to a deterministic URL so the lookup
can finish building each result entry without the real domain
machinery."""
lookup = self._make_lookup(vars_ or {})
class _StubTls:
def run(self, terms_, variables=None, **kwargs):
return [f"https://{terms_[0]}.example.com"]
class _StubApplications:
def run(self, terms_, variables=None, **kwargs):
return [applications]
def _dispatch(name, **_kwargs):
return _StubApplications() if name == "applications" else _StubTls()
with patch.object(self.mod.lookup_loader, "get", side_effect=_dispatch):
return lookup.run(terms, variables=vars_ or {}, **kwargs)
def test_zero_terms_raises(self):
with self.assertRaises(AnsibleError):
self._run([], {})
def test_two_terms_raises(self):
with self.assertRaises(AnsibleError):
self._run(["dashboard", "extra"], {})
def test_blank_service_name_raises(self):
with self.assertRaises(AnsibleError):
self._run([" "], {})
def test_consumer_with_truthy_enabled_and_shared_is_included(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {"enabled": True, "shared": True},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["dashboard"], applications)
self.assertEqual(
result,
[
[
{
"id": "web-app-foo",
"canonical_domain": "foo.example.com",
"canonical_url": "https://web-app-foo.example.com",
"iframe": True,
}
]
],
)
def test_iframe_defaults_to_enabled_when_unset(self):
applications = {
"web-app-foo": {
"services": {"dashboard": {"enabled": True, "shared": True}},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["dashboard"], applications)[0]
self.assertTrue(result[0]["iframe"])
def test_iframe_false_is_preserved(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {
"enabled": True,
"shared": True,
"iframe": False,
},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["dashboard"], applications)[0]
self.assertFalse(result[0]["iframe"])
def test_falsy_enabled_excludes_role(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {"enabled": False, "shared": True},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
self.assertEqual(self._run(["dashboard"], applications), [[]])
def test_falsy_shared_excludes_role(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {"enabled": True, "shared": False},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
self.assertEqual(self._run(["dashboard"], applications), [[]])
def test_scrape_false_excludes_role_but_keeps_other_consumers(self):
applications = {
"web-svc-asset": {
"services": {
"prometheus": {
"enabled": True,
"shared": True,
"scrape": False,
},
},
"domains": {"canonical": ["asset.example.com"]},
},
"web-app-foo": {
"services": {
"prometheus": {"enabled": True, "shared": True},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["prometheus"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-foo"])
def test_scrape_true_keeps_role(self):
applications = {
"web-app-foo": {
"services": {
"prometheus": {
"enabled": True,
"shared": True,
"scrape": True,
},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["prometheus"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-foo"])
def test_track_false_excludes_role_but_keeps_other_consumers(self):
applications = {
"web-svc-cdn": {
"services": {
"matomo": {
"enabled": True,
"shared": True,
"track": False,
},
},
"domains": {"canonical": ["cdn.example.com"]},
},
"web-app-foo": {
"services": {
"matomo": {"enabled": True, "shared": True},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["matomo"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-foo"])
def test_track_true_keeps_role(self):
applications = {
"web-app-foo": {
"services": {
"matomo": {
"enabled": True,
"shared": True,
"track": True,
},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["matomo"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-foo"])
def test_missing_service_block_excludes_role(self):
applications = {
"web-app-foo": {
"services": {"oidc": {"enabled": True}},
"domains": {"canonical": ["foo.example.com"]},
},
}
self.assertEqual(self._run(["dashboard"], applications), [[]])
def test_role_without_canonical_domain_is_skipped(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {"enabled": True, "shared": True},
},
"domains": {},
},
}
self.assertEqual(self._run(["dashboard"], applications), [[]])
def test_canonical_as_string_is_accepted(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {"enabled": True, "shared": True},
},
"domains": {"canonical": "foo.example.com"},
},
}
result = self._run(["dashboard"], applications)
self.assertEqual(
result,
[
[
{
"id": "web-app-foo",
"canonical_domain": "foo.example.com",
"canonical_url": "https://web-app-foo.example.com",
"iframe": True,
}
]
],
)
def test_results_are_sorted_by_role_id(self):
applications = {
"web-app-zeta": {
"services": {"prometheus": {"enabled": True, "shared": True}},
"domains": {"canonical": ["zeta.example.com"]},
},
"web-app-alpha": {
"services": {"prometheus": {"enabled": True, "shared": True}},
"domains": {"canonical": ["alpha.example.com"]},
},
"web-app-mu": {
"services": {"prometheus": {"enabled": True, "shared": True}},
"domains": {"canonical": ["mu.example.com"]},
},
}
result = self._run(["prometheus"], applications)[0]
self.assertEqual(
[r["id"] for r in result],
["web-app-alpha", "web-app-mu", "web-app-zeta"],
)
def test_different_services_resolve_independently(self):
applications = {
"web-app-foo": {
"services": {
"dashboard": {"enabled": True, "shared": True},
"matomo": {"enabled": False, "shared": True},
},
"domains": {"canonical": ["foo.example.com"]},
},
"web-app-bar": {
"services": {
"dashboard": {"enabled": False, "shared": True},
"matomo": {"enabled": True, "shared": True},
},
"domains": {"canonical": ["bar.example.com"]},
},
}
dashboard_consumers = self._run(["dashboard"], applications)[0]
matomo_consumers = self._run(["matomo"], applications)[0]
self.assertEqual(
[r["id"] for r in dashboard_consumers],
["web-app-foo"],
)
self.assertEqual(
[r["id"] for r in matomo_consumers],
["web-app-bar"],
)
def test_non_dict_app_config_skipped(self):
applications = {
"web-app-foo": "not-a-dict",
"web-app-bar": {
"services": {"dashboard": {"enabled": True, "shared": True}},
"domains": {"canonical": ["bar.example.com"]},
},
}
result = self._run(["dashboard"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-bar"])
def test_canonical_with_empty_first_entry_raises(self):
applications = {
"web-app-foo": {
"services": {"dashboard": {"enabled": True, "shared": True}},
"domains": {"canonical": ["", "foo.example.com"]},
},
}
with self.assertRaises(AnsibleError):
self._run(["dashboard"], applications)
def test_canonical_mapping_resolves_to_its_first_domain(self):
applications = {
"web-app-foo": {
"services": {"dashboard": {"enabled": True, "shared": True}},
"domains": {"canonical": {"foo": "foo.example.com"}},
},
}
result = self._run(["dashboard"], applications)[0]
self.assertEqual([r["canonical_domain"] for r in result], ["foo.example.com"])
def test_unrendered_jinja_strings_are_treated_as_truthy(self):
"""In a real Ansible play, ``get_merged_applications`` runs the
merged payload through ``_render_with_templar`` so a value like
``"{{ 'web-app-dashboard' in group_names }}"`` arrives at this
lookup as the literal ``True`` / ``False`` it resolved to.
Without a templar (CLI smoke tests, scripts that call
``get_merged_applications(templar=None)``), the raw Jinja string
passes through unchanged. Python's ``bool()`` on a non-empty
string is ``True``, so the lookup is over-inclusive on that
path. This test pins that contract: callers MUST run the
applications payload through a templar before passing it in,
otherwise every role with both keys present is returned
regardless of deployment state.
"""
applications = {
"web-app-foo": {
"services": {
"dashboard": {
"enabled": "{{ 'web-app-dashboard' in group_names }}",
"shared": "{{ 'web-app-dashboard' in group_names }}",
},
},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["dashboard"], applications)[0]
self.assertEqual(
[r["id"] for r in result],
["web-app-foo"],
"non-empty Jinja string is truthy under bool(); the lookup "
"relies on a templar-rendered payload for correct filtering",
)
def test_group_names_scopes_to_deployed_consumers(self):
applications = {
"web-app-foo": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["foo.example.com"]},
},
"web-app-bar": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["bar.example.com"]},
},
}
result = self._run(
["logout"], applications, vars_={"group_names": ["web-app-foo"]}
)[0]
self.assertEqual([r["id"] for r in result], ["web-app-foo"])
def test_scope_all_ignores_group_names(self):
applications = {
"web-svc-logout": {
"services": {"matomo": {"enabled": True, "shared": True}},
"domains": {"canonical": ["logout.example.com"]},
},
}
gn = {"group_names": ["web-app-matomo"]}
self.assertEqual(
[r["id"] for r in self._run(["matomo"], applications, vars_=gn)[0]],
[],
"default host scope drops a consumer absent from this host's group_names",
)
all_scoped = self._run(["matomo"], applications, vars_=gn, scope="all")[0]
self.assertEqual(
[r["id"] for r in all_scoped],
["web-svc-logout"],
"scope='all' returns deployment-scoped consumers regardless of group_names",
)
def test_deployment_scope_spans_hosts_but_not_the_whole_repository(self):
applications = {
"web-app-foo": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["foo.example.com"]},
},
"web-app-bar": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["bar.example.com"]},
},
"web-app-undeployed": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["undeployed.example.com"]},
},
}
vars_ = {
"group_names": ["web-app-foo"],
"groups": {
"web-app-foo": ["host-a"],
"web-app-bar": ["host-b"],
"web-app-undeployed": [],
},
}
result = self._run(["logout"], applications, vars_=vars_, scope="deployment")[0]
self.assertEqual(
sorted(r["id"] for r in result),
["web-app-bar", "web-app-foo"],
"a role deployed on another host of the same deployment is "
"reachable; a role in the repository but in no group is not",
)
def test_self_provider_is_excluded(self):
applications = {
"web-svc-logout": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["logout.example.com"]},
},
"web-app-foo": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["foo.example.com"]},
},
}
result = self._run(["logout"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-foo"])
def test_absent_group_names_keeps_all_consumers(self):
applications = {
"web-app-foo": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["foo.example.com"]},
},
"web-app-bar": {
"services": {"logout": {"enabled": True, "shared": True}},
"domains": {"canonical": ["bar.example.com"]},
},
}
result = self._run(["logout"], applications)[0]
self.assertEqual([r["id"] for r in result], ["web-app-bar", "web-app-foo"])
def _mcp_role(self, direction: str, **mcp_extra) -> dict:
mcp = {
"enabled": True,
"shared": True,
"direction": direction,
"transport": "streamable_http",
"auth": "bearer_token",
"auth_subject": "user",
"endpoint": {
"service_key": "baserow",
"path": "/mcp",
"port_key": "http",
},
}
mcp.update(mcp_extra)
return {
"services": {
"mcp": mcp,
"baserow": {"ports": {"local": {"http": 8021}}},
},
"domains": {"canonical": ["srv.example.com"]},
}
def test_direction_filter_excludes_other_direction(self):
applications = {
"web-app-server": self._mcp_role("server"),
"web-app-client": self._mcp_role("client"),
}
result = self._run(["mcp"], applications, direction="server")[0]
self.assertEqual([r["id"] for r in result], ["web-app-server"])
def test_direction_filter_includes_both(self):
applications = {"web-app-duplex": self._mcp_role("both")}
result = self._run(["mcp"], applications, direction="server")[0]
self.assertEqual([r["id"] for r in result], ["web-app-duplex"])
def test_direction_entry_carries_endpoint_metadata(self):
applications = {"web-app-server": self._mcp_role("server")}
entry = self._run(["mcp"], applications, direction="server")[0][0]
self.assertEqual(entry["transport"], "streamable_http")
self.assertEqual(entry["auth"], "bearer_token")
self.assertEqual(entry["auth_subject"], "user")
self.assertEqual(
entry["endpoint"],
{
"service_key": "baserow",
"path": "/mcp",
"port": 8021,
"key_credential": None,
"suffix": None,
},
)
def test_direction_endpoint_port_falls_back_to_internal(self):
role = self._mcp_role("server")
role["services"]["baserow"] = {"ports": {"internal": {"http": 80}}}
applications = {"web-app-server": role}
entry = self._run(["mcp"], applications, direction="server")[0][0]
self.assertEqual(entry["endpoint"]["port"], 80)
def test_the_internal_port_wins_without_any_exposure_declaration(self):
role = self._mcp_role("server")
role["services"]["baserow"]["ports"]["internal"] = {"http": 80}
applications = {"web-app-server": role}
entry = self._run(["mcp"], applications, direction="server")[0][0]
self.assertEqual(entry["endpoint"]["port"], 80)
def test_an_exposure_declaration_does_not_change_the_port(self):
role = self._mcp_role("server")
role["services"]["mcp"]["exposure"] = "public"
role["services"]["baserow"]["ports"]["internal"] = {"http": 80}
applications = {"web-app-server": role}
entry = self._run(["mcp"], applications, direction="server")[0][0]
self.assertEqual(entry["endpoint"]["port"], 80)
def test_without_direction_kwarg_shape_is_unchanged(self):
applications = {"web-app-server": self._mcp_role("server")}
entry = self._run(["mcp"], applications)[0][0]
self.assertEqual(
sorted(entry), ["canonical_domain", "canonical_url", "id", "iframe"]
)
if __name__ == "__main__":
unittest.main()