# -*- coding: utf-8 -*-
"""Location: ./tests/unit/mcpgateway/test_translate_grpc.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: ContextForge Contributors

Tests for gRPC to MCP translation module.
"""

# Standard
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

# Third-Party
import pytest

# Check if gRPC is available
try:
    # Third-Party
    import grpc  # noqa: F401

    GRPC_AVAILABLE = True
except ImportError:
    GRPC_AVAILABLE = False

# First-Party
from mcpgateway.translate_grpc import (
    expose_grpc_via_sse,
    GrpcEndpoint,
    GrpcToMcpTranslator,
)


@pytest.mark.skipif(not GRPC_AVAILABLE, reason="gRPC packages not installed")
class TestGrpcEndpoint:
    """Test suite for GrpcEndpoint."""

    @pytest.fixture
    def endpoint(self):
        """Create a basic gRPC endpoint."""
        return GrpcEndpoint(
            target="localhost:50051",
            reflection_enabled=True,
            tls_enabled=False,
        )

    @pytest.fixture
    def endpoint_with_tls(self):
        """Create a gRPC endpoint with TLS."""
        return GrpcEndpoint(
            target="secure.example.com:443",
            reflection_enabled=True,
            tls_enabled=True,
            tls_cert_path="/path/to/cert.pem",
            tls_key_path="/path/to/key.pem",
        )

    @pytest.fixture
    def endpoint_with_metadata(self):
        """Create a gRPC endpoint with metadata."""
        return GrpcEndpoint(
            target="api.example.com:50051",
            reflection_enabled=True,
            metadata={"authorization": "Bearer test-token", "x-tenant-id": "customer-1"},
        )

    def test_endpoint_initialization(self, endpoint):
        """Test basic endpoint initialization."""
        assert endpoint._target == "localhost:50051"
        assert endpoint._reflection_enabled is True
        assert endpoint._tls_enabled is False
        assert endpoint._channel is None
        assert len(endpoint._services) == 0

    def test_endpoint_with_tls_initialization(self, endpoint_with_tls):
        """Test endpoint with TLS configuration."""
        assert endpoint_with_tls._tls_enabled is True
        assert endpoint_with_tls._tls_cert_path == "/path/to/cert.pem"
        assert endpoint_with_tls._tls_key_path == "/path/to/key.pem"

    def test_endpoint_with_metadata_initialization(self, endpoint_with_metadata):
        """Test endpoint with metadata headers."""
        assert endpoint_with_metadata._metadata == {
            "authorization": "Bearer test-token",
            "x-tenant-id": "customer-1",
        }

    @patch("mcpgateway.translate_grpc.grpc")
    async def test_start_insecure_channel(self, mock_grpc, endpoint):
        """Test starting endpoint with insecure channel."""
        mock_channel = MagicMock()
        mock_grpc.insecure_channel.return_value = mock_channel

        with patch.object(endpoint, "_discover_services", new_callable=AsyncMock):
            await endpoint.start()

        mock_grpc.insecure_channel.assert_called_once_with("localhost:50051")
        assert endpoint._channel == mock_channel

    @patch("mcpgateway.translate_grpc.grpc")
    async def test_start_secure_channel_with_certs(self, mock_grpc, endpoint_with_tls):
        """Test starting endpoint with TLS certificates."""
        mock_channel = MagicMock()
        mock_grpc.secure_channel.return_value = mock_channel
        mock_grpc.ssl_channel_credentials.return_value = MagicMock()

        with patch("mcpgateway.translate_grpc.asyncio.to_thread", new_callable=AsyncMock, return_value=b"cert_data"):
            with patch.object(endpoint_with_tls, "_discover_services", new_callable=AsyncMock):
                await endpoint_with_tls.start()

        assert endpoint_with_tls._channel == mock_channel
        mock_grpc.secure_channel.assert_called_once()

    @patch("mcpgateway.translate_grpc.grpc")
    async def test_start_secure_channel_without_certs(self, mock_grpc):
        """Test starting endpoint with TLS but no cert files."""
        endpoint = GrpcEndpoint(
            target="secure.example.com:443",
            reflection_enabled=True,
            tls_enabled=True,
        )

        mock_channel = MagicMock()
        mock_grpc.secure_channel.return_value = mock_channel
        mock_grpc.ssl_channel_credentials.return_value = MagicMock()

        with patch.object(endpoint, "_discover_services", new_callable=AsyncMock):
            await endpoint.start()

        mock_grpc.ssl_channel_credentials.assert_called_once_with()
        assert endpoint._channel == mock_channel

    @patch("mcpgateway.translate_grpc.grpc")
    @patch("mcpgateway.translate_grpc.reflection_pb2_grpc")
    @patch("mcpgateway.translate_grpc.reflection_pb2")
    async def test_discover_services_success(self, mock_reflection_pb2, mock_reflection_grpc, mock_grpc, endpoint):
        """Test successful service discovery."""
        # Setup mocks
        mock_channel = MagicMock()
        endpoint._channel = mock_channel

        mock_stub = MagicMock()
        mock_reflection_grpc.ServerReflectionStub.return_value = mock_stub

        # Mock service discovery response
        mock_service = MagicMock()
        mock_service.name = "test.TestService"

        mock_list_response = MagicMock()
        mock_list_response.service = [mock_service]

        mock_response = MagicMock()
        mock_response.HasField.return_value = True
        mock_response.list_services_response = mock_list_response

        mock_stub.ServerReflectionInfo.return_value = [mock_response]

        # Mock _discover_service_details to populate services
        with patch.object(endpoint, "_discover_service_details", new_callable=AsyncMock) as mock_details:

            async def populate_service(stub, service_name, timeout=None):
                endpoint._services[service_name] = {
                    "name": service_name,
                    "methods": [],
                }

            mock_details.side_effect = populate_service

            await endpoint._discover_services()

        assert "test.TestService" in endpoint._services
        assert endpoint._services["test.TestService"]["name"] == "test.TestService"

    @patch("mcpgateway.translate_grpc.grpc")
    @patch("mcpgateway.translate_grpc.reflection_pb2_grpc")
    async def test_discover_services_skip_reflection_service(self, mock_reflection_grpc, mock_grpc, endpoint):
        """Test that ServerReflection service is skipped."""
        mock_channel = MagicMock()
        endpoint._channel = mock_channel

        mock_stub = MagicMock()
        mock_reflection_grpc.ServerReflectionStub.return_value = mock_stub

        # Mock response with ServerReflection service (should be skipped)
        mock_service1 = MagicMock()
        mock_service1.name = "grpc.reflection.v1alpha.ServerReflection"

        mock_service2 = MagicMock()
        mock_service2.name = "test.TestService"

        mock_list_response = MagicMock()
        mock_list_response.service = [mock_service1, mock_service2]

        mock_response = MagicMock()
        mock_response.HasField.return_value = True
        mock_response.list_services_response = mock_list_response

        mock_stub.ServerReflectionInfo.return_value = [mock_response]

        # Mock _discover_service_details to populate only non-reflection services
        with patch.object(endpoint, "_discover_service_details", new_callable=AsyncMock) as mock_details:

            async def populate_service(stub, service_name, timeout=None):
                endpoint._services[service_name] = {
                    "name": service_name,
                    "methods": [],
                }

            mock_details.side_effect = populate_service

            await endpoint._discover_services()

        # ServerReflection should be skipped
        assert "grpc.reflection.v1alpha.ServerReflection" not in endpoint._services
        # TestService should be included
        assert "test.TestService" in endpoint._services

    @patch("mcpgateway.translate_grpc.grpc")
    @patch("mcpgateway.translate_grpc.reflection_pb2_grpc")
    async def test_discover_services_error(self, mock_reflection_grpc, mock_grpc, endpoint):
        """Test service discovery error handling."""
        mock_channel = MagicMock()
        endpoint._channel = mock_channel

        mock_stub = MagicMock()
        mock_reflection_grpc.ServerReflectionStub.return_value = mock_stub
        mock_stub.ServerReflectionInfo.side_effect = Exception("Connection failed")

        with pytest.raises(Exception) as exc_info:
            await endpoint._discover_services()

        assert "Connection failed" in str(exc_info.value)

    async def test_invoke_service_not_found(self, endpoint):
        """Test invoke with non-existent service."""
        with pytest.raises(ValueError, match="Service .* not found"):
            await endpoint.invoke(
                service="test.TestService",
                method="TestMethod",
                request_data={"param": "value"},
            )

    async def test_invoke_streaming_service_not_found(self, endpoint):
        """Test invoke_streaming with non-existent service."""
        with pytest.raises(ValueError, match="Service .* not found"):
            async for _ in endpoint.invoke_streaming(
                service="test.TestService",
                method="StreamMethod",
                request_data={"param": "value"},
            ):
                pass

    async def test_close(self, endpoint):
        """Test closing the gRPC channel."""
        mock_channel = MagicMock()
        endpoint._channel = mock_channel

        await endpoint.close()

        mock_channel.close.assert_called_once()

    async def test_close_no_channel(self, endpoint):
        """Test closing when no channel exists."""
        # Should not raise an error
        await endpoint.close()

    def test_get_services(self, endpoint):
        """Test getting list of discovered services."""
        endpoint._services = {
            "service1": {"name": "service1"},
            "service2": {"name": "service2"},
        }

        services = endpoint.get_services()

        assert len(services) == 2
        assert "service1" in services
        assert "service2" in services

    def test_get_methods(self, endpoint):
        """Test getting methods for a service."""
        endpoint._services = {
            "test.TestService": {
                "name": "test.TestService",
                "methods": [{"name": "Method1"}, {"name": "Method2"}],
            }
        }

        methods = endpoint.get_methods("test.TestService")

        assert len(methods) == 2
        assert "Method1" in methods
        assert "Method2" in methods

    def test_get_methods_nonexistent_service(self, endpoint):
        """Test getting methods for non-existent service."""
        methods = endpoint.get_methods("nonexistent.Service")

        assert len(methods) == 0


