perf(plugins): share template_shallow and render the front injectors through it

ansible-core's template lookup deep-copies every task variable on each
call. The role-local template_shallow of web-opt-rdr-domains, which hands
the builtin a top-level copy whose values stay shared, moves to
plugins/lookup so every role can use it. The nine template lookups of the
front injectors switch to it: six in sys-front-inj-design and one each in
sys-front-inj-dashboard, sys-front-inj-logout and sys-front-inj-matomo.
None of the switched templates or the lookups they call writes into the
task variables.

Tests:
- the plugin renders like the builtin, never deep-copies the caller's
  values and leaves the caller's variables unchanged
- for every role with design sources, clearnet and onion vhosts, with and
  without a logo, the design facts are identical to the builtin's, or it
  fails with the identical error
- the dashboard, logout and matomo code renders identically
- the redirect loop of web-opt-rdr-domains still renders through the
  shared plugin

The saving per deploy is not measured by a timed deploy yet.

🎯 Fix: 90 %
💥 Break: 3 %
🌱 Efficiency: –
⚡ Speed: –
🛡️ Security: ~100 %
🏆 Score: 87
🚦 Gate: ✅ 90/5, efficiency and speed <= 100, security 100 %

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

roles/sys-front-inj-dashboard/tasks/main.yml

@@ -4,7 +4,7 @@
 
 - name: "📄 Load iFrame init code for '{{ application_id }}'"
   set_fact:
-    iframe_init_code: "{{ lookup('template','iframe/init_one_liner.js.j2') }}"
+    iframe_init_code: "{{ lookup('template_shallow', 'iframe/init_one_liner.js.j2') }}"
 
 - name: "🗜️ Collapse iFrame init code into one-liner for '{{ application_id }}'"
   set_fact:

roles/sys-front-inj-design/tasks/main.yml

@@ -15,13 +15,13 @@
            'files': ['files/logo.css'],
            'paths': [app_role_dir]
          }, errors='ignore') | default('', true) }}
-    app_body_css: "{{ (lookup('template', _app_style_src) | trim) if _app_style_src | length > 0 else '' }}"
-    app_logo_css: "{{ (lookup('template', _app_logo_src) | trim) if _app_logo_src | length > 0 else '' }}"
+    app_body_css: "{{ (lookup('template_shallow', _app_style_src) | trim) if _app_style_src | length > 0 else '' }}"
+    app_logo_css: "{{ (lookup('template_shallow', _app_logo_src) | trim) if _app_logo_src | length > 0 else '' }}"
     _app_compose_src: "{{ lookup('path_absolute', 'roles', 'sys-front-inj-design', 'templates', 'css', 'app.css.j2') }}"
     _cdn_root: "{{ lookup('cdn', application_id).urls.root }}"
     _cdn_origin: "{{ _cdn_root | urlsplit('scheme') }}://{{ _cdn_root | urlsplit('netloc') }}/"
     _app_style_css: >-
-      {{ (lookup('template', _app_compose_src,
+      {{ (lookup('template_shallow', _app_compose_src,
                  template_vars={'app_body_css': app_body_css, 'app_logo_css': app_logo_css})
           | trim | replace(_cdn_origin, '/'))
          if (app_body_css ~ app_logo_css) | length > 0 else '' }}
@@ -64,12 +64,12 @@
          }, errors='ignore') | default('', true) }}
     _design_compose_src: "{{ lookup('path_absolute', 'roles', 'sys-front-inj-design', 'templates', 'js', 'design.js.j2') }}"
     _design_config_src: "{{ lookup('path_absolute', 'roles', 'sys-front-inj-design', 'templates', 'js', 'design-config.json.j2') }}"
-    design_body_js: "{{ (lookup('template', _design_js_src) | trim) if _design_js_src | length > 0 else '' }}"
+    design_body_js: "{{ (lookup('template_shallow', _design_js_src) | trim) if _design_js_src | length > 0 else '' }}"
     _design_js: >-
