"""Tests for the per-server MCP identity header (``identity_header``).

An optional per-server config key in ``mcp_servers`` attaches a static or
profile-derived identity header to that server's HTTP/SSE transport
requests:

    mcp_servers:
      remote_api:
        url: "https://my-mcp-server.example.com/mcp"
        identity_header:
          name: "X-User-Id"
          value_from: "static"      # or "profile"
          value: "alice"            # required for value_from: static

Covers:

1. ``_resolve_identity_header`` helper — static mode, profile mode,
   validation failures (warn + ignore, never break the server).

2. HTTP (new SDK ``streamable_http_client``) path attaches the header to
   the user-owned ``httpx.AsyncClient`` when configured, and not otherwise.

3. Explicit per-server ``headers`` with the same name win over the
   identity header (no silent override of user config).

4. stdio servers: ``identity_header`` is warn-and-ignore (headers don't
   exist on stdio transports).
"""

from __future__ import annotations

import asyncio
import logging
from unittest.mock import MagicMock, patch

import pytest


# ---------------------------------------------------------------------------
# _resolve_identity_header helper
# ---------------------------------------------------------------------------


class TestResolveIdentityHeader:
    def test_returns_none_when_unset(self):
        from tools.mcp_tool import _resolve_identity_header

        assert _resolve_identity_header("srv", {}) is None
        assert _resolve_identity_header("srv", {"url": "https://x"}) is None

    def test_static_mode_returns_name_value(self):
        from tools.mcp_tool import _resolve_identity_header

        result = _resolve_identity_header("srv", {
            "identity_header": {
                "name": "X-User-Id",
                "value_from": "static",
                "value": "alice",
            },
        })
        assert result == ("X-User-Id", "alice")

    def test_static_is_default_value_from(self):
        from tools.mcp_tool import _resolve_identity_header

        result = _resolve_identity_header("srv", {
            "identity_header": {"name": "X-User-Id", "value": "bob"},
        })
        assert result == ("X-User-Id", "bob")

    def test_profile_mode_uses_active_profile_name(self):
        from tools.mcp_tool import _resolve_identity_header

        with patch(
            "hermes_cli.profiles.get_active_profile_name",
            return_value="workbot",
        ):
            result = _resolve_identity_header("srv", {
                "identity_header": {
                    "name": "X-Hermes-Profile",
                    "value_from": "profile",
                },
            })
        assert result == ("X-Hermes-Profile", "workbot")

    def test_missing_name_warns_and_returns_none(self, caplog):
        from tools.mcp_tool import _resolve_identity_header

        with caplog.at_level(logging.WARNING):
            result = _resolve_identity_header("srv", {
                "identity_header": {"value": "alice"},
            })
        assert result is None
        assert any("identity_header" in r.message for r in caplog.records)

    def test_static_missing_value_warns_and_returns_none(self, caplog):
        from tools.mcp_tool import _resolve_identity_header

        with caplog.at_level(logging.WARNING):
            result = _resolve_identity_header("srv", {
                "identity_header": {"name": "X-User-Id"},
            })
        assert result is None
        assert any("identity_header" in r.message for r in caplog.records)

    def test_unknown_value_from_warns_and_returns_none(self, caplog):
        from tools.mcp_tool import _resolve_identity_header

        with caplog.at_level(logging.WARNING):
            result = _resolve_identity_header("srv", {
                "identity_header": {
                    "name": "X-User-Id",
                    "value_from": "per_call",
                    "value": "x",
                },
            })
        assert result is None
        assert any("identity_header" in r.message for r in caplog.records)

    def test_non_dict_config_warns_and_returns_none(self, caplog):
        from tools.mcp_tool import _resolve_identity_header

        with caplog.at_level(logging.WARNING):
            result = _resolve_identity_header("srv", {
                "identity_header": "X-User-Id: alice",
            })
        assert result is None
        assert any("identity_header" in r.message for r in caplog.records)


# ---------------------------------------------------------------------------
# HTTP transport — header attached to httpx.AsyncClient
# ---------------------------------------------------------------------------


