computer-playbook/plugins/lookup/compose_file_args.py

# Build compose "-f <file>" arguments for an application instance.
#
# HARD FAIL PRINCIPLE:
# - No silent defaults that change behavior.
# - Fail loudly if required variables are missing/invalid.
#
# Rules:
# - Always include base compose.yml
# - Include compose.override.yml ONLY when the ROLE (application_id) provides one
#   (same logic as tasks/04_files.yml with_first_found)
# - Include compose.ca.override.yml ONLY when:
#     the app has a domain AND TLS is enabled AND TLS mode == "self_signed"
#
# Optional kwargs:
# - include_ca (bool, default True):
#     - True  -> normal behavior (append CA override when enabled+self_signed)
#     - False -> do NOT append CA override (used during CA-inject bootstrap)
#
# IMPORTANT (your requested change):
# - This lookup NO LONGER reads variables['compose'].
# - It ALWAYS builds compose paths via utils.docker.paths_utils.get_docker_paths()
#   using DIR_COMPOSITIONS.

from __future__ import annotations

from pathlib import Path
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.role_path_by_app_id import abs_role_path_by_application_id
from utils.docker.paths_utils import get_docker_paths
from utils.templating.jinja import render_strict


def _as_str(v: Any) -> str:
    return "" if v is None else str(v).strip()


def _require_dict(d: Any, label: str) -> dict:
    if not isinstance(d, dict):
        raise AnsibleError(f"compose_file_args: {label} must be a dict, got {type(d)}")
    return d


def _maybe_template(templar: Any, value: Any) -> Any:
    """
    Render via Ansible templar if available.
    Unit tests may not inject a templar -> then we skip templating.
    """
    if isinstance(value, (dict, list, tuple, int, float, bool)) or value is None:
        return value
    if templar is None:
        return value
    tpl = getattr(templar, "template", None)
    if callable(tpl):
        return tpl(value)
    return value


def _value_has_domain(v: Any) -> bool:
    if v is None:
        return False
    if isinstance(v, str):
        return v.strip() != ""
    if isinstance(v, (list, tuple, set)):
        return any(_value_has_domain(x) for x in v)
    if isinstance(v, dict):
        return any(_value_has_domain(x) for x in v.values())
    return False


def _has_domain(domains: Any, application_id: str) -> bool:
    """
    Dependency-free domain existence check that matches the intent of the old filter.
    """
    if isinstance(domains, dict):
        return _value_has_domain(domains.get(application_id))
    return _value_has_domain(domains)


def _role_provides_override(*, application_id: str) -> bool:
    """
    Mirror tasks/04_files.yml "with_first_found" logic:

      - "{{ application_id | abs_role_path_by_application_id }}/templates/compose.override.yml.j2"
      - "{{ application_id | abs_role_path_by_application_id }}/files/compose.override.yml"

    Only if one of these exists, we append "-f <compose.files.compose_override>".
    """
    role_base = Path(abs_role_path_by_application_id(application_id))
    return (role_base / "templates" / "compose.override.yml.j2").is_file() or (
        role_base / "files" / "compose.override.yml"
    ).is_file()


class LookupModule(LookupBase):
    def run(self, terms, variables: dict | None = None, **kwargs):
        variables = variables or {}

        if not terms or len(terms) != 1:
            raise AnsibleError(
                "compose_file_args: exactly one term required (application_id)"
            )

        application_id = _as_str(terms[0])
        if not application_id:
            raise AnsibleError("compose_file_args: application_id is empty")

        include_ca = kwargs.get("include_ca", True)
        if not isinstance(include_ca, bool):
            raise AnsibleError("compose_file_args: include_ca must be a bool")

        flag = kwargs.get("flag", "-f")
        if flag not in ("-f", "--compose-file", "-c"):
            raise AnsibleError(
                "compose_file_args: flag must be one of -f, --compose-file, -c"
            )

        templar = getattr(self, "_templar", None)

        raw_base_dir = variables.get("DIR_COMPOSITIONS")
        if (
            isinstance(raw_base_dir, str)
            and "{{" in raw_base_dir
            and templar is not None
        ):
            raw_base_dir = templar.template(raw_base_dir)
        base_dir = _as_str(raw_base_dir)
        if not base_dir:
            raise AnsibleError(
                "compose_file_args: missing required variable 'DIR_COMPOSITIONS'"
            )

        compose = get_docker_paths(application_id, base_dir)
        compose = _require_dict(compose, "compose")

        files = _require_dict(compose.get("files"), "compose.files")

        base = render_strict(
            files.get("compose"),
            variables=variables,
            var_name="compose.files.compose",
            err_prefix="compose_file_args",
        )
        override = render_strict(
            files.get("compose_override"),
            variables=variables,
            var_name="compose.files.compose_override",
            err_prefix="compose_file_args",
        )
        ca_override = render_strict(
            files.get("compose_ca_override"),
            variables=variables,
            var_name="compose.files.compose_ca_override",
            err_prefix="compose_file_args",
        )

        if not _as_str(base):
            raise AnsibleError("compose_file_args: compose.files.compose is required")

        parts = [f"{flag} {base}"]

        if _role_provides_override(application_id=application_id):
            if not _as_str(override):
                raise AnsibleError(
                    "compose_file_args: compose.files.compose_override is required "
                    "when the role provides an override file"
                )
            parts.append(f"{flag} {override}")

        if include_ca:
            domains = lookup_loader.get(
                "domains", loader=self._loader, templar=templar
            ).run([], variables=variables)[0]
            if _has_domain(domains, application_id):
                tlsr = lookup_loader.get("tls", self._loader, self._templar)
                tls = tlsr.run([application_id], variables=variables)[0]

                if not isinstance(tls, dict):
                    raise AnsibleError(
                        f"compose_file_args: tls returned non-dict: {type(tls)}"
                    )
                if "enabled" not in tls:
                    raise AnsibleError(
                        "compose_file_args: tls did not return 'enabled'"
                    )
                if "mode" not in tls:
                    raise AnsibleError("compose_file_args: tls did not return 'mode'")

                enabled = bool(tls["enabled"])
                mode = _as_str(tls["mode"])
                if not mode:
                    raise AnsibleError("compose_file_args: tls returned empty 'mode'")

                if enabled and mode == "self_signed":
                    if not _as_str(ca_override):
                        raise AnsibleError(
                            "compose_file_args: compose.files.compose_ca_override is required "
                            "when TLS is enabled and mode is self_signed"
                        )
                    parts.append(f"{flag} {ca_override}")

        return [" ".join(parts)]