refactor(web-opt-rdr-www): rename canonical to w3redirect + lookup-based redirect mappings

- web-opt-rdr-www canonical: rdr-www.{DOMAIN_PRIMARY} -> w3redirect.{DOMAIN_PRIMARY}.
- New lookup `www_redirect_domains` (role-scoped): returns full {source, target}
  mapping list, resolves DOMAIN_HOMEPAGE target for the role's own canonical via
  lookup('domain', 'web-opt-rdr-www'). REDIRECT_WWW_DOMAINS now carries dicts;
  tasks/main.yml and _01_cloudflare_edge_redirect.yml updated to consume dicts.
- New lookup `current_play_redirect_domains`: replaces the two inline
  set_fact merges in tasks/stages/01_constructor.yml. Merges user-supplied
  redirect_domain_mappings, primary-redirect (conditional on web-opt-rdr-domains
  in deployed), and per-app domain mappings (inlined from the now-removed
  domain_mappings filter).
- domain_redirect_mappings filter removed; web-app-keycloak RedirectUrisTest
  pollution of utils.roles.applications.config in sys.modules fixed via
  tearDownClass cleanup.
- csp_skip_domains lookup: csp.health.skip opt-out path removed (architecture
  no longer needs it).
- Lint test_only_web_app_roles_integrate_dashboard expanded from web-app-* to
  web-*; test_role_and_group_vars_used now scans .py files too so vars consumed
  only inside plugins (DOMAIN_HOMEPAGE, AUTO_BUILD_ALIASES) are seen.
- New role-scoped variants.yml on web-opt-rdr-www with dashboard on/off.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

plugins/filter/domain_redirect_mappings.py

@@ -1,123 +0,0 @@
-from ansible.errors import AnsibleFilterError
-
-from utils.domains.list import render_domain_value
-from utils.roles.entity_name import get_entity_name
-
-
-class FilterModule:
-    def filters(self):
-        return {"domain_mappings": self.domain_mappings}
-
-    def domain_mappings(self, apps, primary_domain, auto_build_alias):
-        """
-        Build a flat list of redirect mappings for all apps:
-          - source: each alias domain
-          - target: the first canonical domain
-        Skip mappings where source == target, since they make no sense.
-
-        IMPORTANT:
-        - If auto_build_alias is False, do NOT automatically add the default alias
-          (<entity>.<primary_domain>) unless it is explicitly configured in aliases.
-        """
-
-        def parse_entry(domains_cfg, key, app_id):
-            if key not in domains_cfg:
-                return None
-            entry = render_domain_value(
-                domains_cfg[key],
-                {"DOMAIN_PRIMARY": primary_domain},
-                f"{app_id}.server.domains.{key}",
-            )
-            if isinstance(entry, dict):
-                values = list(entry.values())
-            elif isinstance(entry, list):
-                values = entry
-            else:
-                raise AnsibleFilterError(
-                    f"Unexpected type for 'domains.{key}' in application '{app_id}': {type(entry).__name__}"
-                )
-            for d in values:
-                if not isinstance(d, str) or not d.strip():
-                    raise AnsibleFilterError(
-                        f"Invalid domain entry in '{key}' for application '{app_id}': {d!r}"
-                    )
-            return values
-
-        def default_domain(app_id: str, primary: str):
-            subdomain = get_entity_name(app_id)
-            return f"{subdomain}.{primary}"
-
-        # 1) Compute canonical domains per app (always as a list)
-        canonical_map = {}
-        for app_id, cfg in apps.items():
-            domains_cfg = cfg.get("server", {}).get("domains", {})
-            entry = (
-                render_domain_value(
-                    domains_cfg.get("canonical"),
-                    {"DOMAIN_PRIMARY": primary_domain},
-                    f"{app_id}.server.domains.canonical",
-                )
-                if "canonical" in domains_cfg
-                else None
-            )
-            if entry is None:
-                canonical_map[app_id] = [default_domain(app_id, primary_domain)]
-            elif isinstance(entry, dict):
-                canonical_map[app_id] = list(entry.values())
-            elif isinstance(entry, list):
-                canonical_map[app_id] = list(entry)
-            else:
-                raise AnsibleFilterError(
-                    f"Unexpected type for 'server.domains.canonical' in application '{app_id}': {type(entry).__name__}"
-                )
-
-        # 2) Compute alias domains per app
-        alias_map = {}
-        for app_id, cfg in apps.items():
-            domains_cfg = cfg.get("server", {}).get("domains", {})
-            if domains_cfg is None:
-                alias_map[app_id] = []
-                continue
-
-            default = default_domain(app_id, primary_domain)
-
-            # If domains_cfg is explicitly an empty dict, that used to mean:
-            # "use the default domain". But that is an auto-build behavior and
-            # must be gated by auto_build_alias.
-            if isinstance(domains_cfg, dict) and not domains_cfg:
-                alias_map[app_id] = [default] if auto_build_alias else []
-                continue
-
-            aliases = parse_entry(domains_cfg, "aliases", app_id) or []
-            has_aliases = "aliases" in domains_cfg
-            has_canonical = "canonical" in domains_cfg
-
-            if has_aliases:
-                # ONLY auto-add default alias when enabled.
-                # If the user set aliases: [], they likely want "no aliases".
-                if auto_build_alias and default not in aliases:
-                    aliases.append(default)
-            elif has_canonical:
-                # If canonical exists but aliases key does not, we may auto-add
-                # the default alias when enabled (and only if it isn't already canonical).
-                canon = canonical_map.get(app_id, [])
-                if default not in canon and default not in aliases and auto_build_alias:
-                    aliases.append(default)
-
-            alias_map[app_id] = aliases
-
-        # 3) Build flat list of {source, target} entries,
-        #    skipping self-mappings
-        mappings = []
-        for app_id, sources in alias_map.items():
-            canon_list = canonical_map.get(app_id, [])
-            target = (
-                canon_list[0] if canon_list else default_domain(app_id, primary_domain)
-            )
-            for src in sources:
-                if src == target:
-                    # skip self-redirects
-                    continue
-                mappings.append({"source": src, "target": target})
-
-        return mappings

