"""Surface-level invariants on the MCP tool registry.

These guard against future tool-author drift on three things that
previous launch-readiness audits keep re-discovering:

1. **Tool-name denylist** — there are categories of operations that
   MUST never reach an MCP / CLI surface, even for IT-Admin (the user's
   stated red lines: "no asset wipe, no clearing history, no 越权").
   A forensic audit-log cleanup, a "force-offboard" override, or an
   "impersonate-user" tool would all be reachable via PAT and would
   hand a stolen-PAT attacker catastrophic power. We refuse them
   structurally rather than relying on every reviewer to remember.

2. **Hint constant in sync with reality** — the
   ``_REGISTERED_TOOLS_COUNT_HINT`` value at the bottom of
   ``app/mcp/tools.py`` is the source of truth that PLATFORM.md /
   ROADMAP.md / external README quote. Drift between the hint and the
   actual registered count means future docs lie about platform
   breadth.

3. **PAT plaintext field never reaches the wire** — even if a future
   tool accidentally returns a row carrying a ``plaintext`` /
   ``pat_plaintext`` / ``token_plaintext`` field, the ``serialize()``
   deny-list must mask it. This catches regressions to that mask list.

Run-cost: imports ``app.mcp.tools`` (heavy), so it is in its own file
rather than appended to a fast-suite.
"""
from __future__ import annotations

import re

# Forbidden tool-name patterns. The CLI reaches every MCP tool, so any
# of these shapes would be a launch blocker.
_FORBIDDEN_TOOL_NAME_PATTERNS = [
    re.compile(r"^force_"),
    re.compile(r"^purge_"),
    re.compile(r"^wipe_"),
    re.compile(r"_wipe$"),
    re.compile(r"^delete_audit"),
    re.compile(r"^clear_audit"),
    re.compile(r"^truncate_"),
    re.compile(r"^impersonate"),
    re.compile(r"^as_user"),
    re.compile(r"^sudo_"),
    re.compile(r"^reset_password\b"),  # cross-user password reset must be web-only
]


def _registered_tool_names() -> list[str]:
    """Pull every registered tool name from the FastMCP server.

    FastMCP exposes the tool registry via ``_tool_manager`` (private but
    stable across the 2025-* protocol revisions Helix targets). If the
    attribute name changes, this test should fail loudly so we update
    the helper rather than silently lose coverage.
    """
    from app.mcp.server import mcp_server  # noqa: PLC0415

    manager = getattr(mcp_server, "_tool_manager", None)
    assert manager is not None, (
        "FastMCP._tool_manager not found — internal API drifted; "
        "update _registered_tool_names()."
    )
    tools = getattr(manager, "_tools", None)
    if tools is None:
        # Newer FastMCP versions expose ``list_tools()`` as the canonical
        # accessor. Fall back to that.
        list_tools = getattr(manager, "list_tools", None)
        assert callable(list_tools), (
            "Neither _tools dict nor list_tools() found on tool manager."
        )
        tools = list_tools()
    if isinstance(tools, dict):
        return list(tools.keys())
    return [getattr(t, "name", str(t)) for t in tools]


def test_no_forbidden_tool_names() -> None:
    """Refuse tool names that map to launch-blocker semantics.

    See the module docstring for rationale. Add to the denylist when a
    new pattern emerges; never edit a tool's name to skirt this test.
    """
    from app import mcp as _mcp_pkg  # noqa: F401, PLC0415  # registers tools

    names = _registered_tool_names()
    offenders: list[str] = []
    for name in names:
        for pat in _FORBIDDEN_TOOL_NAME_PATTERNS:
            if pat.search(name):
                offenders.append(f"{name} (matches {pat.pattern!r})")
                break
    assert not offenders, (
        "MCP tools matching launch-blocker name patterns are NOT "
        "allowed (see test_mcp_tools_surface.py docstring): "
        + ", ".join(offenders)
    )


