import os
import warnings
from pathlib import Path
from tempfile import TemporaryDirectory
from textwrap import dedent
from unittest.mock import patch

import pytest

from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.client.transports import PythonStdioTransport
from fastmcp.settings import Settings
from fastmcp.utilities.tests import caplog_for_fastmcp

# reset deprecation warnings for this module
pytestmark = pytest.mark.filterwarnings("default::DeprecationWarning")


class TestSettingsImport:
    """Test settings import."""

    async def test_settings_from_environment_issue_1749(self):
        """Test that when auth is enabled, the server starts."""

        script = dedent("""
        import os

        os.environ["FASTMCP_SERVER_AUTH"] = "fastmcp.server.auth.providers.azure.AzureProvider"

        os.environ["FASTMCP_SERVER_AUTH_AZURE_TENANT_ID"] = "A_Valid_Value"
        os.environ["FASTMCP_SERVER_AUTH_AZURE_CLIENT_ID"] = "A_Valid_Value"
        os.environ["FASTMCP_SERVER_AUTH_AZURE_CLIENT_SECRET"] = "A_Valid_Value"
        os.environ["FASTMCP_SERVER_AUTH_AZURE_REDIRECT_PATH"] = "/auth/callback"
        os.environ["FASTMCP_SERVER_AUTH_AZURE_BASE_URL"] = "http://localhost:8000"
        os.environ["FASTMCP_SERVER_AUTH_AZURE_REQUIRED_SCOPES"] = "User.Read,email,profile"

        import fastmcp
        
        mcp = fastmcp.FastMCP("TestServer")

        mcp.run()
        """)

        with TemporaryDirectory() as temp_dir:
            server_file = Path(temp_dir) / "server.py"
            server_file.write_text(script)

            transport: PythonStdioTransport = PythonStdioTransport(
                script_path=server_file
            )

            async with Client[PythonStdioTransport](transport=transport) as client:
                tools = await client.list_tools()

                assert tools == []