@pytest.mark.skipif(not GRPC_AVAILABLE, reason="gRPC packages not installed")
class TestGrpcToMcpTranslator:
    """Test suite for GrpcToMcpTranslator."""

    @pytest.fixture
    def endpoint(self):
        """Create a mock gRPC endpoint."""
        endpoint = MagicMock(spec=GrpcEndpoint)
        endpoint.get_methods.return_value = ["Method1", "Method2"]
        endpoint._services = {
            "test.TestService": {
                "name": "test.TestService",
                "methods": [
                    {"name": "Method1", "input_type": ".test.Request1", "output_type": ".test.Response1"},
                    {"name": "Method2", "input_type": ".test.Request2", "output_type": ".test.Response2"},
                ],
            }
        }
        endpoint._pool = MagicMock()
        endpoint._pool.FindMessageTypeByName.side_effect = KeyError("Not found")
        return endpoint

    @pytest.fixture
    def translator(self, endpoint):
        """Create a translator instance."""
        return GrpcToMcpTranslator(endpoint)

    def test_translator_initialization(self, translator, endpoint):
        """Test translator initialization."""
        assert translator._endpoint == endpoint

    def test_grpc_service_to_mcp_server(self, translator, endpoint):
        """Test converting gRPC service to MCP server definition."""
        result = translator.grpc_service_to_mcp_server("test.TestService")

        assert result["name"] == "test.TestService"
        assert result["description"] == "gRPC service: test.TestService"
        assert "sse" in result["transport"]
        assert "http" in result["transport"]
        assert "tools" in result

    def test_grpc_methods_to_mcp_tools(self, translator, endpoint):
        """Test converting gRPC methods to MCP tools."""
        result = translator.grpc_methods_to_mcp_tools("test.TestService")

        assert len(result) == 2
        assert result[0]["name"] == "test.TestService.Method1"
        assert result[0]["description"] == "gRPC method test.TestService.Method1"
        assert "inputSchema" in result[0]

    def test_protobuf_to_json_schema(self, translator):
        """Test converting protobuf descriptor to JSON schema."""
        mock_descriptor = MagicMock()
        mock_descriptor.fields = []  # Empty message

        result = translator.protobuf_to_json_schema(mock_descriptor)

        assert result["type"] == "object"
        assert "properties" in result
        assert "required" in result