def _drive_http(server, config):
    """Run ``_run_http`` with the SDK boundary mocked out, capturing the
    kwargs passed to ``httpx.AsyncClient``. Mirrors the pattern in
    ``test_mcp_client_cert.py``.
    """
    from tools.mcp_tool import MCPServerTask, sdk_httpx

    captured: dict = {}

    class DummyAsyncClient:
        def __init__(self, **kwargs):
            captured.update(kwargs)

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

    class DummyTransportCtx:
        async def __aenter__(self):
            return MagicMock(), MagicMock(), (lambda: None)

        async def __aexit__(self, *a):
            return False

    class DummySession:
        def __init__(self, *args, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def initialize(self):
            return None

    async def _discover_tools(self):
        self._shutdown_event.set()

    async def _drive():
        with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", True), \
             patch("tools.mcp_tool._MCP_NEW_HTTP", True), \
             patch.object(sdk_httpx(), "AsyncClient", DummyAsyncClient), \
             patch("tools.mcp_tool.streamable_http_client",
                   return_value=DummyTransportCtx()), \
             patch("tools.mcp_tool.ClientSession", DummySession), \
             patch.object(MCPServerTask, "_discover_tools", _discover_tools):
            await server._run_http(config)

    asyncio.run(_drive())
    return captured


class TestHTTPIdentityHeader:
    def test_header_attached_when_configured(self):
        from tools.mcp_tool import MCPServerTask

        server = MCPServerTask("remote")
        captured = _drive_http(server, {
            "url": "https://example.com/mcp",
            "identity_header": {
                "name": "X-User-Id",
                "value": "alice",
            },
        })
        headers = captured.get("headers") or {}
        assert headers.get("X-User-Id") == "alice"

    def test_header_absent_when_not_configured(self):
        from tools.mcp_tool import MCPServerTask

        server = MCPServerTask("remote")
        captured = _drive_http(server, {
            "url": "https://example.com/mcp",
        })
        headers = captured.get("headers") or {}
        assert not any(k.lower() == "x-user-id" for k in headers)

    def test_explicit_header_with_same_name_wins(self):
        """A user-set per-server header of the same name (any casing) is
        not overridden by the identity header."""
        from tools.mcp_tool import MCPServerTask

        server = MCPServerTask("remote")
        captured = _drive_http(server, {
            "url": "https://example.com/mcp",
            "headers": {"x-user-id": "explicit-wins"},
            "identity_header": {
                "name": "X-User-Id",
                "value": "alice",
            },
        })
        headers = captured.get("headers") or {}
        assert headers.get("x-user-id") == "explicit-wins"
        assert "X-User-Id" not in headers

    def test_profile_mode_header_attached(self):
        from tools.mcp_tool import MCPServerTask

        server = MCPServerTask("remote")
        with patch(
            "hermes_cli.profiles.get_active_profile_name",
            return_value="workbot",
        ):
            captured = _drive_http(server, {
                "url": "https://example.com/mcp",
                "identity_header": {
                    "name": "X-Hermes-Profile",
                    "value_from": "profile",
                },
            })
        headers = captured.get("headers") or {}
        assert headers.get("X-Hermes-Profile") == "workbot"


# ---------------------------------------------------------------------------
# stdio transport — identity_header is warn-and-ignore
# ---------------------------------------------------------------------------


class TestStdioIdentityHeader:
    def test_stdio_warns_and_ignores(self, caplog):
        """identity_header on a stdio server logs a warning and does not
        break the transport path (headers don't exist on stdio)."""
        from tools.mcp_tool import MCPServerTask

        server = MCPServerTask("local")

        async def _drive():
            # Force the SDK-unavailable fast path so no subprocess spawns;
            # the warning must fire before the availability check.
            with patch("tools.mcp_tool._MCP_AVAILABLE", False):
                await server._run_stdio({
                    "command": "echo",
                    "identity_header": {"name": "X-User-Id", "value": "a"},
                })

        with caplog.at_level(logging.WARNING):
            with pytest.raises(ImportError):
                asyncio.run(_drive())

        assert any(
            "identity_header" in r.message and "stdio" in r.message
            for r in caplog.records
        )