def test_registered_count_hint_matches_actual() -> None:
    """Catch drift between the documented tool count and reality.

    PLATFORM.md / ROADMAP.md cite this number; if the hint and the
    registry disagree, our docs lie.
    """
    from app import mcp as _mcp_pkg  # noqa: F401, PLC0415
    from app.mcp.tools import _REGISTERED_TOOLS_COUNT_HINT  # noqa: PLC0415

    actual = len(_registered_tool_names())
    assert _REGISTERED_TOOLS_COUNT_HINT == actual, (
        f"_REGISTERED_TOOLS_COUNT_HINT={_REGISTERED_TOOLS_COUNT_HINT} "
        f"but actual @mcp_server.tool count is {actual}. "
        "Bump the hint at the bottom of app/mcp/tools.py."
    )


def test_data_access_owner_self_transfer_tools_are_registered() -> None:
    from app import mcp as _mcp_pkg  # noqa: F401, PLC0415

    names = _registered_tool_names()
    assert "transfer_my_data_access_dataset_owner" in names
    assert "reassign_data_access_table_owner" in names


def test_descriptions_never_point_at_unregistered_tools() -> None:
    """A tool description that says "use ``foo``" must name a tool that
    actually exists.

    Regression for 2026-07-30: two cloud descriptions told the client to
    call ``list_my_cloud_subscriptions`` — never registered. An AI client
    following them got tool-not-found and, because that phantom tool was
    the ONLY documented way to discover a ``scope_id``, the entire
    cloud-access self-service flow was unreachable over MCP. The same
    scan then found ``list_my_dns_records`` (also phantom) and
    ``list_persons`` / ``search_persons`` (a real HTTP capability with no
    MCP twin). Descriptions are the only discovery surface an LLM has —
    a wrong name there is a dead end, not a typo.

    Scoped deliberately to *imperative* references ("use X", "call X",
    "``X(``") whose first segment is a verb another tool already uses,
    so parameter names and enum values in prose don't trip it. If this
    fires on a legitimate non-tool token, add it to the referring tool's
    signature or rephrase — do not widen the skip list casually.
    """
    from app import mcp as _mcp_pkg  # noqa: F401, PLC0415
    from app.mcp.server import mcp_server  # noqa: PLC0415

    tools = mcp_server._tool_manager._tools
    names = set(tools)
    # Verb vocabulary derived from the registry itself, so it stays
    # calibrated as new tool families land.
    verbs = {n.split("_")[0] for n in names}
    # Every declared parameter name across the surface — ``dry_run``,
    # ``request_type``, ``owner_person_id`` etc. read like tool names but
    # are arguments the prose legitimately names.
    params = {
        p
        for t in tools.values()
        for p in (t.parameters or {}).get("properties", {})
    }
    ref_pattern = re.compile(
        r"(?:use|call|via|then)\s+`*([a-z][a-z0-9_]*)`*"
        r"|`+([a-z][a-z0-9_]*)\(",
        re.IGNORECASE,
    )
    offenders: dict[str, set[str]] = {}
    for name, tool in tools.items():
        for m in ref_pattern.finditer(tool.description or ""):
            token = m.group(1) or m.group(2)
            if "_" not in token or token in names or token in params:
                continue
            if token.split("_")[0] in verbs:
                offenders.setdefault(token, set()).add(name)
    assert not offenders, (
        "MCP tool descriptions reference tool names that are NOT "
        "registered — an AI client following them gets tool-not-found: "
        + "; ".join(
            f"{tok!r} (cited by {sorted(refs)})"
            for tok, refs in sorted(offenders.items())
        )
    )


def test_serialize_deny_list_masks_pat_plaintext() -> None:
    """A future tool that accidentally returns a PAT-mint result row
    must NOT leak the plaintext through ``serialize()``. The deny-list
    masks this by field name regardless of the wrapping container.
    """
    from app.mcp.runtime import serialize  # noqa: PLC0415

    sample = {
        "id": "abc",
        "name": "Claude Desktop",
        "plaintext": "helix_pat_secretsecretsecret",
        "pat_plaintext": "helix_pat_anothersecret",
        "secret_value": "AKIAEXAMPLE",
    }
    out = serialize(sample)
    assert out["plaintext"] == "***"
    assert out["pat_plaintext"] == "***"
    assert out["secret_value"] == "***"
    # Sanity: a normal field still passes through.
    assert out["name"] == "Claude Desktop"