@pytest.mark.skipif(not GRPC_AVAILABLE, reason="gRPC packages not installed")
class TestExposeGrpcViaSse:
    """Test suite for expose_grpc_via_sse utility function."""

    @patch("mcpgateway.translate_grpc.GrpcEndpoint")
    @patch("mcpgateway.translate_grpc.asyncio.sleep")
    async def test_expose_grpc_via_sse_basic(self, mock_sleep, mock_endpoint_class):
        """Test basic gRPC exposure via SSE."""
        # Mock the endpoint
        mock_endpoint = MagicMock()
        mock_endpoint.start = AsyncMock()
        mock_endpoint.close = AsyncMock()
        mock_endpoint.get_services.return_value = ["test.TestService"]
        mock_endpoint_class.return_value = mock_endpoint

        # Mock sleep to raise KeyboardInterrupt after first call
        mock_sleep.side_effect = KeyboardInterrupt()

        try:
            await expose_grpc_via_sse(target="localhost:50051", port=9000)
        except KeyboardInterrupt:
            pass

        mock_endpoint.start.assert_called_once()
        mock_endpoint.close.assert_called_once()

    @patch("mcpgateway.translate_grpc.GrpcEndpoint")
    @patch("mcpgateway.translate_grpc.asyncio.sleep")
    async def test_expose_grpc_via_sse_with_tls(self, mock_sleep, mock_endpoint_class):
        """Test gRPC exposure with TLS configuration."""
        mock_endpoint = MagicMock()
        mock_endpoint.start = AsyncMock()
        mock_endpoint.close = AsyncMock()
        mock_endpoint.get_services.return_value = []
        mock_endpoint_class.return_value = mock_endpoint

        mock_sleep.side_effect = KeyboardInterrupt()

        try:
            await expose_grpc_via_sse(
                target="secure.example.com:443",
                port=9000,
                tls_enabled=True,
                tls_cert="/path/to/cert.pem",
                tls_key="/path/to/key.pem",
            )
        except KeyboardInterrupt:
            pass

        # Verify endpoint was created with TLS config
        mock_endpoint_class.assert_called_once_with(
            target="secure.example.com:443",
            reflection_enabled=True,
            tls_enabled=True,
            tls_cert_path="/path/to/cert.pem",
            tls_key_path="/path/to/key.pem",
            metadata=None,
        )

    @patch("mcpgateway.translate_grpc.GrpcEndpoint")
    @patch("mcpgateway.translate_grpc.asyncio.sleep")
    async def test_expose_grpc_via_sse_with_metadata(self, mock_sleep, mock_endpoint_class):
        """Test gRPC exposure with metadata headers."""
        mock_endpoint = MagicMock()
        mock_endpoint.start = AsyncMock()
        mock_endpoint.close = AsyncMock()
        mock_endpoint.get_services.return_value = []
        mock_endpoint_class.return_value = mock_endpoint

        mock_sleep.side_effect = KeyboardInterrupt()

        metadata = {"authorization": "Bearer token", "x-tenant": "test"}

        try:
            await expose_grpc_via_sse(
                target="api.example.com:50051",
                port=9000,
                metadata=metadata,
            )
        except KeyboardInterrupt:
            pass

        # Verify metadata was passed
        call_args = mock_endpoint_class.call_args
        assert call_args[1]["metadata"] == metadata