-      {{ (lookup('template', _design_compose_src, template_vars={'design_body_js': design_body_js}) | trim)
+      {{ (lookup('template_shallow', _design_compose_src, template_vars={'design_body_js': design_body_js}) | trim)
          if 'INFINITO_DESIGN' in design_body_js else design_body_js }}
   set_fact:
-    design_config_json: "{{ (lookup('template', _design_config_src) | trim) if 'INFINITO_DESIGN' in design_body_js else '' }}"
+    design_config_json: "{{ (lookup('template_shallow', _design_config_src) | trim) if 'INFINITO_DESIGN' in design_body_js else '' }}"
     design_js_one_liner: "{{ _design_js | to_one_liner }}"
     design_css_states: "{{ design_css_states | combine({css_app_state_key: css_app_state}) }}"
 

roles/sys-front-inj-logout/tasks/main.yml

@@ -5,7 +5,7 @@
 
 - name: "📄 Load logout code for '{{ application_id }}'"
   set_fact:
-    logout_code: "{{ lookup('template', 'logout_one_liner.js.j2') }}"
+    logout_code: "{{ lookup('template_shallow', 'logout_one_liner.js.j2') }}"
   changed_when: false
 
 - name: "🗜️ Collapse logout code into one-liner for '{{ application_id }}'"

roles/sys-front-inj-matomo/tasks/inject.yml

@@ -1,7 +1,7 @@
 ---
 - name: 📄 Load Matomo tracking JS template
   set_fact:
-    matomo_tracking_code: "{{ lookup('template','matomo-tracking.js.j2') }}"
+    matomo_tracking_code: "{{ lookup('template_shallow', 'matomo-tracking.js.j2') }}"
 
 - name: 🗜️ Collapse Matomo code into one-liner
   set_fact:

tests/unit/python/plugins/lookup/test_template_shallow.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from ansible.errors import AnsibleError
+from ansible.parsing.dataloader import DataLoader
+from ansible.plugins.loader import lookup_loader
+from ansible.template import Templar, trust_as_template
+
+from . import PROJECT_ROOT
+
+PROBE_PLUGIN = """
+from ansible.plugins.lookup import LookupBase
+
+
+class LookupModule(LookupBase):
+    def run(self, terms, variables, **kwargs):
+        return [variables["domain"]]
+"""
+MUTATING_PLUGIN = """
+from ansible.plugins.lookup import LookupBase
+
+
+class LookupModule(LookupBase):
+    def run(self, terms, variables, **kwargs):
+        variables["nested"]["hosts"].append("leaked.example")
+        return [""]
+"""
+TEMPLATES = {
+    "vhost.j2": "{{ domain }} {{ lookup('shallow_probe_variables_domain') }}\n",
+    "broken.j2": "{{ domain }} {{ undefined_in_template }}\n",
+    "mutating.j2": "{{ lookup('shallow_probe_mutates_nested') }}\n",
+}
+
+
+class _NoDeepCopy:
+    def __deepcopy__(self, memo):
+        raise AssertionError("the variables were deep-copied")
+
+
+def _causes(error: BaseException | None) -> list[str]:
+    causes = []
+    while error is not None:
+        causes.append(str(error))
+        error = error.__cause__
+    return causes
+
+
+class TestTemplateShallow(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls._tmp = tempfile.TemporaryDirectory()
+        tmp = Path(cls._tmp.name)
+        (tmp / "plugins").mkdir()
+        (tmp / "plugins" / "shallow_probe_variables_domain.py").write_text(PROBE_PLUGIN)
+        (tmp / "plugins" / "shallow_probe_mutates_nested.py").write_text(
+            MUTATING_PLUGIN
+        )
+        (tmp / "templates").mkdir()
+        for name, body in TEMPLATES.items():
+            (tmp / "templates" / name).write_text(body)
+        lookup_loader.add_directory(str(PROJECT_ROOT / "plugins" / "lookup"))
+        lookup_loader.add_directory(str(tmp / "plugins"))
+        cls.search_path = [str(tmp)]
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls._tmp.cleanup()
+
+    def _variables(self) -> dict:
+        return {
+            "domain": "play.example",
+            "nested": {"hosts": ["a.example"]},
+            "ansible_search_path": self.search_path,
+        }
+
+    def _render(self, plugin: str, variables: dict, template: str = "vhost.j2"):
+        return Templar(loader=DataLoader(), variables=variables).template(
+            trust_as_template(
+                "{{ lookup('" + plugin + "', '" + template + "', "
+                "template_vars={'domain': 'source.example'}) }}"
+            )
+        )
+
+    def test_a_lookup_inside_the_template_reads_the_per_call_domain(self) -> None:
+        rendered = self._render("template_shallow", self._variables())
+        self.assertEqual(rendered, "source.example source.example\n")
+        self.assertEqual(rendered, self._render("template", self._variables()))
+
+    def test_the_caller_variables_keep_their_values_and_identity(self) -> None:
+        variables = self._variables()
+        nested = variables["nested"]
+        self._render("template_shallow", variables)
+        self.assertEqual(variables, self._variables())
+        self.assertIs(variables["nested"], nested)
+
+    def test_no_value_of_the_caller_is_deep_copied(self) -> None:
+        variables = {**self._variables(), "untouchable": _NoDeepCopy()}
+        self.assertEqual(
+            self._render("template_shallow", variables),
+            "source.example source.example\n",
+        )
+
+    def test_the_builtin_template_lookup_deep_copies_every_value(self) -> None:
+        variables = {**self._variables(), "untouchable": _NoDeepCopy()}
+        with self.assertRaises(AnsibleError) as raised:
+            self._render("template", variables)
+        causes = _causes(raised.exception)
+        self.assertTrue(any("deep-copied" in cause for cause in causes), causes)
+
+    def test_a_lookup_mutating_a_nested_value_leaks_only_through_the_shallow_copy(
+        self,
+    ) -> None:
+        builtin = self._variables()
+        self._render("template", builtin, "mutating.j2")
+        self.assertEqual(builtin, self._variables())
+        shallow = self._variables()
+        self._render("template_shallow", shallow, "mutating.j2")
+        self.assertEqual(shallow["nested"]["hosts"], ["a.example", "leaked.example"])
+
+    def test_an_undefined_variable_fails_with_the_builtin_message(self) -> None:
+        messages = {}
+        for plugin in ("template", "template_shallow"):
+            with self.assertRaises(AnsibleError) as raised:
+                self._render(plugin, self._variables(), "broken.j2")
+            messages[plugin] = [
+                cause.replace("template_shallow", "template")
+                for cause in _causes(raised.exception)
+            ]
+        self.assertIn("undefined_in_template", " ".join(messages["template"]))
+        self.assertEqual(messages["template_shallow"], messages["template"])
+
+    def test_a_missing_template_fails_with_the_builtin_message(self) -> None:
+        messages = {}
+        for plugin in ("template", "template_shallow"):
+            with self.assertRaises(AnsibleError) as raised:
+                self._render(plugin, self._variables(), "absent.j2")
+            messages[plugin] = [
+                cause.replace("template_shallow", "template")
+                for cause in _causes(raised.exception)
+            ]
+        self.assertIn("absent.j2", " ".join(messages["template"]))
+        self.assertEqual(messages["template_shallow"], messages["template"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/sys-front-inj-all/test_injector_code_shallow.py

@@ -0,0 +1,229 @@
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+from ansible.parsing.dataloader import DataLoader
+from ansible.plugins.loader import filter_loader, lookup_loader
+from ansible.plugins.lookup import LookupBase
+from ansible.template import Templar, trust_as_template
+
+from utils.cache.files import PROJECT_ROOT
+from utils.cache.yaml import load_yaml_any
+from utils.networks.reachability import CLEARNET, TOR, network_of
+from utils.roles.mapping import ROLE_FILE_TASKS_MAIN
+
+filter_loader.add_directory(str(PROJECT_ROOT / "plugins" / "filter"))
+lookup_loader.add_directory(str(PROJECT_ROOT / "plugins" / "lookup"))
+
+SHALLOW = "lookup('template_shallow', "
+BUILTIN = "lookup('template', "
+DOMAIN = "infinito.example"
+ONION = "a" * 56 + ".onion"
+APP_DOMAIN = {CLEARNET: f"next.cloud.{DOMAIN}", TOR: f"next.cloud.{ONION}"}
+BASE_URL = {
+    "web-app-matomo": {
+        CLEARNET: f"https://matomo.{DOMAIN}/",
+        TOR: f"http://matomo.{ONION}/",
+    },
+    "web-svc-logout": {
+        CLEARNET: f"https://logout.{DOMAIN}/",
+        TOR: f"http://logout.{ONION}/",
+    },
+    "web-app-dashboard": {
+        CLEARNET: f"https://dash.{DOMAIN}/",
+        TOR: f"http://dash.{ONION}/",
+    },
+}
+INJECTORS = {
+    "logout": ("sys-front-inj-logout", ROLE_FILE_TASKS_MAIN, "logout_code"),
+    "dashboard": ("sys-front-inj-dashboard", ROLE_FILE_TASKS_MAIN, "iframe_init_code"),
+    "matomo": ("sys-front-inj-matomo", "tasks/inject.yml", "matomo_tracking_code"),
+}
+OIDC = {
+    "CLIENT": {
+        "LOGOUT_URL": "https://auth.example/realms/x/protocol/openid-connect/logout",
+        "ID": "infinito",
+    }
+}
+
+
+class _Stub(LookupBase):
+    def __init__(self, answer):
+        super().__init__()
+        self._answer = answer
+
+    def run(self, terms, variables=None, **kwargs):
+        return [self._answer(variables, *terms)]
+
+
+def _tls(variables, application_id, key):
+    if key != "url.base":
+        raise AssertionError(f"unexpected tls key: {key!r}")
+    return BASE_URL[application_id][network_of(variables["domain"])]
+
+
+def _config(variables, application_id, key, default):
+    if (application_id, key) != ("svc-net-tor", "services.tor.node"):
+        raise AssertionError(f"unexpected config key: {application_id!r} {key!r}")
+    return variables["_tor_node"]
+
+
+STUBS = {
+    "tls": _tls,
+    "domain": lambda variables, application_id: variables["_canonical"],
+    "config": _config,
+}
+
+
+def load_expression(injector: str) -> str:
+    """Return the set_fact expression of an injector's code-loading task.
+
+    Args:
+        injector: a key of ``INJECTORS``.
+
+    Returns:
+        The templated value that renders the injector's code.
+    """
+    role, tasks_file, fact = INJECTORS[injector]
+    tasks = load_yaml_any(str(PROJECT_ROOT / "roles" / role / tasks_file))
+    [task] = [t for t in tasks if fact in t.get("set_fact", {})]
+    return task["set_fact"][fact]
+
+
+def render(injector: str, expression: str, variables: dict):
+    """Render an injector expression with stubbed inventory lookups.
+
+    Args:
+        injector: a key of ``INJECTORS``.
+        expression: the set_fact value to template.
+        variables: the task variables, without the search path.
+
+    Returns:
+        ``("code", text)`` on success, else ``("error", causes)`` with the
+        unaltered message of every exception in the cause chain.
+    """
+    role = INJECTORS[injector][0]
+    real_get = lookup_loader.get
+
+    def get(name, *args, **kwargs):
+        if name in STUBS:
+            return _Stub(STUBS[name])
+        return real_get(name, *args, **kwargs)
+
+    scope = {
+        **variables,
+        "ansible_search_path": [str(PROJECT_ROOT / "roles" / role)],
+    }
+    try:
+        with mock.patch.object(lookup_loader, "get", side_effect=get):
+            code = Templar(loader=DataLoader(), variables=scope).template(
+                trust_as_template(expression)
+            )
+    except Exception as error:
+        causes = []
+        while error is not None:
+            causes.append(f"{type(error).__name__}: {error}")
+            error = error.__cause__
+        return ("error", causes)
+    return ("code", code)
+
+
+def _variants(injector: str) -> list[dict]:
+    variants = []
+    for network in (CLEARNET, TOR):
+        for debug in (True, False):
+            base = {
+                "domain": APP_DOMAIN[network],
+                "application_id": "web-app-nextcloud",
+                "MODE_DEBUG": debug,
+                "DOMAIN_PRIMARY": DOMAIN,
+            }
+            if injector == "logout":
+                for canonical in APP_DOMAIN.values():
+                    variants.append({**base, "OIDC": OIDC, "_canonical": canonical})
+                    variants.append(
+                        {
+                            **base,
+                            "OIDC": OIDC,
+                            "_canonical": canonical,
+                            "OIDC_OVERRIDE": {"URL": "https://idp.example"},
+                        }
+                    )
+            elif injector == "dashboard":
+                variants.extend({**base, "_tor_node": node} for node in ("", ONION))
+            else:
+                variants.extend(
+                    {
+                        **base,
+                        "matomo_site_domain": APP_DOMAIN[network],
+                        "matomo_tracking_scope": scope,
+                        "matomo_site_id": 7,
+                    }
+                    for scope in ("root", "domain")
+                )
+    return variants
+
+
+BROKEN = {
+    "logout": {
+        "domain": APP_DOMAIN[TOR],
+        "application_id": "x",
+        "MODE_DEBUG": False,
+        "_canonical": DOMAIN,
+    },
+    "dashboard": {"domain": APP_DOMAIN[TOR], "_tor_node": ""},
+    "matomo": {
+        "domain": APP_DOMAIN[CLEARNET],
+        "matomo_tracking_scope": "root",
+        "MODE_DEBUG": False,
+    },
+}
+
+
+class TestInjectorCodeRendersShallow(unittest.TestCase):
+    def test_every_code_loading_task_uses_the_shallow_lookup(self) -> None:
+        for injector in INJECTORS:
+            with self.subTest(injector=injector):
+                expression = load_expression(injector)
+                self.assertIn(SHALLOW, expression)
+                self.assertNotIn(BUILTIN, expression)
+
+    def test_every_variant_renders_byte_identical_code(self) -> None:
+        for injector in INJECTORS:
+            shallow = load_expression(injector)
+            builtin = shallow.replace(SHALLOW, BUILTIN)
+            for variables in _variants(injector):
+                with self.subTest(injector=injector, variables=variables):
+                    outcome = render(injector, shallow, variables)
+                    self.assertEqual(outcome[0], "code", outcome)
+                    self.assertEqual(outcome, render(injector, builtin, variables))
+
+    def test_the_variants_reach_every_branch(self) -> None:
+        shallow = load_expression("logout")
+        codes = {render("logout", shallow, v)[1] for v in _variants("logout")}
+        self.assertTrue(any("post_logout_redirect_uri" in code for code in codes))
+        self.assertTrue(any("post_logout_redirect_uri" not in code for code in codes))
+        shallow = load_expression("matomo")
+        codes = {render("matomo", shallow, v)[1] for v in _variants("matomo")}
+        self.assertTrue(any("enableCrossDomainLinking" in code for code in codes))
+        self.assertTrue(any("Matomo is loaded" in code for code in codes))
+        shallow = load_expression("dashboard")
+        codes = {render("dashboard", shallow, v)[1] for v in _variants("dashboard")}
+        self.assertTrue(any(f"'{ONION}'" in code for code in codes))
+
+    def test_a_missing_variable_fails_with_the_builtin_message(self) -> None:
+        for injector, variables in BROKEN.items():
+            with self.subTest(injector=injector):
+                shallow = load_expression(injector)
+                outcome = render(injector, shallow, variables)
+                self.assertEqual(outcome[0], "error", outcome)
+                self.assertIn("is undefined", " ".join(outcome[1]))
+                self.assertEqual(
+                    outcome,
+                    render(injector, shallow.replace(SHALLOW, BUILTIN), variables),
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/sys-front-inj-design/test_resolve_shallow_equivalence.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+from ansible.plugins.loader import lookup_loader
+
+from utils.cache.files import PROJECT_ROOT
+
+from .test_resolve_render_once import (
+    DOMAIN,
+    ONION,
+    resolve_tasks,
+    run_tasks,
+)
+
+SOURCES = (
+    "templates/style.css.j2",
+    "files/style.css",
+    "files/logo.css",
+    "templates/design.js.j2",
+    "templates/design.js",
+)
+SHALLOW = "lookup('template_shallow', "
+BUILTIN = "lookup('template', "
+
+
+def _roles_with_design_sources() -> list[str]:
+    return sorted(
+        role.name
+        for role in (PROJECT_ROOT / "roles").iterdir()
+        if any((role / source).is_file() for source in SOURCES)
+    )
+
+
+def _builtin(value):
+    if isinstance(value, str):
+        return value.replace(SHALLOW, BUILTIN)
+    if isinstance(value, dict):
+        return {key: _builtin(item) for key, item in value.items()}
+    return value
+
+
+def base_tasks():
+    """Return the resolve tasks with every shallow lookup swapped back.
+
+    Returns:
+        ``[style_task, script_task]`` rendering through ``ansible.builtin.template``.
+    """
+    return [_builtin(task) for task in resolve_tasks()]
+
+
+def _outcome(tasks, app, vhost, logo):
+    try:
+        facts, renders = run_tasks(tasks, app, vhost, logo)
+    except Exception as error:
+        causes = []
+        while error is not None:
+            causes.append(f"{type(error).__name__}: {error}".replace(SHALLOW, BUILTIN))
+            error = error.__cause__
+        return ("error", causes)
+    return ("facts", facts, {k: type(v) for k, v in facts.items()}, renders)
+
+
+class TestResolveTasksRenderShallow(unittest.TestCase):
+    def test_every_template_lookup_of_both_tasks_is_shallow(self) -> None:
+        for task in resolve_tasks():
+            with self.subTest(task=task["name"]):
+                text = repr(task)
+                self.assertNotIn(BUILTIN, text)
+                self.assertIn(SHALLOW, text)
+
+    def test_the_base_overlay_restores_the_builtin_lookup(self) -> None:
+        for task in base_tasks():
+            with self.subTest(task=task["name"]):
+                self.assertNotIn(SHALLOW, repr(task))
+                self.assertIn(BUILTIN, repr(task))
+
+
+class TestShallowMatchesBuiltinForEveryRole(unittest.TestCase):
+    def test_every_role_variant_sets_identical_facts_or_fails_identically(
+        self,
+    ) -> None:
+        roles = _roles_with_design_sources()
+        self.assertGreater(len(roles), 70)
+        rendered = 0
+        for app in roles:
+            for vhost in (f"x.app.{DOMAIN}", f"x.app.{ONION}"):
+                for logo in (True, False):
+                    with self.subTest(app=app, vhost=vhost, logo=logo):
+                        shallow = _outcome(resolve_tasks(), app, vhost, logo)
+                        builtin = _outcome(base_tasks(), app, vhost, logo)
+                        self.assertEqual(shallow, builtin)
+                        rendered += shallow[0] == "facts"
+        self.assertGreater(rendered, len(roles))
+
+    def test_the_shallow_lookup_never_deep_copies_the_task_variables(self) -> None:
+        real_get = lookup_loader.get
+        copies = []
+
+        def get(name, *args, **kwargs):
+            plugin = real_get(name, *args, **kwargs)
+            if name == "template":
+                original = plugin.run
+
+                def watched(terms, variables=None, **options):
+                    copies.append(type(variables).__name__)
+                    return original(terms, variables=variables, **options)
+
+                plugin.run = watched
+            return plugin
+
+        with mock.patch.object(lookup_loader, "get", side_effect=get):
+            run_tasks(resolve_tasks(), "web-svc-logout", f"x.app.{DOMAIN}", True)
+        self.assertTrue(copies)
+        self.assertEqual(set(copies), {"_SharedVariables"})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/web-opt-rdr-domains/lookup_plugins/__init__.py

@@ -1,5 +0,0 @@
-from pathlib import Path
-
-PROJECT_ROOT: Path = Path(__file__).resolve().parents[6]
-
-__all__ = ["PROJECT_ROOT"]

tests/unit/python/roles/web-opt-rdr-domains/lookup_plugins/test_template_shallow.py

@@ -1,103 +0,0 @@
-from __future__ import annotations
-
-import json
-import tempfile
-import unittest
-from pathlib import Path
-
-from ansible.errors import AnsibleError
-from ansible.parsing.dataloader import DataLoader
-from ansible.plugins.loader import lookup_loader
-from ansible.template import Templar, trust_as_template
-
-from utils.cache.yaml import load_yaml_any
-from utils.roles.mapping import ROLE_FILE_TASKS_MAIN
-
-from . import PROJECT_ROOT
-
-ROLE = PROJECT_ROOT / "roles" / "web-opt-rdr-domains"
-PROBE_PLUGIN = """
-from ansible.plugins.lookup import LookupBase
-
-
-class LookupModule(LookupBase):
-    def run(self, terms, variables, **kwargs):
-        return [variables["domain"]]
-"""
-TEMPLATE = "{{ domain }} {{ lookup('rdr_probe_variables_domain') }}\n"
-
-
-class _NoDeepCopy:
-    def __deepcopy__(self, memo):
-        raise AssertionError("the variables were deep-copied")
-
-
-class TestTemplateShallow(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls) -> None:
-        cls._tmp = tempfile.TemporaryDirectory()
-        tmp = Path(cls._tmp.name)
-        (tmp / "plugins").mkdir()
-        (tmp / "plugins" / "rdr_probe_variables_domain.py").write_text(PROBE_PLUGIN)
-        (tmp / "templates").mkdir()
-        (tmp / "templates" / "vhost.j2").write_text(TEMPLATE)
-        lookup_loader.add_directory(str(ROLE / "lookup_plugins"))
-        lookup_loader.add_directory(str(tmp / "plugins"))
-        cls.search_path = [str(tmp)]
-
-    @classmethod
-    def tearDownClass(cls) -> None:
-        cls._tmp.cleanup()
-
-    def _variables(self) -> dict:
-        return {"domain": "play.example", "ansible_search_path": self.search_path}
-
-    def _render(self, plugin: str, variables: dict) -> str:
-        return Templar(loader=DataLoader(), variables=variables).template(
-            trust_as_template(
-                "{{ lookup('" + plugin + "', 'vhost.j2', "
-                "template_vars={'domain': 'source.example'}) }}"
-            )
-        )
-
-    def test_a_lookup_inside_the_template_reads_the_per_call_domain(self) -> None:
-        rendered = self._render("template_shallow", self._variables())
-        self.assertEqual(rendered, "source.example source.example\n")
-        self.assertEqual(rendered, self._render("template", self._variables()))
-
-    def test_the_caller_variables_keep_their_values(self) -> None:
-        variables = self._variables()
-        self._render("template_shallow", variables)
-        self.assertEqual(variables, self._variables())
-
-    def test_no_value_of_the_caller_is_deep_copied(self) -> None:
-        variables = {**self._variables(), "untouchable": _NoDeepCopy()}
-        self.assertEqual(
-            self._render("template_shallow", variables),
-            "source.example source.example\n",
-        )
-        with self.assertRaises(AnsibleError) as raised:
-            self._render("template", variables)
-        causes = []
-        error: BaseException | None = raised.exception
-        while error is not None:
-            causes.append(str(error))
-            error = error.__cause__
-        self.assertTrue(any("deep-copied" in cause for cause in causes), causes)
-
-
-class TestRedirectLoopRendersShallow(unittest.TestCase):
-    def test_both_files_of_every_source_render_through_template_shallow(
-        self,
-    ) -> None:
-        tasks = load_yaml_any(str(ROLE / ROLE_FILE_TASKS_MAIN))
-        [loop_task] = [
-            task for task in tasks if "redirect_source" in task.get("vars", {})
-        ]
-        files = loop_task["vars"]["redirect_source"]["files"]
-        self.assertEqual(files.count("lookup('template_shallow', "), 2)
-        self.assertNotIn("lookup('template', ", json.dumps(loop_task))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/python/roles/web-opt-rdr-domains/test_redirect_loop_shallow.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import json
+import unittest
+
+from utils.cache.files import PROJECT_ROOT
+from utils.cache.yaml import load_yaml_any
+from utils.roles.mapping import ROLE_FILE_TASKS_MAIN
+
+ROLE = PROJECT_ROOT / "roles" / "web-opt-rdr-domains"
+
+
+class TestRedirectLoopRendersShallow(unittest.TestCase):
+    def test_both_files_of_every_source_render_through_template_shallow(
+        self,
+    ) -> None:
+        tasks = load_yaml_any(str(ROLE / ROLE_FILE_TASKS_MAIN))
+        [loop_task] = [
+            task for task in tasks if "redirect_source" in task.get("vars", {})
+        ]
+        files = loop_task["vars"]["redirect_source"]["files"]
+        self.assertEqual(files.count("lookup('template_shallow', "), 2)
+        self.assertNotIn("lookup('template', ", json.dumps(loop_task))
+
+    def test_the_lookup_is_the_shared_plugin_not_a_role_local_copy(self) -> None:
+        self.assertTrue(
+            (PROJECT_ROOT / "plugins" / "lookup" / "template_shallow.py").is_file()
+        )
+        self.assertFalse((ROLE / "lookup_plugins").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()