class TestDeprecatedServerInitKwargs:
    """Test deprecated server initialization keyword arguments."""

    def test_log_level_deprecation_warning(self):
        """Test that log_level raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `log_level` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", log_level="DEBUG")

        # Verify the setting is still applied
        assert server._deprecated_settings.log_level == "DEBUG"

    def test_debug_deprecation_warning(self):
        """Test that debug raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `debug` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", debug=True)

        # Verify the setting is still applied
        assert server._deprecated_settings.debug is True

    def test_host_deprecation_warning(self):
        """Test that host raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `host` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", host="0.0.0.0")

        # Verify the setting is still applied
        assert server._deprecated_settings.host == "0.0.0.0"

    def test_port_deprecation_warning(self):
        """Test that port raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `port` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", port=8080)

        # Verify the setting is still applied
        assert server._deprecated_settings.port == 8080

    def test_sse_path_deprecation_warning(self):
        """Test that sse_path raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `sse_path` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", sse_path="/custom-sse")

        # Verify the setting is still applied
        assert server._deprecated_settings.sse_path == "/custom-sse"

    def test_message_path_deprecation_warning(self):
        """Test that message_path raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `message_path` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", message_path="/custom-message")

        # Verify the setting is still applied
        assert server._deprecated_settings.message_path == "/custom-message"

    def test_streamable_http_path_deprecation_warning(self):
        """Test that streamable_http_path raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `streamable_http_path` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", streamable_http_path="/custom-http")

        # Verify the setting is still applied
        assert server._deprecated_settings.streamable_http_path == "/custom-http"

    def test_json_response_deprecation_warning(self):
        """Test that json_response raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `json_response` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", json_response=True)

        # Verify the setting is still applied
        assert server._deprecated_settings.json_response is True

    def test_stateless_http_deprecation_warning(self):
        """Test that stateless_http raises a deprecation warning."""
        with pytest.warns(
            DeprecationWarning,
            match=r"Providing `stateless_http` when creating a server is deprecated\. Provide it when calling `run` or as a global setting instead\.",
        ):
            server = FastMCP("TestServer", stateless_http=True)

        # Verify the setting is still applied
        assert server._deprecated_settings.stateless_http is True

    def test_multiple_deprecated_kwargs_warnings(self):
        """Test that multiple deprecated kwargs each raise their own warning."""
        with warnings.catch_warnings(record=True) as recorded_warnings:
            warnings.simplefilter("always")
            server = FastMCP(
                "TestServer",
                log_level="INFO",
                debug=False,
                host="127.0.0.1",
                port=9999,
                sse_path="/sse/",
                message_path="/msg",
                streamable_http_path="/http",
                json_response=False,
                stateless_http=False,
            )

        # Should have 9 deprecation warnings (one for each deprecated parameter)
        deprecation_warnings = [
            w for w in recorded_warnings if issubclass(w.category, DeprecationWarning)
        ]
        assert len(deprecation_warnings) == 9

        # Verify all expected parameters are mentioned in warnings
        expected_params = {
            "log_level",
            "debug",
            "host",
            "port",
            "sse_path",
            "message_path",
            "streamable_http_path",
            "json_response",
            "stateless_http",
        }
        mentioned_params = set()
        for warning in deprecation_warnings:
            message = str(warning.message)
            for param in expected_params:
                if f"Providing `{param}`" in message:
                    mentioned_params.add(param)

        assert mentioned_params == expected_params

        # Verify all settings are still applied
        assert server._deprecated_settings.log_level == "INFO"
        assert server._deprecated_settings.debug is False
        assert server._deprecated_settings.host == "127.0.0.1"
        assert server._deprecated_settings.port == 9999
        assert server._deprecated_settings.sse_path == "/sse/"
        assert server._deprecated_settings.message_path == "/msg"
        assert server._deprecated_settings.streamable_http_path == "/http"
        assert server._deprecated_settings.json_response is False
        assert server._deprecated_settings.stateless_http is False

    def test_non_deprecated_kwargs_no_warnings(self):
        """Test that non-deprecated kwargs don't raise warnings."""
        with warnings.catch_warnings(record=True) as recorded_warnings:
            warnings.simplefilter("always")
            server = FastMCP(
                name="TestServer",
                instructions="Test instructions",
                on_duplicate_tools="warn",
                on_duplicate_resources="error",
                on_duplicate_prompts="replace",
                resource_prefix_format="path",
                mask_error_details=True,
            )

        # Should have no deprecation warnings
        deprecation_warnings = [
            w for w in recorded_warnings if issubclass(w.category, DeprecationWarning)
        ]
        assert len(deprecation_warnings) == 0

        # Verify server was created successfully
        assert server.name == "TestServer"
        assert server.instructions == "Test instructions"

    def test_none_values_no_warnings(self):
        """Test that None values for deprecated kwargs don't raise warnings."""
        with warnings.catch_warnings(record=True) as recorded_warnings:
            warnings.simplefilter("always")
            FastMCP(
                "TestServer",
                log_level=None,
                debug=None,
                host=None,
                port=None,
                sse_path=None,
                message_path=None,
                streamable_http_path=None,
                json_response=None,
                stateless_http=None,
            )

        # Should have no deprecation warnings for None values
        deprecation_warnings = [
            w for w in recorded_warnings if issubclass(w.category, DeprecationWarning)
        ]
        assert len(deprecation_warnings) == 0

    def test_deprecated_settings_inheritance_from_global(self):
        """Test that deprecated settings inherit from global settings when not provided."""
        # Mock fastmcp.settings to test inheritance
        with patch("fastmcp.settings") as mock_settings:
            mock_settings.model_dump.return_value = {
                "log_level": "WARNING",
                "debug": True,
                "host": "0.0.0.0",
                "port": 3000,
                "sse_path": "/events",
                "message_path": "/messages",
                "streamable_http_path": "/stream",
                "json_response": True,
                "stateless_http": True,
            }
            mock_settings.server_auth = None  # Add server_auth attribute

            server = FastMCP("TestServer")

            # Verify settings are inherited from global settings
            assert server._deprecated_settings.log_level == "WARNING"
            assert server._deprecated_settings.debug is True
            assert server._deprecated_settings.host == "0.0.0.0"
            assert server._deprecated_settings.port == 3000
            assert server._deprecated_settings.sse_path == "/events"
            assert server._deprecated_settings.message_path == "/messages"
            assert server._deprecated_settings.streamable_http_path == "/stream"
            assert server._deprecated_settings.json_response is True
            assert server._deprecated_settings.stateless_http is True

    def test_deprecated_settings_override_global(self):
        """Test that deprecated settings override global settings when provided."""
        # Mock fastmcp.settings to test override behavior
        with patch("fastmcp.settings") as mock_settings:
            mock_settings.model_dump.return_value = {
                "log_level": "WARNING",
                "debug": True,
                "host": "0.0.0.0",
                "port": 3000,
                "sse_path": "/events",
                "message_path": "/messages",
                "streamable_http_path": "/stream",
                "json_response": True,
                "stateless_http": True,
            }
            mock_settings.server_auth = None  # Add server_auth attribute

            with warnings.catch_warnings():
                warnings.simplefilter("ignore")  # Ignore warnings for this test
                server = FastMCP(
                    "TestServer",
                    log_level="ERROR",
                    debug=False,
                    host="127.0.0.1",
                    port=8080,
                )

            # Verify provided settings override global settings
            assert server._deprecated_settings.log_level == "ERROR"
            assert server._deprecated_settings.debug is False
            assert server._deprecated_settings.host == "127.0.0.1"
            assert server._deprecated_settings.port == 8080
            # Non-overridden settings should still come from global
            assert server._deprecated_settings.sse_path == "/events"
            assert server._deprecated_settings.message_path == "/messages"
            assert server._deprecated_settings.streamable_http_path == "/stream"
            assert server._deprecated_settings.json_response is True
            assert server._deprecated_settings.stateless_http is True

    def test_stacklevel_points_to_constructor_call(self):
        """Test that deprecation warnings point to the FastMCP constructor call."""
        with warnings.catch_warnings(record=True) as recorded_warnings:
            warnings.simplefilter("always")

            FastMCP("TestServer", log_level="DEBUG")

        # Should have exactly one deprecation warning
        deprecation_warnings = [
            w for w in recorded_warnings if issubclass(w.category, DeprecationWarning)
        ]
        assert len(deprecation_warnings) == 1

        # The warning should point to the server.py file where FastMCP.__init__ is called
        # This verifies the stacklevel is working as intended (pointing to constructor)
        warning = deprecation_warnings[0]
        assert "server.py" in warning.filename