@pytest.mark.asyncio
async def test_invoke_and_invoke_streaming_without_grpc(monkeypatch):
    # Standard
    from types import SimpleNamespace

    # First-Party
    import mcpgateway.translate_grpc as tg

    class DummyRequest:
        def SerializeToString(self):
            return b"req"

    class DummyResponse:
        @staticmethod
        def FromString(_data):
            return DummyResponse()

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {
        "TestService": {
            "methods": [
                {"name": "Unary", "input_type": ".Input", "output_type": ".Output", "client_streaming": False, "server_streaming": False},
                {"name": "Stream", "input_type": ".Input", "output_type": ".Output", "client_streaming": False, "server_streaming": True},
            ]
        }
    }
    endpoint._pool = MagicMock()
    endpoint._pool.FindMessageTypeByName.side_effect = [object(), object(), object(), object()]
    endpoint._factory = MagicMock()

    # protobuf>=5 uses module-level message_factory.GetMessageClass(); patch it instead of MessageFactory.GetPrototype.
    proto_classes = iter([DummyRequest, DummyResponse, DummyRequest, DummyResponse])
    monkeypatch.setattr(tg.message_factory, "GetMessageClass", lambda _desc: next(proto_classes))

    class DummyChannel:
        def unary_unary(self, _path, request_serializer=None, response_deserializer=None):
            def call(_req, timeout=None):
                return DummyResponse()

            return call

        def unary_stream(self, _path, request_serializer=None, response_deserializer=None):
            def call(_req):
                return [DummyResponse(), DummyResponse()]

            return call

    endpoint._channel = DummyChannel()

    monkeypatch.setattr(
        tg,
        "json_format",
        SimpleNamespace(
            ParseDict=lambda _data, _cls: DummyRequest(),
            MessageToDict=lambda _msg, **_kwargs: {"ok": True},
        ),
    )

    result = await endpoint.invoke("TestService", "Unary", {"a": 1})
    assert result == {"ok": True}

    chunks = []
    async for item in endpoint.invoke_streaming("TestService", "Stream", {"a": 1}):
        chunks.append(item)
    assert chunks == [{"ok": True}, {"ok": True}]


@pytest.mark.asyncio
async def test_endpoint_start_without_reflection(monkeypatch):
    monkeypatch.setattr("mcpgateway.translate_grpc.descriptor_pool", SimpleNamespace(DescriptorPool=lambda **_kw: MagicMock()))
    monkeypatch.setattr("mcpgateway.translate_grpc.message_factory", SimpleNamespace(MessageFactory=lambda *_a, **_kw: MagicMock()))
    endpoint = GrpcEndpoint(target="localhost:50051", reflection_enabled=False, tls_enabled=False)
    mock_grpc = MagicMock()
    mock_grpc.insecure_channel.return_value = "chan"
    monkeypatch.setattr("mcpgateway.translate_grpc.grpc", mock_grpc)
    monkeypatch.setattr(endpoint, "_discover_services", AsyncMock())

    await endpoint.start()
    assert endpoint._channel == "chan"
    endpoint._discover_services.assert_not_called()


