computer-playbook/tests/integration/ansible/lookups/config/_validate.py

"""Pure validation helpers shared by the `tests/integration/lookups/config`
test modules. Kept separate from `_scan.py` so each test class wires
exactly the helpers it needs."""

from __future__ import annotations

import fnmatch
from collections.abc import Mapping
from typing import Any

from utils.manager.credential_key import (
    CREDENTIALS_KEY,
    OVERRIDE_SECTION,
    SECRETS_KEY,
)


class PathNotFoundError(Exception):
    """Raised by helpers when a dotted path does not resolve."""


def assert_nested(mapping: Mapping[str, Any], dotted: str, context: str) -> None:
    """Walk ``dotted`` through ``mapping``; raise ``PathNotFoundError`` on miss."""
    keys = dotted.split(".")
    cur: Any = mapping
    for k in keys:
        if not isinstance(cur, Mapping):
            raise PathNotFoundError(f"{context}: expected dict at {k}")
        if k not in cur:
            raise PathNotFoundError(f"{context}: missing '{k}' in '{dotted}'")
        cur = cur[k]


def match_wildcard_segment(mapping: Any, segment: str) -> bool:
    """Return True when ``segment`` matches at least one top-level key.

    ``segment`` may contain ``*`` glob characters; bare ``*`` matches any
    key.
    """
    if not isinstance(mapping, Mapping):
        return False
    if "*" not in segment:
        return segment in mapping
    return any(fnmatch.fnmatchcase(k, segment) for k in mapping)


def match_wildcard_path(mapping: Any, dotted: str) -> bool:
    """Return True when ``dotted`` matches at least one nested path in
    ``mapping``. Each segment may be a literal key, a bare ``*``
    (matches any single key), or a glob containing ``*`` (matches any
    key whose name matches the glob, e.g. ``*_jwt_secret``)."""
    keys = dotted.split(".")

    def walk(cur: Any, idx: int) -> bool:
        if idx == len(keys):
            return True
        if not isinstance(cur, Mapping):
            return False
        key = keys[idx]
        if "*" not in key:
            return key in cur and walk(cur[key], idx + 1)
        return any(
            fnmatch.fnmatchcase(child_key, key) and walk(child_val, idx + 1)
            for child_key, child_val in cur.items()
        )

    return walk(mapping, 0)


def validate_app_path(
    application_defaults: dict[str, Any],
    role_schemas: dict[str, dict[str, Any]],
    user_defaults: dict[str, Any],
    app_id: str,
    dotted: str,
) -> None:
    """Resolve ``dotted`` against the same fallback chain the runtime
    plugin uses: app defaults → users → credentials in defaults →
    credentials in schema → images presence. Raise ``PathNotFoundError`` on
    a final miss."""
    cfg = application_defaults.get(app_id, {})
    try:
        assert_nested(cfg, dotted, app_id)
    except PathNotFoundError:
        pass
    else:
        return
    if dotted.startswith("users."):
        sub = dotted.split(".", 1)[1]
        if sub in user_defaults:
            return
    if dotted.startswith(f"{OVERRIDE_SECTION}."):
        key = dotted.split(f"{OVERRIDE_SECTION}.", 1)[1]
        secrets_cfg = cfg.get(SECRETS_KEY, {}) if isinstance(cfg, Mapping) else {}
        creds_cfg = (
            secrets_cfg.get(CREDENTIALS_KEY, {})
            if isinstance(secrets_cfg, Mapping)
            else {}
        )
        if isinstance(creds_cfg, Mapping) and key in creds_cfg:
            return
        schema = role_schemas.get(app_id, {})
        creds = schema.get(CREDENTIALS_KEY, {}) if isinstance(schema, Mapping) else {}
        if isinstance(creds, Mapping) and key in creds:
            return
        raise PathNotFoundError(f"Credential '{key}' missing for app '{app_id}'")
    if dotted.startswith("images.") and isinstance(cfg.get("images"), Mapping):
        return
    raise PathNotFoundError(f"'{dotted}' not found for '{app_id}'")