def test_sensitive_keys_subset_of_serialize_deny_fields() -> None:
    """Symmetric invariant: every audit-redacted key MUST also be
    masked outbound on the serialize() path.

    A previous launch-readiness audit (2026-05-16) discovered the two
    sets had drifted: ``_SERIALIZE_DENY_FIELDS`` carried 11 entries
    not in ``_SENSITIVE_KEYS``. A future tool author adding a column
    to one set without the other would re-introduce the same hole.
    Pin the invariant in CI.
    """
    from app.mcp.runtime import _SENSITIVE_KEYS, _SERIALIZE_DENY_FIELDS  # noqa: PLC0415

    missing = _SENSITIVE_KEYS - _SERIALIZE_DENY_FIELDS
    assert not missing, (
        f"Outbound serializer deny-list is missing keys that the "
        f"audit-args redactor scrubs: {sorted(missing)}. Add them to "
        f"_SERIALIZE_DENY_FIELDS in app/mcp/runtime.py."
    )


def test_serialize_redacts_header_style_key_variants() -> None:
    """The outbound serializer normalises keys through the same
    ``_is_sensitive_key`` logic as the inbound audit redactor, so a
    server-returned dict carrying ``X-Api-Key`` / ``X-Refresh-Token`` /
    ``Hashed-Password`` is masked even though those exact strings are
    not literal members of ``_SERIALIZE_DENY_FIELDS``.
    """
    from app.mcp.runtime import serialize  # noqa: PLC0415

    out = serialize({
        "X-Api-Key": "sensitive",
        "Authorization": "Bearer realtoken",
        "X-Refresh-Token": "rt-value",
        "Hashed-Password": "$bcrypt$...",
        "harmless": "value",
    })
    assert out["X-Api-Key"] == "***"
    assert out["Authorization"] == "***"
    assert out["X-Refresh-Token"] == "***"
    assert out["Hashed-Password"] == "***"
    assert out["harmless"] == "value"


def test_serialize_strict_mode_no_false_positives() -> None:
    """The OUTBOUND serializer runs ``_is_sensitive_key(..., strict=True)``
    which skips the dash-split component scan. This guards against the
    over-rotation that bit a prior audit pass: legitimate field names
    containing a short deny word (``token``, ``cookie``, ``pin``,
    ``otp``, ``bearer``) used to be masked. They must NOT be now.
    """
    from app.mcp.runtime import serialize  # noqa: PLC0415

    out = serialize({
        "last_token_rotation_at": "2026-05-15T12:00:00Z",
        "pin_location": "Bldg 4F",
        "bearer_responsibility": "team lead",
        "otp_required_flag": False,
        "session_count": 7,
    })
    assert out["last_token_rotation_at"] == "2026-05-15T12:00:00Z"
    assert out["pin_location"] == "Bldg 4F"
    assert out["bearer_responsibility"] == "team lead"
    assert out["otp_required_flag"] is False
    assert out["session_count"] == 7


def test_redact_str_catches_openai_sk_proj_keys_with_hyphens() -> None:
    """Real OpenAI project keys are ``sk-proj-<seg>-<seg>-...`` with
    hyphens INSIDE the body. An earlier tightening of the regex
    (round-2 fix) accidentally required a contiguous alphanumeric
    body which missed those keys. Pin the corrected coverage.
    """
    from app.mcp.runtime import _redact_str  # noqa: PLC0415

    sample = "sk-proj-AbCdEfGhIjKl-MnOpQrStUvWx-YzAbCdEfGhIj-Kl"
    assert "[redacted-credential]" in _redact_str(sample)
    # Legacy un-prefixed shape still caught.
    assert "[redacted-credential]" in _redact_str(
        "sk-AbCdEfGhIjKlMnOpQrStUvWxYzAbCdEfGhIjKl",
    )
    # Legitimate hyphenated English prose still passes through —
    # without this guard the regex over-rotates on "sk-learn-..."
    # and similar method-name hyphenations.
    assert _redact_str(
        "sk-learn-model-evaluation-pipeline-and-tools",
    ) == "sk-learn-model-evaluation-pipeline-and-tools"