plugins/lookup/current_play_redirect_domains.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from plugins.filter.merge_mapping import merge_mapping
+from utils.domains.list import render_domain_value
+from utils.roles.entity_name import get_entity_name
+
+
+def _per_app_redirect_mappings(
+    apps: Mapping[str, Any],
+    primary_domain: str,
+    auto_build_alias: bool,
+) -> list[dict[str, str]]:
+    """Build a flat list of {source, target} 301 mappings for all apps.
+
+    For each app, the canonical (first entry of ``server.domains.canonical``,
+    falling back to ``<entity>.<primary_domain>``) is the redirect target.
+    Each alias from ``server.domains.aliases`` becomes a source. When
+    ``auto_build_alias`` is true, ``<entity>.<primary_domain>`` is also
+    added as an alias unless already canonical. Self-mappings are skipped.
+    """
+
+    def _parse_entry(domains_cfg, key, app_id):
+        if key not in domains_cfg:
+            return None
+        entry = render_domain_value(
+            domains_cfg[key],
+            {"DOMAIN_PRIMARY": primary_domain},
+            f"{app_id}.server.domains.{key}",
+        )
+        if isinstance(entry, dict):
+            values = list(entry.values())
+        elif isinstance(entry, list):
+            values = entry
+        else:
+            raise AnsibleError(
+                f"Unexpected type for 'domains.{key}' in application '{app_id}': "
+                f"{type(entry).__name__}"
+            )
+        for d in values:
+            if not isinstance(d, str) or not d.strip():
+                raise AnsibleError(
+                    f"Invalid domain entry in '{key}' for application '{app_id}': {d!r}"
+                )
+        return values
+
+    def _default_domain(app_id: str) -> str:
+        return f"{get_entity_name(app_id)}.{primary_domain}"
+
+    canonical_map: dict[str, list[str]] = {}
+    for app_id, cfg in apps.items():
+        domains_cfg = (cfg or {}).get("server", {}).get("domains", {}) or {}
+        if "canonical" in domains_cfg:
+            entry = render_domain_value(
+                domains_cfg["canonical"],
+                {"DOMAIN_PRIMARY": primary_domain},
+                f"{app_id}.server.domains.canonical",
+            )
+            if isinstance(entry, dict):
+                canonical_map[app_id] = list(entry.values())
+            elif isinstance(entry, list):
+                canonical_map[app_id] = list(entry)
+            else:
+                raise AnsibleError(
+                    f"Unexpected type for 'server.domains.canonical' in "
+                    f"application '{app_id}': {type(entry).__name__}"
+                )
+        else:
+            canonical_map[app_id] = [_default_domain(app_id)]
+
+    alias_map: dict[str, list[str]] = {}
+    for app_id, cfg in apps.items():
+        domains_cfg = (cfg or {}).get("server", {}).get("domains", {})
+        if domains_cfg is None:
+            alias_map[app_id] = []
+            continue
+
+        default = _default_domain(app_id)
+
+        if isinstance(domains_cfg, dict) and not domains_cfg:
+            alias_map[app_id] = [default] if auto_build_alias else []
+            continue
+
+        aliases = _parse_entry(domains_cfg, "aliases", app_id) or []
+        has_aliases = "aliases" in domains_cfg
+        has_canonical = "canonical" in domains_cfg
+
+        if has_aliases:
+            if auto_build_alias and default not in aliases:
+                aliases.append(default)
+        elif has_canonical:
+            canon = canonical_map.get(app_id, [])
+            if default not in canon and default not in aliases and auto_build_alias:
+                aliases.append(default)
+
+        alias_map[app_id] = aliases
+
+    mappings: list[dict[str, str]] = []
+    for app_id, sources in alias_map.items():
+        canon_list = canonical_map.get(app_id, [])
+        target = canon_list[0] if canon_list else _default_domain(app_id)
+        for src in sources:
+            if src == target:
+                continue
+            mappings.append({"source": src, "target": target})
+
+    return mappings
+
+
+class LookupModule(LookupBase):
+    """Return CURRENT_PLAY_REDIRECT_DOMAINS as a list of {source,target} dicts.
+
+    Merges, in order:
+      1. user-supplied ``redirect_domain_mappings`` (if any)
+      2. the hardcoded ``{DOMAIN_PRIMARY -> DOMAIN_HOMEPAGE}`` primary-redirect,
+         appended only when ``web-opt-rdr-domains`` is in
+         ``lookup('deployment').deployed`` (so isolated CI variants do not
+         schedule a redirect whose target is not served)
+      3. per-app alias-to-canonical mappings derived from
+         ``lookup('applications_current_play')``, gated by
+         ``AUTO_BUILD_ALIASES``
+
+    Steps 1-2 win over conflicting entries from step 3 (same merge order as
+    the previous inline two-stage set_fact in ``01_constructor.yml``).
+    """
+
+    def _run_lookup(self, name: str, variables: dict[str, Any] | None) -> Any:
+        plugin = lookup_loader.get(
+            name,
+            loader=getattr(self, "_loader", None),
+            templar=getattr(self, "_templar", None),
+        )
+        if plugin is None:
+            return None
+        result = plugin.run([], variables=variables)
+        return result[0] if result else None
+
+    def run(
+        self,
+        terms: list[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[list[dict[str, str]]]:
+        vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
+
+        domain_primary = str(vars_.get("DOMAIN_PRIMARY") or "")
+        domain_homepage = str(vars_.get("DOMAIN_HOMEPAGE") or "")
+        auto_build_aliases = bool(vars_.get("AUTO_BUILD_ALIASES", False))
+
+        user_mappings = vars_.get("redirect_domain_mappings") or []
+        if not isinstance(user_mappings, list):
+            user_mappings = []
+
+        merged = merge_mapping([], list(user_mappings), "source")
+
+        deployment = self._run_lookup("deployment", vars_)
+        deployed = (
+            list(deployment.get("deployed") or [])
+            if isinstance(deployment, Mapping)
+            else []
+        )
+        if "web-opt-rdr-domains" in deployed and domain_primary and domain_homepage:
+            merged = merge_mapping(
+                merged,
+                [{"source": domain_primary, "target": domain_homepage}],
+                "source",
+            )
+
+        current_play_apps = self._run_lookup("applications_current_play", vars_)
+        if isinstance(current_play_apps, Mapping):
+            per_app = _per_app_redirect_mappings(
+                current_play_apps, domain_primary, auto_build_aliases
+            )
+            merged = merge_mapping(per_app, merged, "source")
+
+        return [merged]

roles/sys-ctl-hlth-csp/lookup_plugins/csp_skip_domains.py

@@ -58,17 +58,9 @@ def _selection_from(group_names: Any) -> set[str]:
 class LookupModule(LookupBase):
     """Return domains the CSP probe should skip.
 
-    Resolves the applications dict via `get_merged_applications` itself, so
-    callers do not need to pipe `lookup('applications')` through a filter.
-
-    Usage:
-        lookup('csp_skip_domains', group_names=lookup('deployment').deployed)
-
-    `group_names` is optional. If omitted, all applications are considered.
-
-    Per-application opt-outs honored:
-      * server.status_codes.default with any HTTP code >= 400
-      * server.csp.health.skip: true (also covers `www.<domain>` variants)
+    Skips canonical + alias domains of every selected application whose
+    ``server.status_codes.default`` declares any HTTP code >= 400
+    (e.g. federation-only roles that legitimately serve 4xx at ``/``).
     """
 
     def run(
@@ -102,17 +94,7 @@ class LookupModule(LookupBase):
                 strict=False,
                 default=None,
             )
-            opt_out = bool(
-                get(
-                    applications,
-                    app_id,
-                    "server.csp.health.skip",
-                    strict=False,
-                    default=False,
-                )
-            )
-
-            if not opt_out and not _has_code_ge_400(default):
+            if not _has_code_ge_400(default):
                 continue
 
             canonical = get(
@@ -130,10 +112,7 @@ class LookupModule(LookupBase):
                 default=[],
             )
             for d in _to_list(canonical) + _to_list(aliases):
-                if not d:
-                    continue
-                skip.add(d)
-                if opt_out:
-                    skip.add(f"www.{d}")
+                if d:
+                    skip.add(d)
 
         return [sorted(skip)]

roles/web-opt-rdr-www/lookup_plugins/www_redirect_domains.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from utils.cache.applications import get_merged_applications
+from utils.roles.applications.config import get
+
+
+def _to_list(x: Any) -> list[str]:
+    if x is None:
+        return []
+    if isinstance(x, str):
+        return [x]
+    if isinstance(x, (list, tuple, set)):
+        out: list[str] = []
+        for v in x:
+            if isinstance(v, (list, tuple, set)):
+                out.extend(_to_list(v))
+            elif isinstance(v, str):
+                out.append(v)
+            elif isinstance(v, Mapping):
+                out.extend(_to_list(list(v.values())))
+        return out
+    if isinstance(x, Mapping):
+        out = []
+        for v in x.values():
+            out.extend(_to_list(v))
+        return out
+    return []
+
+
+def _selection_from(group_names: Any) -> set[str]:
+    if isinstance(group_names, (list, set, tuple)):
+        return {str(x) for x in group_names if str(x)}
+    if isinstance(group_names, str):
+        return {g.strip() for g in group_names.split(",") if g.strip()}
+    return set()
+
+
+class LookupModule(LookupBase):
+    """Return the full redirect-mapping list for the www-redirect helper.
+
+    For every selected application, iterate ``server.domains.canonical`` and
+    ``server.domains.aliases`` and emit a ``{source, target}`` 301-mapping:
+      * if the entry already begins with ``www.``: source = entry,
+        target = entry without the ``www.`` prefix,
+      * otherwise: source = ``www.<entry>``, target = entry.
+
+    In addition, when ``DOMAIN_HOMEPAGE`` is set in the playbook variables,
+    the role's own canonical (resolved via ``lookup('domain', 'web-opt-rdr-www')``)
+    is appended as a 301-mapping with ``DOMAIN_HOMEPAGE`` as target.
+
+    The deployed-app selection is resolved internally via the ``deployment``
+    lookup. Tests can override it by passing ``group_names=[...]`` explicitly.
+    """
+
+    def _resolve_deployed(self, variables: dict[str, Any] | None) -> list[str] | None:
+        dep_plugin = lookup_loader.get(
+            "deployment",
+            loader=getattr(self, "_loader", None),
+            templar=getattr(self, "_templar", None),
+        )
+        if dep_plugin is None:
+            return None
+        result = dep_plugin.run([], variables=variables)
+        if result and isinstance(result[0], Mapping):
+            return list(result[0].get("deployed") or [])
+        return None
+
+    def run(
+        self,
+        terms: list[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[list[dict[str, str]]]:
+        vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
+
+        applications = get_merged_applications(
+            variables=vars_,
+            roles_dir=kwargs.get("roles_dir"),
+            templar=getattr(self, "_templar", None),
+        )
+
+        if not isinstance(applications, Mapping):
+            return [[]]
+
+        if "group_names" in kwargs:
+            selection = _selection_from(kwargs["group_names"])
+        else:
+            selection = _selection_from(self._resolve_deployed(vars_))
+
+        seen: set[tuple[str, str]] = set()
+        mappings: list[dict[str, str]] = []
+
+        for app_id in sorted(applications):
+            if selection and app_id not in selection:
+                continue
+            canonical = get(
+                applications,
+                app_id,
+                "server.domains.canonical",
+                strict=False,
+                default=[],
+            )
+            aliases = get(
+                applications,
+                app_id,
+                "server.domains.aliases",
+                strict=False,
+                default=[],
+            )
+            for d in _to_list(canonical) + _to_list(aliases):
+                if not d:
+                    continue
+                if d.startswith("www."):
+                    source, target = d, d[4:]
+                else:
+                    source, target = f"www.{d}", d
+                key = (source, target)
+                if key in seen:
+                    continue
+                seen.add(key)
+                mappings.append({"source": source, "target": target})
+
+        domain_homepage = vars_.get("DOMAIN_HOMEPAGE")
+        role_id = "web-opt-rdr-www"
+        if (
+            isinstance(domain_homepage, str)
+            and domain_homepage
+            and role_id in applications
+        ):
+            domain_plugin = lookup_loader.get(
+                "domain",
+                loader=getattr(self, "_loader", None),
+                templar=getattr(self, "_templar", None),
+            )
+            if domain_plugin is not None:
+                role_canonical = domain_plugin.run([role_id], variables=vars_)[0]
+                if isinstance(role_canonical, str) and role_canonical:
+                    key = (role_canonical, domain_homepage)
+                    if key not in seen:
+                        seen.add(key)
+                        mappings.append({"source": key[0], "target": key[1]})
+
+        mappings.sort(key=lambda m: (m["source"], m["target"]))
+        return [mappings]

roles/web-opt-rdr-www/meta/server.yml

@@ -1,4 +1,4 @@
 ---
-csp:
-  health:
-    skip: true
+domains:
+  canonical:
+    - "w3redirect.{{ DOMAIN_PRIMARY }}"

roles/web-opt-rdr-www/meta/variants.yml

@@ -0,0 +1,9 @@
+---
+- services:
+    dashboard: # nocheck: variants-services-match
+      enabled: true
+      shared: true
+- services:
+    dashboard: # nocheck: variants-services-match
+      enabled: false
+      shared: false

roles/web-opt-rdr-www/tasks/_01_cloudflare_edge_redirect.yml

@@ -2,7 +2,7 @@
 - name: Include Cloudflare redirect rule to enforce www → apex
   include_tasks: _02_cloudflare_redirect_rule.yml
   vars:
-    domain: "{{ item | regex_replace('^www\\.', '') }}"
-    www_fqdn: "{{ item }}"
-    apex_url: "{{ lookup('tls', application_id, 'protocols.web') }}://{{ item | regex_replace('^www\\.', '') }}"
+    domain: "{{ item.target }}"
+    www_fqdn: "{{ item.source }}"
+    apex_url: "{{ lookup('tls', application_id, 'protocols.web') }}://{{ item.target }}"
   loop: "{{ REDIRECT_WWW_DOMAINS }}"

roles/web-opt-rdr-www/tasks/_02_cloudflare_redirect_rule.yml

@@ -7,10 +7,10 @@
 #   apex_url:   "https://example.com"
 #   CLOUDFLARE_API_TOKEN
 
-- name: "Derive zone and apex from www_fqdn"
+- name: "Derive zone from www_fqdn; apex comes from caller"
   set_fact:
-    cf_zone_name: "{{ www_fqdn | to_zone }}" # e.g. "cymais.cloud"
-    domain_apex: "{{ www_fqdn | regex_replace('^www\\.', '') }}" # e.g. "academy.cymais.cloud"
+    cf_zone_name: "{{ www_fqdn | to_zone }}"
+    domain_apex: "{{ domain }}"
 
 - name: "Cloudflare: Lookup zone id for {{ cf_zone_name }}"
   uri_retry:

roles/web-opt-rdr-www/tasks/main.yml

@@ -3,8 +3,7 @@
   include_tasks: "utils/load_app.yml"
   vars:
     load_app_id: web-opt-rdr-domains
-    redirect_domain_mappings: "{{ REDIRECT_WWW_DOMAINS | map('regex_replace', '^www\\.(.+)$', '{ source: \"www.\\1\", target: \"\\1\" }') | map('from_yaml') | list
-      }}"
+    redirect_domain_mappings: "{{ REDIRECT_WWW_DOMAINS }}"
   when: REDIRECT_WWW_FLAVOR == 'origin'
 
 - name: Include Cloudflare edge redirect

roles/web-opt-rdr-www/vars/main.yml

@@ -4,4 +4,4 @@ application_id: "web-opt-rdr-www"
 
 # Redirect WWW
 REDIRECT_WWW_FLAVOR: "{{ lookup('config', application_id, 'services.rdr-www.prefered_flavor') if DNS_PROVIDER == 'cloudflare' else 'origin' }}"
-REDIRECT_WWW_DOMAINS: "{{ CURRENT_PLAY_DOMAINS_ALL | select('match', '^www\\.') | list }}"
+REDIRECT_WWW_DOMAINS: "{{ lookup('www_redirect_domains') }}"

tasks/stages/01_constructor.yml

@@ -51,32 +51,9 @@
               )
           }}
 