@pytest.mark.asyncio
async def test_endpoint_start_with_tls_and_reflection(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    monkeypatch.setattr(tg, "descriptor_pool", SimpleNamespace(DescriptorPool=lambda **_kw: MagicMock()))
    monkeypatch.setattr(tg, "message_factory", SimpleNamespace(MessageFactory=lambda *_a, **_kw: MagicMock()))
    endpoint = tg.GrpcEndpoint(
        target="secure.example.com:443",
        reflection_enabled=True,
        tls_enabled=True,
        tls_cert_path="/tmp/cert.pem",
        tls_key_path="/tmp/key.pem",
    )

    mock_grpc = MagicMock()
    mock_grpc.ssl_channel_credentials.return_value = "creds"
    mock_grpc.secure_channel.return_value = "secure-chan"
    monkeypatch.setattr(tg, "grpc", mock_grpc)
    monkeypatch.setattr(tg.asyncio, "to_thread", AsyncMock(return_value=b"cert-data"))
    monkeypatch.setattr(endpoint, "_discover_services", AsyncMock())

    await endpoint.start()

    assert endpoint._channel == "secure-chan"
    mock_grpc.ssl_channel_credentials.assert_called_once()
    endpoint._discover_services.assert_awaited()


@pytest.mark.asyncio
async def test_discover_services_success_no_grpc(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._target = "localhost:50051"
    endpoint._channel = "chan"
    endpoint._services = {}

    mock_stub = MagicMock()

    mock_service = MagicMock()
    mock_service.name = "test.TestService"

    mock_list_response = MagicMock()
    mock_list_response.service = [mock_service]

    mock_response = MagicMock()
    mock_response.HasField.return_value = True
    mock_response.list_services_response = mock_list_response

    mock_stub.ServerReflectionInfo.return_value = [mock_response]

    monkeypatch.setattr(tg, "reflection_pb2_grpc", SimpleNamespace(ServerReflectionStub=lambda _chan: mock_stub))
    monkeypatch.setattr(tg, "reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    async def _populate(_stub, service_name, timeout=None):
        endpoint._services[service_name] = {"name": service_name, "methods": []}

    monkeypatch.setattr(endpoint, "_discover_service_details", AsyncMock(side_effect=_populate))

    await tg.GrpcEndpoint._discover_services(endpoint)

    assert "test.TestService" in endpoint._services


@pytest.mark.asyncio
async def test_discover_service_details_error_fallback(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {}
    endpoint._descriptors = {}
    endpoint._pool = MagicMock()

    class DummyStub:
        def ServerReflectionInfo(self, _request_iter):
            raise RuntimeError("boom")

    monkeypatch.setattr(tg, "reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    await tg.GrpcEndpoint._discover_service_details(endpoint, DummyStub(), "pkg.TestService")

    assert endpoint._services["pkg.TestService"]["methods"] == []


@pytest.mark.asyncio
async def test_invoke_validation_errors(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {"svc": {"methods": [{"name": "Stream", "client_streaming": True, "server_streaming": True}]}}

    with pytest.raises(ValueError, match="Service .* not found"):
        await endpoint.invoke("missing", "Ping", {})

    with pytest.raises(ValueError, match="Method .* not found"):
        await endpoint.invoke("svc", "Ping", {})

    with pytest.raises(ValueError, match="is streaming"):
        await endpoint.invoke("svc", "Stream", {})


@pytest.mark.asyncio
async def test_invoke_message_type_missing(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {"svc": {"methods": [{"name": "Ping", "input_type": ".Input", "output_type": ".Output", "client_streaming": False, "server_streaming": False}]}}
    endpoint._pool = MagicMock()
    endpoint._pool.FindMessageTypeByName.side_effect = KeyError("missing")

    with pytest.raises(ValueError, match="Message type not found"):
        await endpoint.invoke("svc", "Ping", {})


@pytest.mark.asyncio
async def test_invoke_streaming_validation_errors(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {"svc": {"methods": [{"name": "Ping", "server_streaming": False, "client_streaming": False, "input_type": ".Input", "output_type": ".Output"}]}}

    with pytest.raises(ValueError, match="Service .* not found"):
        async for _ in endpoint.invoke_streaming("missing", "Ping", {}):
            pass

    with pytest.raises(ValueError, match="Method .* not found"):
        async for _ in endpoint.invoke_streaming("svc", "Missing", {}):
            pass

    with pytest.raises(ValueError, match="not server-streaming"):
        async for _ in endpoint.invoke_streaming("svc", "Ping", {}):
            pass

    endpoint._services["svc"]["methods"][0]["server_streaming"] = True
    endpoint._services["svc"]["methods"][0]["client_streaming"] = True

    with pytest.raises(ValueError, match="Client streaming"):
        async for _ in endpoint.invoke_streaming("svc", "Ping", {}):
            pass


@pytest.mark.asyncio
async def test_invoke_streaming_message_type_missing(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {"svc": {"methods": [{"name": "Stream", "server_streaming": True, "client_streaming": False, "input_type": ".Input", "output_type": ".Output"}]}}
    endpoint._pool = MagicMock()
    endpoint._pool.FindMessageTypeByName.side_effect = KeyError("missing")

    with pytest.raises(ValueError, match="Message type not found"):
        async for _ in endpoint.invoke_streaming("svc", "Stream", {}):
            pass


@pytest.mark.asyncio
async def test_invoke_streaming_rpc_error(monkeypatch):
    # First-Party
    import mcpgateway.translate_grpc as tg

    class DummyRequest:
        def SerializeToString(self):
            return b"req"

    class DummyResponse:
        @staticmethod
        def FromString(_data):
            return DummyResponse()

    class DummyRpcError(Exception):
        pass

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {"svc": {"methods": [{"name": "Stream", "server_streaming": True, "client_streaming": False, "input_type": ".Input", "output_type": ".Output"}]}}
    endpoint._pool = MagicMock()
    endpoint._pool.FindMessageTypeByName.side_effect = [object(), object()]
    endpoint._factory = MagicMock()
    proto_classes = iter([DummyRequest, DummyResponse])
    monkeypatch.setattr(tg.message_factory, "GetMessageClass", lambda _desc: next(proto_classes))

    class DummyChannel:
        def unary_stream(self, _path, request_serializer=None, response_deserializer=None):
            def call(_req):
                class _Stream:
                    def __iter__(self_inner):
                        raise DummyRpcError("boom")

                return _Stream()

            return call

    endpoint._channel = DummyChannel()

    monkeypatch.setattr(
        tg,
        "json_format",
        SimpleNamespace(
            ParseDict=lambda _data, _cls: DummyRequest(),
            MessageToDict=lambda _msg, **_kwargs: {"ok": True},
        ),
    )
    monkeypatch.setattr(tg, "grpc", SimpleNamespace(RpcError=DummyRpcError))

    with pytest.raises(DummyRpcError):
        async for _ in endpoint.invoke_streaming("svc", "Stream", {"a": 1}):
            pass


@pytest.mark.asyncio
async def test_close_with_channel_unskipped():
    endpoint = GrpcEndpoint.__new__(GrpcEndpoint)
    endpoint._channel = MagicMock()
    endpoint._target = "localhost:50051"

    await GrpcEndpoint.close(endpoint)

    endpoint._channel.close.assert_called_once()


def test_get_services_and_methods_unskipped():
    endpoint = GrpcEndpoint.__new__(GrpcEndpoint)
    endpoint._services = {
        "svc": {"methods": [{"name": "Ping"}]},
    }
    assert GrpcEndpoint.get_services(endpoint) == ["svc"]
    assert GrpcEndpoint.get_methods(endpoint, "svc") == ["Ping"]
    assert GrpcEndpoint.get_methods(endpoint, "missing") == []


def test_grpc_service_to_mcp_server_unskipped():
    endpoint = SimpleNamespace(_services={"svc": {"methods": []}}, _pool=MagicMock())
    translator = GrpcToMcpTranslator(endpoint=endpoint)
    result = translator.grpc_service_to_mcp_server("svc")
    assert result["name"] == "svc"
    assert "sse" in result["transport"]


@pytest.mark.asyncio
@patch("mcpgateway.translate_grpc.GrpcEndpoint")
@patch("mcpgateway.translate_grpc.asyncio.sleep")
async def test_expose_grpc_via_sse_keyboard_interrupt_unskipped(mock_sleep, mock_endpoint_class):
    mock_endpoint = MagicMock()
    mock_endpoint.start = AsyncMock()
    mock_endpoint.close = AsyncMock()
    mock_endpoint.get_services.return_value = ["svc"]
    mock_endpoint_class.return_value = mock_endpoint

    mock_sleep.side_effect = KeyboardInterrupt()

    try:
        await expose_grpc_via_sse(target="localhost:50051", port=9000)
    except KeyboardInterrupt:
        pass

    mock_endpoint.start.assert_called_once()
    mock_endpoint.close.assert_called_once()


@pytest.mark.asyncio
async def test_discover_service_details_success(monkeypatch):
    endpoint = GrpcEndpoint.__new__(GrpcEndpoint)
    endpoint._services = {}
    endpoint._descriptors = {}
    endpoint._pool = MagicMock()

    class DummyMethod:
        def __init__(self, name, input_type, output_type):
            self.name = name
            self.input_type = input_type
            self.output_type = output_type
            self.client_streaming = False
            self.server_streaming = False

    class DummyService:
        def __init__(self, name):
            self.name = name
            self.method = [DummyMethod("Ping", ".Input", ".Output")]

    class DummyFileDesc:
        def __init__(self):
            self.package = "pkg"
            self.service = [DummyService("TestService")]

        def ParseFromString(self, _data):
            return None

    class DummyResponse:
        def __init__(self):
            self.file_descriptor_response = SimpleNamespace(file_descriptor_proto=[b"blob"])

        def HasField(self, name):
            return name == "file_descriptor_response"

    class DummyStub:
        def ServerReflectionInfo(self, _request_iter):
            return [DummyResponse()]

    monkeypatch.setattr("mcpgateway.translate_grpc.FileDescriptorProto", DummyFileDesc)
    monkeypatch.setattr("mcpgateway.translate_grpc.reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    await GrpcEndpoint._discover_service_details(endpoint, DummyStub(), "pkg.TestService")
    assert "pkg.TestService" in endpoint._services
    assert endpoint._services["pkg.TestService"]["methods"][0]["name"] == "Ping"


@pytest.mark.asyncio
async def test_discover_services_ignores_non_list_services_response(monkeypatch):
    """HasField(False) responses should be ignored without attempting detail discovery."""
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._target = "localhost:50051"
    endpoint._channel = "chan"
    endpoint._services = {}

    mock_stub = MagicMock()
    mock_response = MagicMock()
    mock_response.HasField.return_value = False
    mock_stub.ServerReflectionInfo.return_value = [mock_response]

    monkeypatch.setattr(tg, "reflection_pb2_grpc", SimpleNamespace(ServerReflectionStub=lambda _chan: mock_stub))
    monkeypatch.setattr(tg, "reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    endpoint._discover_service_details = AsyncMock()
    await tg.GrpcEndpoint._discover_services(endpoint)
    endpoint._discover_service_details.assert_not_called()
    assert endpoint._services == {}


@pytest.mark.asyncio
async def test_discover_service_details_ignores_non_descriptor_response(monkeypatch):
    """HasField(False) responses should not mutate the services map."""
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {}
    endpoint._descriptors = {}
    endpoint._pool = MagicMock()

    class DummyResponse:
        def HasField(self, _name):
            return False

    class DummyStub:
        def ServerReflectionInfo(self, _request_iter):
            return [DummyResponse()]

    monkeypatch.setattr(tg, "reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    await tg.GrpcEndpoint._discover_service_details(endpoint, DummyStub(), "pkg.TestService")
    assert endpoint._services == {}


@pytest.mark.asyncio
async def test_discover_service_details_pool_add_error_is_swallowed(monkeypatch):
    """pool.Add exceptions are swallowed as 'already in pool'."""
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {}
    endpoint._descriptors = {}
    endpoint._pool = MagicMock()
    endpoint._pool.Add.side_effect = Exception("duplicate")

    class DummyMethod:
        def __init__(self, name, input_type, output_type):
            self.name = name
            self.input_type = input_type
            self.output_type = output_type
            self.client_streaming = False
            self.server_streaming = False

    class DummyService:
        def __init__(self, name):
            self.name = name
            self.method = [DummyMethod("Ping", ".Input", ".Output")]

    class DummyFileDesc:
        def __init__(self):
            self.package = "pkg"
            self.service = [DummyService("TestService")]

        def ParseFromString(self, _data):
            return None

    class DummyResponse:
        def __init__(self):
            self.file_descriptor_response = SimpleNamespace(file_descriptor_proto=[b"blob"])

        def HasField(self, name):
            return name == "file_descriptor_response"

    class DummyStub:
        def ServerReflectionInfo(self, _request_iter):
            return [DummyResponse()]

    monkeypatch.setattr("mcpgateway.translate_grpc.FileDescriptorProto", DummyFileDesc)
    monkeypatch.setattr("mcpgateway.translate_grpc.reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    await tg.GrpcEndpoint._discover_service_details(endpoint, DummyStub(), "pkg.TestService")
    assert "pkg.TestService" in endpoint._services


@pytest.mark.asyncio
async def test_discover_service_details_skips_unrelated_service(monkeypatch):
    """Services that don't match the requested name are ignored."""
    # First-Party
    import mcpgateway.translate_grpc as tg

    endpoint = tg.GrpcEndpoint.__new__(tg.GrpcEndpoint)
    endpoint._services = {}
    endpoint._descriptors = {}
    endpoint._pool = MagicMock()

    class DummyMethod:
        def __init__(self, name, input_type, output_type):
            self.name = name
            self.input_type = input_type
            self.output_type = output_type
            self.client_streaming = False
            self.server_streaming = False

    class DummyService:
        def __init__(self, name):
            self.name = name
            self.method = [DummyMethod("Ping", ".Input", ".Output")]

    class DummyFileDesc:
        def __init__(self):
            self.package = "pkg"
            self.service = [DummyService("OtherService")]

        def ParseFromString(self, _data):
            return None

    class DummyResponse:
        def __init__(self):
            self.file_descriptor_response = SimpleNamespace(file_descriptor_proto=[b"blob"])

        def HasField(self, name):
            return name == "file_descriptor_response"

    class DummyStub:
        def ServerReflectionInfo(self, _request_iter):
            return [DummyResponse()]

    monkeypatch.setattr("mcpgateway.translate_grpc.FileDescriptorProto", DummyFileDesc)
    monkeypatch.setattr("mcpgateway.translate_grpc.reflection_pb2", SimpleNamespace(ServerReflectionRequest=lambda **_kwargs: MagicMock()))

    await tg.GrpcEndpoint._discover_service_details(endpoint, DummyStub(), "pkg.TestService")
    assert endpoint._services == {}


def test_translator_methods_fallback_schema():
    endpoint = SimpleNamespace(
        _services={"svc": {"methods": [{"name": "Ping", "input_type": ".Missing", "output_type": ".Output"}]}},
        _pool=MagicMock(),
    )
    endpoint._pool.FindMessageTypeByName.side_effect = KeyError("missing")
    translator = GrpcToMcpTranslator(endpoint=endpoint)
    tools = translator.grpc_methods_to_mcp_tools("svc")
    assert tools[0]["inputSchema"]["properties"] == {}


def test_translator_methods_returns_empty_when_service_missing():
    endpoint = SimpleNamespace(_services={}, _pool=None)
    translator = GrpcToMcpTranslator(endpoint=endpoint)
    assert translator.grpc_methods_to_mcp_tools("missing") == []


def test_translator_methods_input_schema_success():
    dummy_desc = object()
    endpoint = SimpleNamespace(
        _services={"svc": {"methods": [{"name": "Ping", "input_type": ".Input", "output_type": ".Output"}]}},
        _pool=SimpleNamespace(FindMessageTypeByName=lambda _name: dummy_desc),
    )
    translator = GrpcToMcpTranslator(endpoint=endpoint)
    translator.protobuf_to_json_schema = MagicMock(return_value={"type": "object"})
    tools = translator.grpc_methods_to_mcp_tools("svc")
    assert tools[0]["inputSchema"] == {"type": "object"}


def test_import_fallback_sets_grpc_unavailable(monkeypatch):
    """ImportError branch sets GRPC_AVAILABLE False and placeholders to None."""
    # Standard
    import builtins
    import runpy

    # First-Party
    import mcpgateway.translate_grpc as tg

    real_import = builtins.__import__

    def fake_import(name, globals=None, locals=None, fromlist=(), level=0):  # noqa: A002 - signature matches builtin
        if name == "grpc":
            raise ImportError("forced missing grpc")
        return real_import(name, globals, locals, fromlist, level)

    monkeypatch.setattr(builtins, "__import__", fake_import)
    ns = runpy.run_path(tg.__file__, run_name="__translate_grpc_import_fallback_test__")
    assert ns["GRPC_AVAILABLE"] is False
    assert ns["grpc"] is None


@pytest.mark.skipif(not GRPC_AVAILABLE, reason="gRPC packages not installed")
class TestGrpcEndpointLoadFileDescriptors:
    """Tests for GrpcEndpoint.load_file_descriptors()."""

    def test_load_file_descriptors_populates_pool(self):
        """Test that valid FileDescriptorProto bytes are loaded into the pool."""
        # Third-Party
        from google.protobuf.descriptor_pb2 import FileDescriptorProto  # pylint: disable=no-name-in-module

        # Create a minimal valid FileDescriptorProto
        fd = FileDescriptorProto()
        fd.name = "test_load_fd.proto"
        fd.package = "test_load_fd"
        fd.syntax = "proto3"
        proto_bytes = fd.SerializeToString()

        endpoint = GrpcEndpoint(target="localhost:50051", reflection_enabled=False)
        endpoint.load_file_descriptors([proto_bytes])

        # The descriptor should now be findable in the pool
        found = endpoint._pool.FindFileByName("test_load_fd.proto")
        assert found is not None
        assert found.package == "test_load_fd"

    def test_load_file_descriptors_skips_duplicates(self):
        """Test that loading the same descriptor twice doesn't raise."""
        # Third-Party
        from google.protobuf.descriptor_pb2 import FileDescriptorProto  # pylint: disable=no-name-in-module

        fd = FileDescriptorProto()
        fd.name = "test_dup_fd.proto"
        fd.package = "test_dup_fd"
        fd.syntax = "proto3"
        proto_bytes = fd.SerializeToString()

        endpoint = GrpcEndpoint(target="localhost:50051", reflection_enabled=False)
        # Load twice — should not raise
        endpoint.load_file_descriptors([proto_bytes])
        endpoint.load_file_descriptors([proto_bytes])

    def test_load_file_descriptors_error_corrupt_bytes(self):
        """Test that corrupt/invalid bytes raise an error."""
        endpoint = GrpcEndpoint(target="localhost:50051", reflection_enabled=False)
        with pytest.raises(ValueError):
            endpoint.load_file_descriptors([b"not-valid-protobuf-data"])

    def test_load_file_descriptors_rejects_single_bytes(self):
        """Passing a single bytes object (not a sequence) is the classic Python footgun: the
        for-loop would iterate byte-by-byte. The guard must raise TypeError eagerly."""
        endpoint = GrpcEndpoint(target="localhost:50051", reflection_enabled=False)
        with pytest.raises(TypeError, match="must be a sequence of bytes"):
            endpoint.load_file_descriptors(b"\x00\x01\x02")
        with pytest.raises(TypeError, match="must be a sequence of bytes"):
            endpoint.load_file_descriptors(bytearray(b"\x00\x01\x02"))

    def test_load_file_descriptors_skips_pool_conflict(self):
        """A FileDescriptorProto conflict at pool.Add should be logged and skipped, not raised."""
        endpoint = GrpcEndpoint(target="localhost:50051", reflection_enabled=False)
        endpoint._pool = MagicMock()
        endpoint._pool.Add = MagicMock(side_effect=TypeError("Conflict register for file 'a.proto': type already exists"))

        # Build a minimally-valid FileDescriptorProto so ParseFromString succeeds,
        # then verify the TypeError from Add is swallowed (no exception propagates).
        # Third-Party
        from google.protobuf.descriptor_pb2 import FileDescriptorProto

        fd = FileDescriptorProto()
        fd.name = "a.proto"
        fd.package = "a"
        proto_bytes = fd.SerializeToString()

        endpoint.load_file_descriptors([proto_bytes])
        assert endpoint._pool.Add.call_count == 1