def test_cost_resolve_resource_creator_declares_mutates_true() -> None:
    """The tool writes a ``cmdb_resource_creators`` cache row via
    INSERT...ON DUPLICATE KEY UPDATE. With ``mutates=False`` the body
    session never commits and the row rolls back on context exit —
    every MCP call would re-hit the audit API (rate-limited!), the
    HTTP twin would also re-resolve because there's no cache row,
    and ``cron.cmdb.resource_creator_sweep`` would keep re-queuing
    the same resources forever.

    Caught on prod 2026-05-25 after PR #503's deploy: a sidecar
    backfill that copied the same no-commit pattern wrote 658 cache
    rows that all rolled back; the rows only persisted once
    ``await db.commit()`` was added. The MCP path through
    ``tool_runtime(mutates=False)`` has the same failure mode.

    This regression test reads the source of ``cost_resolve_resource_creator``
    and asserts the ``tool_runtime(...)`` call inside it passes
    ``mutates=True``. AST-level is enough — the runtime semantics are
    determined by that flag.
    """
    import ast
    import inspect

    from app.mcp.tools import cost_resolve_resource_creator  # noqa: PLC0415

    source = inspect.getsource(cost_resolve_resource_creator)
    tree = ast.parse(source.lstrip())

    # Walk: find the `async with tool_runtime(...)` call inside the
    # function body and check its keyword args.
    found_mutates_true = False
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func = node.func
        name = (
            func.id if isinstance(func, ast.Name)
            else (func.attr if isinstance(func, ast.Attribute) else None)
        )
        if name != "tool_runtime":
            continue
        for kw in node.keywords:
            if kw.arg == "mutates" and isinstance(kw.value, ast.Constant):
                if kw.value.value is True:
                    found_mutates_true = True
                    break
    assert found_mutates_true, (
        "cost_resolve_resource_creator must call tool_runtime(..., "
        "mutates=True) — the resolve writes a cmdb_resource_creators "
        "cache row that needs to persist. Without mutates=True the "
        "runtime closes the body session without commit and the row "
        "rolls back. See test docstring for the prod incident."
    )


def test_network_admin_mcp_surface_complete() -> None:
    """Pin the Round-30 follow-up #1 additions so a future refactor
    cannot silently drop them.

    The HTTP surface for network annotate + domain-registration admin
    actions exists at:
      * `PATCH /api/v1/cmdb/domains/{domain_id}` (migration 218 added
        the 3 sync-owned editable fields registrar / expiry_date /
        auto_renew — MCP twin gives AI clients the same surface that
        humans use to fix stale legacy-imported domain rows).
      * `PATCH /api/v1/cmdb/cdn-domains/{cdn_id}`
      * `POST /api/v1/domain-registrations/{reg_id}/handover`
      * `POST /api/v1/domain-registrations/{reg_id}/revoke`

    Each must have its 1:1 MCP tool per the MCP-sync rule.
    """
    from app import mcp as _mcp_pkg  # noqa: F401, PLC0415

    names = set(_registered_tool_names())
    required = {
        "annotate_domain",
        "annotate_cdn_domain",
        "handover_domain_registration",
        "revoke_domain_registration",
    }
    missing = required - names
    assert not missing, (
        f"MCP-sync rule: HTTP routes for {sorted(missing)} have no MCP "
        "twin. Re-add the @mcp_server.tool decorators in "
        "server/app/mcp/tools.py."
    )