"""Unit tests for MCP Composer Service port management and process killing."""

import asyncio
import contextlib
import socket
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from lfx.services.mcp_composer.service import (
    COMPOSER_BACKEND_AUTH_HEADER,
    MCPComposerPortError,
    MCPComposerService,
)


@pytest.fixture
def mcp_service():
    """Create an MCP Composer service instance for testing."""
    return MCPComposerService()


def get_available_local_port() -> int:
    """Ask the OS for a currently available localhost TCP port."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


class TestPortAvailability:
    """Test port availability checking."""

    def test_is_port_available_when_free(self, mcp_service):
        """Test that is_port_available returns True for an available port."""
        test_port = get_available_local_port()
        assert mcp_service._is_port_available(test_port, host="127.0.0.1") is True

    def test_is_port_available_when_in_use(self, mcp_service):
        """Test that is_port_available returns False when port is in use."""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            sock.bind(("127.0.0.1", 0))
            test_port = sock.getsockname()[1]
            sock.listen(1)
            # Port should now be unavailable
            assert mcp_service._is_port_available(test_port, host="127.0.0.1") is False
        finally:
            sock.close()


class TestBackendAuthToken:
    """Test internal Composer-to-Langflow backend authentication config."""

    def test_backend_auth_member_config_adds_internal_header(self, mcp_service):
        project_id = "backend-auth-test"

        config = mcp_service._build_backend_auth_member_config(
            project_id,
            "http://localhost:7860/api/v1/mcp/project/test/streamable",
            legacy_sse_url="http://localhost:7860/api/v1/mcp/project/test/sse",
        )

        token = mcp_service.get_or_create_backend_auth_token(project_id)
        assert mcp_service.validate_backend_auth_token(project_id, token)
        assert [entry["type"] for entry in config] == ["http", "sse"]
        assert config[0]["headers"] == {COMPOSER_BACKEND_AUTH_HEADER: token}
        assert config[1]["headers"] == {COMPOSER_BACKEND_AUTH_HEADER: token}

    @pytest.mark.asyncio
    async def test_oauth_startup_uses_config_path_for_backend_auth(self, mcp_service):
        project_id = "backend-auth-process-test"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }

        mock_settings = MagicMock()
        mock_settings.settings.mcp_composer_version = "==0.1.0.8.10"
        mock_process = MagicMock(pid=2222, poll=MagicMock(return_value=None))

        with (
            patch("lfx.services.mcp_composer.service.get_settings_service", return_value=mock_settings),
            patch.object(
                mcp_service,
                "_write_backend_auth_member_config",
                return_value=Path("/tmp/mcp-composer-test-config.json"),
            ) as mock_write_config,
            patch("subprocess.Popen", return_value=mock_process) as mock_popen,
            patch.object(mcp_service, "_is_port_available", return_value=False),
        ):
            await mcp_service._start_project_composer_process(
                project_id=project_id,
                host="localhost",
                port=9000,
                streamable_http_url="http://test/mcp",
                auth_config=auth_config,
                max_startup_checks=1,
                startup_delay=0.01,
                legacy_sse_url="http://test/sse",
            )

        cmd = mock_popen.call_args[0][0]
        assert "--config_path" in cmd
        assert "--endpoint" not in cmd
        assert "--sse-url" not in cmd
        assert COMPOSER_BACKEND_AUTH_HEADER not in cmd
        mock_write_config.assert_called_once_with(
            project_id,
            "http://test/mcp",
            legacy_sse_url="http://test/sse",
        )


class TestKillProcessOnPort:
    """Test process killing functionality."""

    @pytest.mark.asyncio
    async def test_kill_process_on_port_no_process(self, mcp_service):
        """Test that _kill_process_on_port returns False when no process is found."""
        with patch("asyncio.to_thread") as mock_to_thread:
            # Mock lsof returning no processes
            mock_result = MagicMock()
            mock_result.returncode = 1  # lsof returns 1 when no matches
            mock_result.stdout = ""
            mock_result.stderr = ""
            mock_to_thread.return_value = mock_result

            result = await mcp_service._kill_process_on_port(9999)
            assert result is False

    @pytest.mark.asyncio
    async def test_kill_process_on_port_success(self, mcp_service):
        """Test that _kill_process_on_port successfully kills a process."""
        with patch("asyncio.to_thread") as mock_to_thread:
            # Mock lsof returning a PID
            mock_lsof_result = MagicMock()
            mock_lsof_result.returncode = 0
            mock_lsof_result.stdout = "12345\n"
            mock_lsof_result.stderr = ""

            # Mock kill command succeeding
            mock_kill_result = MagicMock()
            mock_kill_result.returncode = 0
            mock_kill_result.stdout = ""
            mock_kill_result.stderr = ""

            # Set up side effects for two calls: lsof, then kill
            mock_to_thread.side_effect = [mock_lsof_result, mock_kill_result]

            result = await mcp_service._kill_process_on_port(9000)
            assert result is True
            assert mock_to_thread.call_count == 2

    @pytest.mark.asyncio
    async def test_kill_process_on_port_multiple_pids(self, mcp_service):
        """Test that _kill_process_on_port handles multiple PIDs."""
        with patch("asyncio.to_thread") as mock_to_thread:
            # Mock lsof returning multiple PIDs
            mock_lsof_result = MagicMock()
            mock_lsof_result.returncode = 0
            mock_lsof_result.stdout = "12345\n67890\n"
            mock_lsof_result.stderr = ""

            # Mock kill command succeeding for first PID
            mock_kill_result = MagicMock()
            mock_kill_result.returncode = 0

            mock_to_thread.side_effect = [mock_lsof_result, mock_kill_result]

            result = await mcp_service._kill_process_on_port(9000)
            assert result is True

    @pytest.mark.asyncio
    async def test_kill_process_on_port_kill_fails(self, mcp_service):
        """Test that _kill_process_on_port handles kill command failure."""
        with patch("asyncio.to_thread") as mock_to_thread:
            # Mock lsof returning a PID
            mock_lsof_result = MagicMock()
            mock_lsof_result.returncode = 0
            mock_lsof_result.stdout = "12345\n"
            mock_lsof_result.stderr = ""

            # Mock kill command failing
            mock_kill_result = MagicMock()
            mock_kill_result.returncode = 1

            mock_to_thread.side_effect = [mock_lsof_result, mock_kill_result]

            result = await mcp_service._kill_process_on_port(9000)
            assert result is False

    @pytest.mark.asyncio
    async def test_kill_process_on_port_exception_handling(self, mcp_service):
        """Test that _kill_process_on_port handles exceptions gracefully."""
        with patch("asyncio.to_thread", side_effect=Exception("Test error")):
            result = await mcp_service._kill_process_on_port(9000)
            assert result is False


class TestAuthConfigChanges:
    """Test authentication configuration change detection."""

    def test_has_auth_config_changed_port_changed(self, mcp_service):
        """Test that port change is detected."""
        existing_auth = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
        }
        new_auth = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9001",
            "oauth_server_url": "http://localhost:9001",
        }

        assert mcp_service._has_auth_config_changed(existing_auth, new_auth) is True

    def test_has_auth_config_changed_no_change(self, mcp_service):
        """Test that identical configs are not detected as changed."""
        existing_auth = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
        }
        new_auth = existing_auth.copy()

        assert mcp_service._has_auth_config_changed(existing_auth, new_auth) is False

    def test_has_auth_config_changed_auth_type_changed(self, mcp_service):
        """Test that auth type change is detected."""
        existing_auth = {"auth_type": "oauth", "oauth_port": "9000"}
        new_auth = {"auth_type": "apikey", "api_key": "test_key"}

        assert mcp_service._has_auth_config_changed(existing_auth, new_auth) is True

    def test_has_auth_config_changed_both_none(self, mcp_service):
        """Test that two None configs are not detected as changed."""
        assert mcp_service._has_auth_config_changed(None, None) is False

    def test_has_auth_config_changed_one_none(self, mcp_service):
        """Test that changing from None to config is detected."""
        existing_auth = None
        new_auth = {"auth_type": "oauth", "oauth_port": "9000"}

        assert mcp_service._has_auth_config_changed(existing_auth, new_auth) is True


class TestPortChangeHandling:
    """Test handling of port changes in composer restart."""

    @pytest.mark.asyncio
    async def test_port_change_triggers_restart(self, mcp_service):
        """Test that changing ports triggers a restart via auth config change detection."""
        project_id = "test-project"
        old_port = 9000
        new_port = 9001

        # Set up existing composer
        mock_process = MagicMock(poll=MagicMock(return_value=None), pid=12345)
        mcp_service.project_composers[project_id] = {
            "process": mock_process,
            "host": "localhost",
            "port": old_port,
            "streamable_http_url": "http://test",
            "legacy_sse_url": "http://test/sse",
            "auth_config": {
                "auth_type": "oauth",
                "oauth_host": "localhost",
                "oauth_port": str(old_port),
                "oauth_server_url": f"http://localhost:{old_port}",
                "oauth_client_id": "test",
                "oauth_client_secret": "test",
                "oauth_auth_url": "http://test",
                "oauth_token_url": "http://test",
            },
        }
        mcp_service._port_to_project[old_port] = project_id
        mcp_service._pid_to_project[12345] = project_id

        new_auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": str(new_port),
            "oauth_server_url": f"http://localhost:{new_port}",
            "oauth_client_id": "test",
            "oauth_client_secret": "test",
            "oauth_auth_url": "http://test",
            "oauth_token_url": "http://test",
        }

        with (
            patch.object(mcp_service, "_do_stop_project_composer", new=AsyncMock()) as mock_stop,
            patch.object(mcp_service, "_is_port_available", return_value=True),
            patch.object(mcp_service, "_start_project_composer_process", new=AsyncMock()),
        ):
            # Initialize locks
            mcp_service._start_locks[project_id] = asyncio.Lock()

            with contextlib.suppress(Exception):
                await mcp_service._do_start_project_composer(
                    project_id=project_id,
                    streamable_http_url="http://test",
                    auth_config=new_auth_config,
                    max_retries=1,
                    max_startup_checks=1,
                    startup_delay=0.1,
                )

            # Verify composer was stopped (because config changed)
            mock_stop.assert_called_once_with(project_id)

    @pytest.mark.asyncio
    async def test_port_in_use_by_own_project_triggers_kill(self, mcp_service):
        """Test that when port is in use by the current project, it kills the process."""
        project_id = "test-project"
        test_port = 9001

        # Register the port as owned by this project
        mcp_service._port_to_project[test_port] = project_id

        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": str(test_port),
            "oauth_server_url": f"http://localhost:{test_port}",
            "oauth_client_id": "test",
            "oauth_client_secret": "test",
            "oauth_auth_url": "http://test",
            "oauth_token_url": "http://test",
        }

        with (
            patch.object(mcp_service, "_is_port_available") as mock_port_check,
            patch.object(mcp_service, "_kill_process_on_port", new=AsyncMock(return_value=True)) as mock_kill,
        ):
            # First check: port is in use, second check after kill: port is free
            mock_port_check.side_effect = [False, True]

            # Initialize locks
            mcp_service._start_locks[project_id] = asyncio.Lock()

            with (
                patch.object(mcp_service, "_start_project_composer_process", new=AsyncMock()),
                contextlib.suppress(Exception),
            ):
                await mcp_service._do_start_project_composer(
                    project_id=project_id,
                    streamable_http_url="http://test",
                    auth_config=auth_config,
                    max_retries=1,
                    max_startup_checks=1,
                    startup_delay=0.1,
                )

            # Verify kill was attempted on own project's port
            mock_kill.assert_called_with(test_port)

    @pytest.mark.asyncio
    async def test_port_in_use_by_unknown_process_raises_error(self, mcp_service):
        """Test that error is raised when port is in use by unknown process (security)."""
        project_id = "test-project"
        test_port = 9001

        # Port is NOT tracked (unknown process)
        # mcp_service._port_to_project does NOT contain test_port

        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": str(test_port),
            "oauth_server_url": f"http://localhost:{test_port}",
            "oauth_client_id": "test",
            "oauth_client_secret": "test",
            "oauth_auth_url": "http://test",
            "oauth_token_url": "http://test",
        }

        with patch.object(mcp_service, "_is_port_available", return_value=False):  # Port in use
            # Initialize locks
            mcp_service._start_locks[project_id] = asyncio.Lock()

            with pytest.raises(MCPComposerPortError) as exc_info:
                await mcp_service._do_start_project_composer(
                    project_id=project_id,
                    streamable_http_url="http://test",
                    auth_config=auth_config,
                    max_retries=1,
                    max_startup_checks=1,
                    startup_delay=0.1,
                )

            # New security message: won't kill unknown processes
            assert "already in use by another application" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_legacy_sse_url_preserved_in_composer_state(self, mcp_service):
        """Ensure legacy SSE URLs are passed through to the composer process and stored."""
        project_id = "legacy-project"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9100",
            "oauth_server_url": "http://localhost:9100",
            "oauth_client_id": "legacy",
            "oauth_client_secret": "secret",
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }
        legacy_url = "http://test/legacy-sse"
        streamable_url = "http://test/streamable"

        mcp_service._start_locks[project_id] = asyncio.Lock()

        mock_process = MagicMock(pid=4321)
        with (
            patch.object(mcp_service, "_is_port_available", return_value=True),
            patch.object(
                mcp_service,
                "_start_project_composer_process",
                new=AsyncMock(return_value=mock_process),
            ) as mock_start,
        ):
            await mcp_service._do_start_project_composer(
                project_id=project_id,
                streamable_http_url=streamable_url,
                auth_config=auth_config,
                max_retries=1,
                max_startup_checks=1,
                startup_delay=0.1,
                legacy_sse_url=legacy_url,
            )

        mock_start.assert_awaited()
        kwargs = mock_start.call_args.kwargs
        assert kwargs["legacy_sse_url"] == legacy_url
        assert mcp_service.project_composers[project_id]["legacy_sse_url"] == legacy_url
        assert mcp_service.project_composers[project_id]["streamable_http_url"] == streamable_url

    @pytest.mark.asyncio
    async def test_legacy_sse_url_defaults_when_not_provided(self, mcp_service):
        """Verify that a default SSE URL is derived when none is supplied."""
        project_id = "legacy-default"
        streamable_url = "http://test/default"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9200",
            "oauth_server_url": "http://localhost:9200",
            "oauth_client_id": "legacy",
            "oauth_client_secret": "secret",
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }

        mcp_service._start_locks[project_id] = asyncio.Lock()

        mock_process = MagicMock(pid=9876)
        with (
            patch.object(mcp_service, "_is_port_available", return_value=True),
            patch.object(
                mcp_service,
                "_start_project_composer_process",
                new=AsyncMock(return_value=mock_process),
            ) as mock_start,
        ):
            await mcp_service._do_start_project_composer(
                project_id=project_id,
                streamable_http_url=streamable_url,
                auth_config=auth_config,
                max_retries=1,
                max_startup_checks=1,
                startup_delay=0.1,
            )

        mock_start.assert_awaited()
        kwargs = mock_start.call_args.kwargs
        assert kwargs["legacy_sse_url"] == f"{streamable_url}/sse"


class TestOAuthCallbackConfigForwarding:
    """Test that OAuth callback config is forwarded to process startup unchanged."""

    @pytest.mark.asyncio
    async def test_callback_url_forwarded_to_start_process_auth_config(self, mcp_service):
        """Verify oauth_callback_url is preserved when passed to process startup.

        This test patches _start_project_composer_process, so it does not verify
        env-var mapping such as OAUTH_CALLBACK_PATH vs OAUTH_CALLBACK_URL. It
        only verifies that _do_start_project_composer forwards the callback URL
        in auth_config to the process-start method unchanged.
        """
        project_id = "callback-test"
        callback = "http://localhost:9000/auth/idaas/callback"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_callback_url": callback,
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
            "oauth_mcp_scope": "mcp",
            "oauth_provider_scope": "openid",
        }

        mcp_service._start_locks[project_id] = asyncio.Lock()

        mock_process = MagicMock(pid=1111)
        with (
            patch.object(mcp_service, "_is_port_available", return_value=True),
            patch.object(
                mcp_service,
                "_start_project_composer_process",
                new=AsyncMock(return_value=mock_process),
            ) as mock_start,
        ):
            await mcp_service._do_start_project_composer(
                project_id=project_id,
                streamable_http_url="http://test/mcp",
                auth_config=auth_config,
                max_retries=1,
                max_startup_checks=1,
                startup_delay=0.1,
            )

        mock_start.assert_awaited()
        # auth_config is the 5th positional arg (index 4)
        passed_auth = mock_start.call_args[0][4]

        # The auth_config should still carry the value under oauth_callback_url.
        assert passed_auth["oauth_callback_url"] == callback

    @pytest.mark.asyncio
    async def test_callback_url_backwards_compat(self, mcp_service):
        """Verify legacy oauth_callback_url is promoted and sent as OAUTH_CALLBACK_PATH."""
        project_id = "compat-test"
        callback_url = "http://localhost:9000/auth/callback"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_callback_url": callback_url,
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }

        mock_settings = MagicMock()
        mock_settings.settings.mcp_composer_version = "==0.1.0.8.10"

        mock_process = MagicMock(pid=2222, poll=MagicMock(return_value=None))

        with (
            patch("lfx.services.mcp_composer.service.get_settings_service", return_value=mock_settings),
            patch("subprocess.Popen", return_value=mock_process) as mock_popen,
            patch.object(mcp_service, "_is_port_available", return_value=False),
        ):
            await mcp_service._start_project_composer_process(
                project_id=project_id,
                host="localhost",
                port=9000,
                streamable_http_url="http://test/mcp",
                auth_config=auth_config,
                max_startup_checks=1,
                startup_delay=0.01,
            )

        cmd = mock_popen.call_args[0][0]

        # Legacy oauth_callback_url should be promoted and sent as OAUTH_CALLBACK_PATH
        assert "OAUTH_CALLBACK_PATH" in cmd, f"Expected OAUTH_CALLBACK_PATH in command but got: {cmd}"
        idx = cmd.index("OAUTH_CALLBACK_PATH")
        assert cmd[idx + 1] == callback_url

    @pytest.mark.asyncio
    async def test_callback_url_wins_when_both_callback_fields_are_present(self, mcp_service):
        """Verify canonical oauth_callback_url takes precedence over deprecated oauth_callback_path."""
        project_id = "canonical-callback-test"
        callback_url = "http://localhost:9000/auth/idaas/callback"
        stale_callback_path = "http://localhost:9999/stale/callback"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_callback_url": callback_url,
            "oauth_callback_path": stale_callback_path,
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }

        mock_settings = MagicMock()
        mock_settings.settings.mcp_composer_version = "==0.1.0.8.10"

        mock_process = MagicMock(pid=4444, poll=MagicMock(return_value=None))

        with (
            patch("lfx.services.mcp_composer.service.get_settings_service", return_value=mock_settings),
            patch("subprocess.Popen", return_value=mock_process) as mock_popen,
            patch.object(mcp_service, "_is_port_available", return_value=False),
        ):
            await mcp_service._start_project_composer_process(
                project_id=project_id,
                host="localhost",
                port=9000,
                streamable_http_url="http://test/mcp",
                auth_config=auth_config,
                max_startup_checks=1,
                startup_delay=0.01,
            )

        cmd = mock_popen.call_args[0][0]
        idx = cmd.index("OAUTH_CALLBACK_PATH")
        assert cmd[idx + 1] == callback_url

    @pytest.mark.asyncio
    async def test_subprocess_receives_callback_path_env_var(self, mcp_service):
        """Verify the subprocess command uses OAUTH_CALLBACK_PATH, not OAUTH_CALLBACK_URL.

        This is the core regression test: mcp-composer's ServerSettings has a
        'callback_path' field (env_prefix='OAUTH_'), so the env var must be
        OAUTH_CALLBACK_PATH.  Sending OAUTH_CALLBACK_URL causes a Pydantic
        'extra_forbidden' validation error.
        """
        project_id = "env-var-test"
        callback = "http://localhost:9000/auth/idaas/callback"
        auth_config = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_callback_url": callback,
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
            "oauth_mcp_scope": "mcp",
            "oauth_provider_scope": "openid",
        }

        mock_settings = MagicMock()
        mock_settings.settings.mcp_composer_version = "==0.1.0.8.10"

        mock_process = MagicMock(pid=3333, poll=MagicMock(return_value=None))

        with (
            patch("lfx.services.mcp_composer.service.get_settings_service", return_value=mock_settings),
            patch("subprocess.Popen", return_value=mock_process) as mock_popen,
            patch.object(mcp_service, "_is_port_available", return_value=False),
        ):
            await mcp_service._start_project_composer_process(
                project_id=project_id,
                host="localhost",
                port=9000,
                streamable_http_url="http://test/mcp",
                auth_config=auth_config,
                max_startup_checks=1,
                startup_delay=0.01,
            )

        cmd = mock_popen.call_args[0][0]

        # OAUTH_CALLBACK_PATH must appear in the command
        assert "OAUTH_CALLBACK_PATH" in cmd, f"Expected OAUTH_CALLBACK_PATH in command but got: {cmd}"
        # The callback value should follow the env key
        idx = cmd.index("OAUTH_CALLBACK_PATH")
        assert cmd[idx + 1] == callback

        # OAUTH_CALLBACK_URL must NOT appear (would cause extra_forbidden error)
        assert "OAUTH_CALLBACK_URL" not in cmd, (
            f"OAUTH_CALLBACK_URL should not be in command (mcp-composer rejects it): {cmd}"
        )

    def test_callback_aliases_do_not_trigger_restart_when_effective_value_matches(self, mcp_service):
        """Alias-only callback field changes should not force composer restarts."""
        existing_auth = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_callback_path": "http://localhost:9000/auth/idaas/callback",
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }
        new_auth = {
            "auth_type": "oauth",
            "oauth_host": "localhost",
            "oauth_port": "9000",
            "oauth_server_url": "http://localhost:9000",
            "oauth_callback_url": "http://localhost:9000/auth/idaas/callback",
            "oauth_client_id": "cid",
            "oauth_client_secret": "csecret",  # pragma: allowlist secret
            "oauth_auth_url": "http://auth",
            "oauth_token_url": "http://token",
        }

        assert mcp_service._has_auth_config_changed(existing_auth, new_auth) is False