class TestDeprecatedEnvironmentVariables:
    """Test deprecated environment variable prefixes."""

    def test_fastmcp_server_env_var_deprecation_warning(self, caplog):
        """Test that FASTMCP_SERVER_ environment variables emit deprecation warnings."""
        env_var_name = "FASTMCP_SERVER_HOST"
        original_value = os.environ.get(env_var_name)

        try:
            os.environ[env_var_name] = "192.168.1.1"

            with caplog_for_fastmcp(caplog):
                settings = Settings()

            # Check that a warning was logged
            assert any(
                "Using `FASTMCP_SERVER_` environment variables is deprecated. Use `FASTMCP_` instead."
                in record.message
                for record in caplog.records
                if record.levelname == "WARNING"
            )

            # Verify the setting is still applied
            assert settings.host == "192.168.1.1"

        finally:
            # Clean up environment variable
            if original_value is not None:
                os.environ[env_var_name] = original_value
            else:
                os.environ.pop(env_var_name, None)


class TestDeprecatedSettingsProperty:
    """Test deprecated settings property access."""

    def test_settings_property_deprecation_warning(self, caplog):
        """Test that accessing fastmcp.settings.settings logs a deprecation warning."""
        from fastmcp import settings

        with caplog_for_fastmcp(caplog):
            # Access the deprecated property
            deprecated_settings = settings.settings

        # Check that a warning was logged
        assert any(
            "Using fastmcp.settings.settings is deprecated. Use fastmcp.settings instead."
            in record.message
            for record in caplog.records
            if record.levelname == "WARNING"
        )

        # Verify it still returns the same settings object
        assert deprecated_settings is settings
        assert isinstance(deprecated_settings, Settings)