"""Strict validation of user-facing variable dicts against role argument specs.

Each spec file's ``argument_specs.inputs.options`` declares top-level variables. Every one the host
defines is checked without Ansible's lenient coercion: a string is never a list, bool or dict. A dict
option whose only sub-option is named ``<name>`` is a collection keyed by name; each entry is checked
against that sub-option, and a null entry means the entry is not managed. Messages name the key path
and the types involved but never a value, so secrets cannot leak into job output.
"""

import glob
from collections.abc import Mapping

from ansible.errors import AnsibleError
from ansible.parsing.vault import EncryptedString
from ansible.plugins.action import ActionBase

KEYED = "<name>"
SPEC_KEYS = {"type", "required", "default", "choices", "elements", "options", "description", "no_log"}
UNRESOLVED = object()

# Numbers pass where a string is declared and digit strings where an int is: the roles read those
# through `| string` and `| int` unchanged. A string posing as a list, bool or dict never passes.
TYPES = {
    "str": lambda v: isinstance(v, (str, int, float, EncryptedString)) and not isinstance(v, bool),
    "int": lambda v: (isinstance(v, int) and not isinstance(v, bool)) or (isinstance(v, str) and v.isdigit()),
    "float": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
    "bool": lambda v: isinstance(v, bool),
    "list": lambda v: isinstance(v, list),
    "dict": lambda v: isinstance(v, Mapping),
    "raw": lambda _: True,
}


def _kind(value):
    if value is None:
        return "null"
    # bool before int: bool is an int subclass.
    for name, cls in (("bool", bool), ("dict", Mapping), ("str", (str, EncryptedString)), ("int", int),
                      ("float", float), ("list", list)):
        if isinstance(value, cls):
            return name
    return type(value).__name__


def _check_spec(options, path):
    for key, spec in options.items():
        unknown = set(spec) - SPEC_KEYS
        if unknown:
            raise AnsibleError(f"{path}.{key}: unsupported spec attributes {', '.join(sorted(unknown))}")
        if spec.get("type", "str") not in TYPES:
            raise AnsibleError(f"{path}.{key}: unsupported spec type {spec['type']}")
        if "options" in spec:
            _check_spec(spec["options"], f"{path}.{key}")


class _Validator:
    def __init__(self, templar):
        self.templar = templar
        self.errors = []

    def resolve(self, value):
        # Task vars arrive untemplated. Render one node at a time so a vaulted leaf is never decrypted;
        # a node that cannot render here (no vault secret, a runtime-only variable) is left unchecked.
        if isinstance(value, str) and self.templar.is_template(value):
            try:
                return self.templar.template(value)
            except AnsibleError:
                return UNRESOLVED
        return value

    def check(self, spec, value, path):
        value = self.resolve(value)
        if value is UNRESOLVED:
            return
        kind = spec.get("type", "str")
        if not TYPES[kind](value):
            self.errors.append(f"{path}: expected {kind}, got {_kind(value)}")
            return
        if "choices" in spec and value not in spec["choices"]:
            shown = "" if spec.get("no_log") else f", got {value!r}"
            self.errors.append(f"{path}: must be one of {', '.join(map(str, spec['choices']))}{shown}")
        # Keyed on the value, not the declared type, so a `raw` option ("false, or a dict of...") is
        # still checked whenever it holds a dict or a list.
        if "options" in spec and isinstance(value, Mapping):
            self.check_options(spec["options"], value, path)
        elif "elements" in spec and isinstance(value, list):
            element = {"type": spec["elements"]}
            if "options" in spec:
                element["options"] = spec["options"]
            for index, item in enumerate(value):
                self.check(element, item, f"{path}[{index}]")

    def check_options(self, options, value, path):
        if set(options) == {KEYED}:
            for name, entry in value.items():
                if entry is not None:
                    self.check(options[KEYED], entry, f"{path}.{name}")
            return
        for key in value:
            if key not in options:
                self.errors.append(f"{path}.{key}: unsupported key, {path} accepts {', '.join(sorted(options))}")
        for key, spec in options.items():
            if key in value:
                self.check(spec, value[key], f"{path}.{key}")
            elif spec.get("required"):
                self.errors.append(f"{path}.{key}: required key is missing")


class ActionModule(ActionBase):
    _requires_connection = False

    def run(self, tmp=None, task_vars=None):
        result = super().run(tmp, task_vars)
        task_vars = task_vars or {}
        validator = _Validator(self._templar)
        declared = {}
        paths = []
        # A pattern (`.../*/roles/*/meta/argument_specs.yml`) picks up every role's spec without a list
        # to keep in sync; one that matches nothing is a broken path, not an empty set.
        for entry in self._task.args["specs"]:
            matches = sorted(glob.glob(entry)) if glob.has_magic(entry) else [entry]
            if not matches:
                raise AnsibleError(f"no spec file matches {entry}")
            paths.extend(matches)
        for path in paths:
            options = self._loader.load_from_file(path)["argument_specs"]["inputs"]["options"]
            _check_spec(options, path)
            for name, spec in options.items():
                if name in declared:
                    raise AnsibleError(f"{name} is declared by both {declared[name]} and {path}")
                declared[name] = path
                if name in task_vars:
                    validator.check(spec, task_vars[name], name)
                elif spec.get("required"):
                    validator.errors.append(f"{name}: required variable is not defined")
        if validator.errors:
            result.update(failed=True, errors=validator.errors,
                          msg="Invalid input variables:\n" + "\n".join(validator.errors))
        else:
            result.update(changed=False, msg="All input variables are valid")
        return result