-    - name: Merge CURRENT_PLAY_REDIRECT_DOMAINS
+    - name: Set CURRENT_PLAY_REDIRECT_DOMAINS
       ansible.builtin.set_fact:
-      # The following mapping is necessary to define the exceptions for domains which are created, but which aren't used
-        CURRENT_PLAY_REDIRECT_DOMAINS: >-
-          {{
-            []
-            | merge_mapping(redirect_domain_mappings | default([]), 'source')
-            | merge_mapping(
-                [
-                  {
-                    'source': ([ DOMAIN_PRIMARY  ] | flatten | map('string') | first),
-                    'target': ([ DOMAIN_HOMEPAGE ] | flatten | map('string') | first)
-                  }
-                ],
-                'source'
-              )
-          }}
-
-    - name: Set current play redirect domain mappings
-      ansible.builtin.set_fact:
-        CURRENT_PLAY_REDIRECT_DOMAINS: >-
-          {{
-            CURRENT_PLAY_APPLICATIONS
-            | domain_mappings(DOMAIN_PRIMARY, AUTO_BUILD_ALIASES)
-            | merge_mapping(CURRENT_PLAY_REDIRECT_DOMAINS, 'source')
-          }}
+        CURRENT_PLAY_REDIRECT_DOMAINS: "{{ lookup('current_play_redirect_domains') }}"
 
     - name: Set current play all domains incl. www redirect if enabled
       ansible.builtin.set_fact:

tests/integration/roles/dashboard/test_integration_scope.py

@@ -1,4 +1,4 @@
-"""Guard: only ``web-app-*`` roles may integrate the dashboard.
+"""Guard: only ``web-*`` roles may integrate the dashboard.
 
 A role "integrates" the dashboard when its ``meta/services.yml`` (or
 any entry in ``meta/variants.yml``) declares a ``dashboard:`` block
@@ -7,19 +7,18 @@ canonical helper :func:`utils.roles.applications.services.registry.is_explicit_t
 
 Truthy means either the literal Python ``True`` *or* the dynamic Jinja
 form ``"{{ '<role>' in group_names }}"``. Both produce a runtime tile
-in the dashboard grid, so both are forbidden for non-``web-app-*``
-roles.
+in the dashboard grid, so both are forbidden for non-``web-*`` roles.
 
 Why
 ---
 
-The dashboard is the user-facing tile grid. Only end-user-visible
-``web-app-*`` surfaces belong in it. A shared service (``web-svc-*``),
-system role (``sys-*``), desktop role (``desk-*``), or driver role
-(``drv-*``) embedding the dashboard would spam the tile grid with
-non-clickable infrastructure entries.
+The dashboard is the user-facing tile grid. Only web-facing surfaces
+belong in it. A system role (``sys-*``), desktop role (``desk-*``),
+service role (``svc-*``), or driver role (``drv-*``) embedding the
+dashboard would spam the tile grid with non-clickable infrastructure
+entries.
 
-A non-``web-app-*`` role MAY still ship a ``dashboard:`` block with
+A non-``web-*`` role MAY still ship a ``dashboard:`` block with
 *explicitly disabled* flags (``enabled: false`` / ``shared: false``)
 - that's a static "we know this service exists in the registry but
 this role does not contribute a tile" declaration and is fine.
@@ -88,10 +87,10 @@ def _find_variants_violations(role_dir: Path) -> list[str]:
 
 
 class TestDashboardIntegrationScope(unittest.TestCase):
-    def test_only_web_app_roles_integrate_dashboard(self) -> None:
+    def test_only_web_roles_integrate_dashboard(self) -> None:
         offenders: list[str] = []
         for role_dir in sorted(p for p in ROLES_DIR.iterdir() if p.is_dir()):
-            if role_dir.name.startswith("web-app-"):
+            if role_dir.name.startswith("web-"):
                 continue
             services_violation = _find_services_violation(role_dir)
             if services_violation is not None:
@@ -100,10 +99,10 @@ class TestDashboardIntegrationScope(unittest.TestCase):
 
         if offenders:
             self.fail(
-                "Non-web-app roles MUST NOT integrate the dashboard via a "
+                "Non-web roles MUST NOT integrate the dashboard via a "
                 "truthy `enabled`/`shared` flag (literal `true` or the "
                 "`'<role>' in group_names` Jinja form). The dashboard tile "
-                "grid is reserved for end-user-visible web-app surfaces. "
+                "grid is reserved for web-facing surfaces. "
                 "Either drop the `dashboard:` block, or set both flags to "
                 "`false` if the registry declaration is intentional:\n"
                 + "\n".join(f"  - {o}" for o in offenders)

tests/lint/ansible/variables/test_role_and_group_vars_used.py

@@ -203,8 +203,13 @@ def _scan_ansible_expr_idents(node, sink: set[str]) -> None:
             _scan_ansible_expr_idents(item, sink)
 
 
-def _build_usage_indices(repo_root: Path) -> tuple[set[str], set[str]]:
-    """Single project-tree walk that builds both identifier indices.
+def _scan_py_idents(text: str, sink: set[str]) -> None:
+    for m in _IDENT_RE.finditer(text):
+        sink.add(m.group(1))
+
+
+def _build_usage_indices(repo_root: Path) -> tuple[set[str], set[str], set[str]]:
+    """Single project-tree walk that builds three identifier indices.
 
     Routes every read through the project's cached helpers:
     * ``iter_project_files_with_content`` walks the path list once
@@ -214,29 +219,41 @@ def _build_usage_indices(repo_root: Path) -> tuple[set[str], set[str]]:
       ``(path, mtime, size)`` signature and reuses the parsed
       structure on every later call in the same process — no
       `yaml.safe_load*` call bypasses the cache.
+
+    Python files (plugins, filter/lookup modules, utils) are also
+    scanned so that vars consumed only via ``variables.get("FOO")``
+    inside an Ansible plugin count as referenced.
     """
     jinja_idents: set[str] = set()
     ansible_expr_idents: set[str] = set()
+    py_idents: set[str] = set()
 
     for path_str, text in iter_project_files_with_content(
-        extensions=(".yml", ".yaml", ".j2"),
+        extensions=(".yml", ".yaml", ".j2", ".py"),
         exclude_tests=True,
         exclude_dirs=("docs",),
     ):
+        if path_str.endswith(".py"):
+            _scan_py_idents(text, py_idents)
+            continue
         _scan_jinja_block_idents(text, jinja_idents)
         if path_str.endswith((".yml", ".yaml")):
             data = load_yaml_any(path_str, default_if_missing=None)
             if data is not None:
                 _scan_ansible_expr_idents(data, ansible_expr_idents)
 
-    return jinja_idents, ansible_expr_idents
+    return jinja_idents, ansible_expr_idents, py_idents
 
 
 class TestRoleAndGroupVarsUsed(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.repo_root = PROJECT_ROOT
-        cls.jinja_idents, cls.ansible_expr_idents = _build_usage_indices(cls.repo_root)
+        (
+            cls.jinja_idents,
+            cls.ansible_expr_idents,
+            cls.py_idents,
+        ) = _build_usage_indices(cls.repo_root)
 
     def test_role_and_group_vars_referenced(self):
         unused: list[str] = []
@@ -245,14 +262,18 @@ class TestRoleAndGroupVarsUsed(unittest.TestCase):
             for name, lineno in _collect_top_level_keys(def_file):
                 if name in _WHITELIST:
                     continue
-                if name in self.jinja_idents or name in self.ansible_expr_idents:
+                if (
+                    name in self.jinja_idents
+                    or name in self.ansible_expr_idents
+                    or name in self.py_idents
+                ):
                     continue
                 unused.append(f"{rel}:{lineno}: '{name}' is never referenced")
 
         if unused:
             self.fail(
                 f"{len(unused)} role/group var(s) declared but never consumed "
-                f"in any project .yml/.yaml/.j2 file:\n"
+                f"in any project .yml/.yaml/.j2/.py file:\n"
                 + "\n".join(f"- {u}" for u in unused)
             )
 

tests/unit/plugins/filter/test_domain_mappings.py

@@ -1,177 +0,0 @@
-import unittest
-
-from ansible.errors import AnsibleFilterError
-
-from plugins.filter.domain_redirect_mappings import FilterModule
-
-
-class TestDomainMappings(unittest.TestCase):
-    def setUp(self):
-        self.filter = FilterModule()
-        self.primary = "example.com"
-
-    def test_empty_apps(self):
-        apps = {}
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertEqual(result, [])
-
-    def test_app_without_domains(self):
-        apps = {"web-app-dashboard": {}}
-        # no server/domains key → no mappings
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertEqual(result, [])
-
-    def test_empty_domains_cfg(self):
-        # Explicitly empty server.domains dict.
-        # With auto_build_alias=True the default alias becomes the target itself,
-        # so it would be skipped as a self-mapping -> no mappings.
-        apps = {"web-app-dashboard": {"server": {"domains": {}}}}
-        expected = []
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertEqual(result, expected)
-
-    def test_explicit_aliases(self):
-        apps = {
-            "web-app-dashboard": {"server": {"domains": {"aliases": ["alias.com"]}}}
-        }
-        default = "dashboard.example.com"
-        expected = [
-            {"source": "alias.com", "target": default},
-        ]
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertCountEqual(result, expected)
-
-    def test_canonical_not_default(self):
-        apps = {
-            "web-app-dashboard": {"server": {"domains": {"canonical": ["foo.com"]}}}
-        }
-        expected = [{"source": "dashboard.example.com", "target": "foo.com"}]
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertEqual(result, expected)
-
-    def test_canonical_dict(self):
-        apps = {
-            "web-app-dashboard": {
-                "server": {
-                    "domains": {"canonical": {"one": "one.com", "two": "two.com"}}
-                }
-            }
-        }
-        # first canonical key 'one' -> one.com (dict insertion order)
-        expected = [{"source": "dashboard.example.com", "target": "one.com"}]
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertEqual(result, expected)
-
-    def test_rendered_template_domains(self):
-        apps = {
-            "web-app-dashboard": {
-                "server": {
-                    "domains": {
-                        "aliases": ["www.{{ DOMAIN_PRIMARY }}"],
-                        "canonical": ["dash.{{ DOMAIN_PRIMARY }}"],
-                    }
-                }
-            }
-        }
-        expected = [{"source": "www.example.com", "target": "dash.example.com"}]
-        result = self.filter.domain_mappings(apps, self.primary, False)
-        self.assertEqual(result, expected)
-
-    def test_multiple_apps(self):
-        apps = {
-            "web-app-dashboard": {"server": {"domains": {"aliases": ["a1.com"]}}},
-            "web-app-mastodon": {"server": {"domains": {"canonical": ["c2.com"]}}},
-        }
-        expected = [
-            {"source": "a1.com", "target": "dashboard.example.com"},
-            {"source": "mastodon.example.com", "target": "c2.com"},
-        ]
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertCountEqual(result, expected)
-
-    def test_multiple_aliases(self):
-        apps = {
-            "web-app-dashboard": {
-                "server": {"domains": {"aliases": ["a1.com", "a2.com"]}}
-            }
-        }
-        expected = [
-            {"source": "a1.com", "target": "dashboard.example.com"},
-            {"source": "a2.com", "target": "dashboard.example.com"},
-        ]
-        result = self.filter.domain_mappings(apps, self.primary, True)
-        self.assertCountEqual(result, expected)
-
-    def test_invalid_aliases_type(self):
-        apps = {"web-app-dashboard": {"server": {"domains": {"aliases": 123}}}}
-        with self.assertRaises(AnsibleFilterError):
-            self.filter.domain_mappings(apps, self.primary, True)
-
-    def test_canonical_not_default_no_autobuild(self):
-        """
-        When only a canonical different from the default exists and auto_build_aliases is False,
-        we should NOT auto-generate a default alias -> canonical mapping.
-        """
-        apps = {
-            "web-app-dashboard": {"server": {"domains": {"canonical": ["foo.com"]}}}
-        }
-        result = self.filter.domain_mappings(apps, self.primary, False)
-        self.assertEqual(result, [])  # no auto-added default alias
-
-    def test_aliases_and_canonical_no_autobuild(self):
-        """
-        With auto_build_aliases=False we do NOT append the default domain to aliases.
-        Only explicit aliases should map to canonical.
-        """
-        apps = {
-            "web-app-dashboard": {
-                "server": {
-                    "domains": {"aliases": ["alias.com"], "canonical": ["foo.com"]}
-                }
-            }
-        }
-        expected = [
-            {"source": "alias.com", "target": "foo.com"},
-        ]
-        result = self.filter.domain_mappings(apps, self.primary, False)
-        self.assertCountEqual(result, expected)
-
-    def test_mixed_apps_no_autobuild(self):
-        """
-        One app with only canonical (no aliases) and one app with only aliases:
-        - The canonical-only app produces no mappings when auto_build_aliases is False.
-        - The alias-only app maps its aliases to its default domain; default self-mapping is skipped.
-        """
-        apps = {
-            "web-app-dashboard": {"server": {"domains": {"canonical": ["c1.com"]}}},
-            "web-app-mastodon": {"server": {"domains": {"aliases": ["m1.com"]}}},
-        }
-        expected = [
-            {"source": "m1.com", "target": "mastodon.example.com"},
-        ]
-        result = self.filter.domain_mappings(apps, self.primary, False)
-        self.assertCountEqual(result, expected)
-
-    def test_no_domains_key_no_autobuild(self):
-        """
-        App without 'server.domains' produces no mappings regardless of auto_build_aliases.
-        """
-        apps = {
-            "web-app-dashboard": {
-                # no 'server' or 'domains'
-            }
-        }
-        result = self.filter.domain_mappings(apps, self.primary, False)
-        self.assertEqual(result, [])
-
-    def test_empty_server_domains_no_autobuild(self):
-        """
-        Explicit empty server.domains should not generate aliases when auto_build_aliases is False.
-        """
-        apps = {"web-app-dashboard": {"server": {"domains": {}}}}
-        result = self.filter.domain_mappings(apps, self.primary, False)
-        self.assertEqual(result, [])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/plugins/lookup/test_current_play_redirect_domains.py

@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from plugins.lookup.current_play_redirect_domains import LookupModule
+
+
+class _DummyTemplar:
+    def __init__(self, available_variables=None):
+        self.available_variables = available_variables or {}
+
+
+class CurrentPlayRedirectDomainsLookupTests(unittest.TestCase):
+    def _make_lookup(self, variables=None):
+        lm = LookupModule()
+        lm._templar = _DummyTemplar(variables or {})
+        lm._loader = None
+        return lm
+
+    def _patch_lookups(self, *, deployed=None, current_play_apps=None):
+        def _get(name, *args, **kwargs):
+            plugin = MagicMock()
+            if name == "deployment":
+                plugin.run.return_value = [{"deployed": list(deployed or [])}]
+            elif name == "applications_current_play":
+                plugin.run.return_value = [current_play_apps or {}]
+            else:
+                plugin.run.return_value = [None]
+            return plugin
+
+        return patch(
+            "plugins.lookup.current_play_redirect_domains.lookup_loader.get",
+            side_effect=_get,
+        )
+
+    def test_empty_inputs_returns_empty_list(self):
+        lm = self._make_lookup()
+        with self._patch_lookups(deployed=[], current_play_apps={}):
+            result = lm.run(terms=[], variables={})[0]
+        self.assertEqual(result, [])
+
+    def test_user_redirect_domain_mappings_preserved(self):
+        lm = self._make_lookup()
+        variables = {
+            "redirect_domain_mappings": [
+                {"source": "old.example.com", "target": "new.example.com"},
+            ],
+        }
+        with self._patch_lookups(deployed=[], current_play_apps={}):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertEqual(
+            result,
+            [{"source": "old.example.com", "target": "new.example.com"}],
+        )
+
+    def test_primary_redirect_appended_when_rdr_domains_deployed(self):
+        lm = self._make_lookup()
+        variables = {
+            "DOMAIN_PRIMARY": "infinito.example",
+            "DOMAIN_HOMEPAGE": "infinito.nexus",
+        }
+        with self._patch_lookups(
+            deployed=["web-opt-rdr-domains"], current_play_apps={}
+        ):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertIn(
+            {"source": "infinito.example", "target": "infinito.nexus"},
+            result,
+        )
+
+    def test_primary_redirect_absent_when_rdr_domains_not_deployed(self):
+        lm = self._make_lookup()
+        variables = {
+            "DOMAIN_PRIMARY": "infinito.example",
+            "DOMAIN_HOMEPAGE": "infinito.nexus",
+        }
+        with self._patch_lookups(deployed=["web-opt-rdr-www"], current_play_apps={}):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertEqual(result, [])
+
+    def test_primary_redirect_absent_when_domain_homepage_missing(self):
+        lm = self._make_lookup()
+        variables = {"DOMAIN_PRIMARY": "infinito.example"}
+        with self._patch_lookups(
+            deployed=["web-opt-rdr-domains"], current_play_apps={}
+        ):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertEqual(result, [])
+
+    def test_per_app_domain_mappings_included(self):
+        lm = self._make_lookup()
+        apps = {
+            "web-app-foo": {
+                "server": {
+                    "domains": {
+                        "canonical": ["foo.example.com"],
+                        "aliases": ["alt-foo.example.com"],
+                    }
+                }
+            }
+        }
+        variables = {
+            "DOMAIN_PRIMARY": "example.com",
+            "AUTO_BUILD_ALIASES": False,
+        }
+        with self._patch_lookups(deployed=[], current_play_apps=apps):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertIn(
+            {"source": "alt-foo.example.com", "target": "foo.example.com"},
+            result,
+        )
+
+    def test_user_mapping_overrides_per_app_mapping_on_same_source(self):
+        lm = self._make_lookup()
+        apps = {
+            "web-app-foo": {
+                "server": {
+                    "domains": {
+                        "canonical": ["foo.example.com"],
+                        "aliases": ["alt.example.com"],
+                    }
+                }
+            }
+        }
+        variables = {
+            "DOMAIN_PRIMARY": "example.com",
+            "AUTO_BUILD_ALIASES": False,
+            "redirect_domain_mappings": [
+                {"source": "alt.example.com", "target": "OVERRIDE.example.com"},
+            ],
+        }
+        with self._patch_lookups(deployed=[], current_play_apps=apps):
+            result = lm.run(terms=[], variables=variables)[0]
+        for entry in result:
+            if entry["source"] == "alt.example.com":
+                self.assertEqual(entry["target"], "OVERRIDE.example.com")
+                break
+        else:
+            self.fail("expected entry with source alt.example.com")
+
+    def test_combined_user_primary_and_per_app(self):
+        lm = self._make_lookup()
+        apps = {
+            "web-app-foo": {
+                "server": {
+                    "domains": {
+                        "canonical": ["foo.example.com"],
+                        "aliases": ["alt-foo.example.com"],
+                    }
+                }
+            }
+        }
+        variables = {
+            "DOMAIN_PRIMARY": "infinito.example",
+            "DOMAIN_HOMEPAGE": "infinito.nexus",
+            "AUTO_BUILD_ALIASES": False,
+            "redirect_domain_mappings": [
+                {"source": "legacy.example.com", "target": "current.example.com"},
+            ],
+        }
+        with self._patch_lookups(
+            deployed=["web-opt-rdr-domains"], current_play_apps=apps
+        ):
+            result = lm.run(terms=[], variables=variables)[0]
+        sources = {entry["source"]: entry["target"] for entry in result}
+        self.assertEqual(sources.get("legacy.example.com"), "current.example.com")
+        self.assertEqual(sources.get("infinito.example"), "infinito.nexus")
+        self.assertEqual(sources.get("alt-foo.example.com"), "foo.example.com")
+
+    def test_non_list_redirect_domain_mappings_treated_as_empty(self):
+        lm = self._make_lookup()
+        variables = {"redirect_domain_mappings": "not-a-list"}
+        with self._patch_lookups(deployed=[], current_play_apps={}):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertEqual(result, [])
+
+    def test_non_mapping_applications_current_play_skipped(self):
+        lm = self._make_lookup()
+        variables = {
+            "DOMAIN_PRIMARY": "infinito.example",
+            "DOMAIN_HOMEPAGE": "infinito.nexus",
+        }
+        with self._patch_lookups(
+            deployed=["web-opt-rdr-domains"], current_play_apps=["not", "a", "dict"]
+        ):
+            result = lm.run(terms=[], variables=variables)[0]
+        self.assertEqual(
+            result,
+            [{"source": "infinito.example", "target": "infinito.nexus"}],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/unit/roles/sys-ctl-hlth-csp/lookup_plugins/test_csp_skip_domains.py

@@ -56,7 +56,7 @@ class CspSkipDomainsLookupTests(unittest.TestCase):
         }
         self.assertEqual(self._run(apps), ["bridgy.example.com", "fed.example.com"])
 
-    def test_app_without_opt_out_is_not_skipped(self):
+    def test_app_without_status_code_ge_400_is_not_skipped(self):
         apps = {
             "web-app-foo": {
                 "server": {
@@ -67,135 +67,58 @@ class CspSkipDomainsLookupTests(unittest.TestCase):
         }
         self.assertEqual(self._run(apps), [])
 
-    def test_csp_health_skip_covers_canonical_and_www_prefix(self):
+    def test_app_without_status_codes_field_is_not_skipped(self):
         apps = {
-            "web-opt-rdr-www": {
-                "server": {
-                    "domains": {"canonical": ["redirect.example.com"]},
-                    "csp": {"health": {"skip": True}},
-                }
-            }
+            "web-app-foo": {"server": {"domains": {"canonical": ["foo.example.com"]}}}
         }
-        self.assertEqual(
-            self._run(apps),
-            ["redirect.example.com", "www.redirect.example.com"],
-        )
+        self.assertEqual(self._run(apps), [])
 
-    def test_csp_health_skip_covers_aliases_and_their_www_prefix(self):
-        apps = {
-            "web-opt-rdr-www": {
-                "server": {
-                    "domains": {
-                        "canonical": ["redirect.example.com"],
-                        "aliases": ["alias.example.com"],
-                    },
-                    "csp": {"health": {"skip": True}},
-                }
-            }
-        }
-        self.assertEqual(
-            self._run(apps),
-            [
-                "alias.example.com",
-                "redirect.example.com",
-                "www.alias.example.com",
-                "www.redirect.example.com",
-            ],
-        )
-
-    def test_status_code_path_does_not_add_www_prefix(self):
+    def test_group_names_selection_filters_apps(self):
         apps = {
             "web-app-bridgy": {
                 "server": {
                     "domains": {"canonical": ["bridgy.example.com"]},
                     "status_codes": {"default": [404]},
                 }
-            }
-        }
-        result = self._run(apps)
-        self.assertIn("bridgy.example.com", result)
-        self.assertNotIn("www.bridgy.example.com", result)
-
-    def test_group_names_selection_filters_apps(self):
-        apps = {
-            "web-opt-rdr-www": {
-                "server": {
-                    "domains": {"canonical": ["redirect.example.com"]},
-                    "csp": {"health": {"skip": True}},
-                }
             },
-            "web-app-foo": {
+            "web-app-other": {
                 "server": {
-                    "domains": {"canonical": ["foo.example.com"]},
+                    "domains": {"canonical": ["other.example.com"]},
                     "status_codes": {"default": [404]},
                 }
             },
         }
-        result = self._run(apps, group_names=["web-opt-rdr-www"])
-        self.assertEqual(result, ["redirect.example.com", "www.redirect.example.com"])
+        self.assertEqual(
+            self._run(apps, group_names=["web-app-bridgy"]),
+            ["bridgy.example.com"],
+        )
 
     def test_group_names_csv_string_is_accepted(self):
         apps = {
             "a": {
                 "server": {
                     "domains": {"canonical": ["a.example.com"]},
-                    "csp": {"health": {"skip": True}},
+                    "status_codes": {"default": [404]},
                 }
             },
             "b": {
                 "server": {
                     "domains": {"canonical": ["b.example.com"]},
-                    "csp": {"health": {"skip": True}},
-                }
-            },
-        }
-        result = self._run(apps, group_names="a,b")
-        self.assertEqual(
-            result,
-            [
-                "a.example.com",
-                "b.example.com",
-                "www.a.example.com",
-                "www.b.example.com",
-            ],
-        )
-
-    def test_combined_opt_outs_dedupe(self):
-        apps = {
-            "web-opt-rdr-www": {
-                "server": {
-                    "domains": {"canonical": ["redirect.example.com"]},
-                    "csp": {"health": {"skip": True}},
-                }
-            },
-            "web-app-bridgy": {
-                "server": {
-                    "domains": {"canonical": ["bridgy.example.com"]},
                     "status_codes": {"default": [404]},
                 }
             },
-            "web-app-foo": {
-                "server": {
-                    "domains": {"canonical": ["foo.example.com"]},
-                    "status_codes": {"default": [200]},
-                }
-            },
         }
         self.assertEqual(
-            self._run(apps),
-            [
-                "bridgy.example.com",
-                "redirect.example.com",
-                "www.redirect.example.com",
-            ],
+            self._run(apps, group_names="a,b"),
+            ["a.example.com", "b.example.com"],
         )
 
-    def test_skip_false_is_not_opt_out(self):
+    def test_non_4xx_codes_dont_count(self):
         apps = {
             "web-app-foo": {
                 "server": {
                     "domains": {"canonical": ["foo.example.com"]},
-                    "csp": {"health": {"skip": False}},
+                    "status_codes": {"default": [200, 301, 302]},
                 }
             }
         }

tests/unit/roles/web-app-keycloak/filter_plugins/test_redirect_uris.py

@@ -18,10 +18,23 @@ def _load_module_from_path(name, file_path):
     return module
 
 
+_STUBBED_MODULE_NAMES = (
+    "utils",
+    "utils.roles.applications",
+    "utils.roles.applications.config",
+    "utils.get_url",
+)
+
+
 class RedirectUrisTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
-        # Create stub package: utils, with applications.config and get_url submodules.
+        cls._orig_sys_modules = {
+            name: sys.modules[name]
+            for name in _STUBBED_MODULE_NAMES
+            if name in sys.modules
+        }
+
         mu = types.ModuleType("utils")
         mu_apps = types.ModuleType("utils.roles.applications")
         mu_config = types.ModuleType("utils.roles.applications.config")
@@ -78,6 +91,14 @@ class RedirectUrisTest(unittest.TestCase):
         self.plugin.get = self._orig_get
         self.plugin.get_url = self._orig_get_url
 
+    @classmethod
+    def tearDownClass(cls):
+        for name in _STUBBED_MODULE_NAMES:
+            if name in cls._orig_sys_modules:
+                sys.modules[name] = cls._orig_sys_modules[name]
+            else:
+                sys.modules.pop(name, None)
+
     def test_single_domain_oauth2_enabled(self):
         domains = {"app1": "example.org"}
         applications = {"app1": {"services": {"oauth2": {"enabled": True}}}}

tests/unit/roles/web-opt-rdr-www/__init__.py

@@ -0,0 +1,3 @@
+from pathlib import Path
+
+PROJECT_ROOT: Path = Path(__file__).resolve().parents[4]

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

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

tests/unit/roles/web-opt-rdr-www/lookup_plugins/test_www_redirect_domains.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+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
+    sys.modules[name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class _DummyTemplar:
+    def __init__(self, available_variables: dict | None = None):
+        self.available_variables = available_variables or {}
+
+
+class WwwRedirectDomainsLookupTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.module = _load_module(
+            "roles/web-opt-rdr-www/lookup_plugins/www_redirect_domains.py",
+            "www_redirect_domains_lookup",
+        )
+        # Rebind possibly globally-mocked names to the real implementations
+        # so this suite is robust against test ordering / pollution.
+        from utils.roles.applications.config import get as _real_get
+
+        cls.module.get = _real_get
+
+    def _run(self, applications, variables=None, role_canonical=None, **kwargs):
+        lm = self.module.LookupModule()
+        lm._templar = _DummyTemplar(variables or {})
+
+        domain_plugin = MagicMock()
+        domain_plugin.run.return_value = [role_canonical or ""]
+
+        with (
+            patch.object(
+                self.module, "get_merged_applications", return_value=applications
+            ),
+            patch.object(self.module.lookup_loader, "get", return_value=domain_plugin),
+        ):
+            return lm.run(terms=[], variables=variables or {}, **kwargs)[0]
+
+    def test_empty_applications_returns_empty_list(self):
+        self.assertEqual(self._run({}), [])
+
+    def test_canonical_gets_www_prefix(self):
+        apps = {
+            "web-app-foo": {"server": {"domains": {"canonical": ["foo.example.com"]}}}
+        }
+        self.assertEqual(
+            self._run(apps),
+            [{"source": "www.foo.example.com", "target": "foo.example.com"}],
+        )
+
+    def test_already_www_prefixed_alias_targets_bare_form(self):
+        apps = {
+            "web-opt-rdr-www": {
+                "server": {
+                    "domains": {
+                        "canonical": ["w3redirect.example.com"],
+                        "aliases": ["www.example.com"],
+                    }
+                }
+            }
+        }
+        self.assertEqual(
+            self._run(apps),
+            [
+                {"source": "www.example.com", "target": "example.com"},
+                {
+                    "source": "www.w3redirect.example.com",
+                    "target": "w3redirect.example.com",
+                },
+            ],
+        )
+
+    def test_aliases_without_www_get_prefix(self):
+        apps = {
+            "web-app-foo": {
+                "server": {
+                    "domains": {
+                        "canonical": ["foo.example.com"],
+                        "aliases": ["alt.example.com"],
+                    }
+                }
+            }
+        }
+        self.assertEqual(
+            self._run(apps),
+            [
+                {"source": "www.alt.example.com", "target": "alt.example.com"},
+                {"source": "www.foo.example.com", "target": "foo.example.com"},
+            ],
+        )
+
+    def test_group_names_selection_filters_apps(self):
+        apps = {
+            "web-app-foo": {"server": {"domains": {"canonical": ["foo.example.com"]}}},
+            "web-app-bar": {"server": {"domains": {"canonical": ["bar.example.com"]}}},
+        }
+        self.assertEqual(
+            self._run(apps, group_names=["web-app-foo"]),
+            [{"source": "www.foo.example.com", "target": "foo.example.com"}],
+        )
+
+    def test_group_names_csv_string_accepted(self):
+        apps = {
+            "a": {"server": {"domains": {"canonical": ["a.example.com"]}}},
+            "b": {"server": {"domains": {"canonical": ["b.example.com"]}}},
+            "c": {"server": {"domains": {"canonical": ["c.example.com"]}}},
+        }
+        self.assertEqual(
+            self._run(apps, group_names="a, c"),
+            [
+                {"source": "www.a.example.com", "target": "a.example.com"},
+                {"source": "www.c.example.com", "target": "c.example.com"},
+            ],
+        )
+
+    def test_dedupes_when_same_mapping_appears_twice(self):
+        apps = {
+            "x": {
+                "server": {
+                    "domains": {
+                        "canonical": ["x.example.com"],
+                        "aliases": ["www.x.example.com"],
+                    }
+                }
+            }
+        }
+        self.assertEqual(
+            self._run(apps),
+            [{"source": "www.x.example.com", "target": "x.example.com"}],
+        )
+
+    def test_empty_string_canonical_is_ignored(self):
+        apps = {"x": {"server": {"domains": {"canonical": ["", "valid.example.com"]}}}}
+        self.assertEqual(
+            self._run(apps),
+            [{"source": "www.valid.example.com", "target": "valid.example.com"}],
+        )
+
+    def test_non_mapping_applications_returns_empty(self):
+        self.assertEqual(self._run([]), [])
+
+    def test_role_canonical_mapping_appended_via_domain_lookup(self):
+        apps = {
+            "web-opt-rdr-www": {
+                "server": {
+                    "domains": {
+                        "canonical": ["w3redirect.example.com"],
+                        "aliases": ["www.example.com"],
+                    }
+                }
+            }
+        }
+        result = self._run(
+            apps,
+            variables={"DOMAIN_HOMEPAGE": "infinito.nexus"},
+            role_canonical="w3redirect.example.com",
+        )
+        self.assertIn(
+            {"source": "w3redirect.example.com", "target": "infinito.nexus"},
+            result,
+        )
+
+    def test_role_mapping_absent_when_domain_homepage_missing(self):
+        apps = {
+            "web-opt-rdr-www": {
+                "server": {"domains": {"canonical": ["w3redirect.example.com"]}}
+            }
+        }
+        result = self._run(apps, role_canonical="w3redirect.example.com")
+        sources = [m["source"] for m in result]
+        self.assertNotIn("w3redirect.example.com", sources)
+
+    def test_role_mapping_absent_when_role_not_deployed(self):
+        apps = {
+            "web-app-foo": {"server": {"domains": {"canonical": ["foo.example.com"]}}}
+        }
+        result = self._run(
+            apps,
+            variables={"DOMAIN_HOMEPAGE": "infinito.nexus"},
+            role_canonical="w3redirect.example.com",
+        )
+        sources = [m["source"] for m in result]
+        self.assertNotIn("w3redirect.example.com", sources)
+
+    def test_role_mapping_not_duplicated(self):
+        apps = {
+            "web-opt-rdr-www": {
+                "server": {"domains": {"canonical": ["w3redirect.example.com"]}}
+            }
+        }
+        result = self._run(
+            apps,
+            variables={"DOMAIN_HOMEPAGE": "infinito.nexus"},
+            role_canonical="w3redirect.example.com",
+        )
+        keys = [(m["source"], m["target"]) for m in result]
+        self.assertEqual(keys.count(("w3redirect.example.com", "infinito.nexus")), 1)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)