Repository: home-assistant/core
File: tests/components/mcp/test_init.py (Python)

Code before:

    This exercises the case where the HTTP transport fails with method not
    allowed, indicating an SSE server, then also fails with SSE.
    """
    http_405 = httpx.HTTPStatusError(
        "Method not allowed", request=None, response=httpx.Response(405)
    )
    mock_http_streamable_client.side_effect = ExceptionGroup(
        "Method not allowed", [http_405]
    )

    mock_sse_client.side_effect = ExceptionGroup(
        "Connection error", [httpx.ConnectError("Connection failed")]
    )


@pytest.mark.parametrize(
    ("side_effect"),
    [
        (
            ExceptionGroup(
                "Method not allowed",
                [
                    httpx.HTTPStatusError(
                        "Method not allowed",
                        request=None,  # type: ignore[arg-type]
                        response=httpx.Response(405),
                    )
                ],
            ),
        ),
        (
            ExceptionGroup(
                "Some exception group",
                [McpError(ErrorData(code=500, message="Session terminated"))],
            )
        ),
    ],
)
async def test_mcp_client_fallback_to_sse_success(
    hass: HomeAssistant,
    config_entry: MockConfigEntry,
    mock_http_streamable_client: AsyncMock,
    mock_sse_client: AsyncMock,
    mock_mcp_client: Mock,
    side_effect: Exception,
) -> None:
    """Test mcp_client falls back to SSE on method not allowed error.

    This exercises the backwards compatibility part of the MCP Transport
    specification.

Review comment (suggestion):
A short suggestion, you could avoid this `type: ignore` by creating:

```python
dummy_request = httpx.Request("GET", "https://test.com")
# than using
request=dummy_request instead of None
```

Code after:

    This exercises the case where the HTTP transport fails with method not
    allowed, indicating an SSE server, then also fails with SSE.
    """
    http_405 = httpx.HTTPStatusError(
        "Method not allowed", request=None, response=httpx.Response(405)
    )
    mock_http_streamable_client.side_effect = ExceptionGroup(
        "Method not allowed", [http_405]
    )

    mock_sse_client.side_effect = ExceptionGroup(
        "Connection error", [httpx.ConnectError("Connection failed")]
    )


@pytest.mark.parametrize(
    ("side_effect"),
    [
        (
            ExceptionGroup(
                "Method not allowed",
                [
                    httpx.HTTPStatusError(
                        "Method not allowed",
                        request=None,
                        response=httpx.Response(405),
                    )
                ],
            ),
        ),
        (
            ExceptionGroup(
                "Some exception group",
                [McpError(ErrorData(code=500, message="Session terminated"))],
            )
        ),
    ],
)
async def test_mcp_client_fallback_to_sse_success(
    hass: HomeAssistant,
    config_entry: MockConfigEntry,
    mock_http_streamable_client: AsyncMock,
    mock_sse_client: AsyncMock,
    mock_mcp_client: Mock,
    side_effect: Exception,
) -> None:
    """Test mcp_client falls back to SSE on some errors.

    This exercises the backwards compatibility part of the MCP Transport
    specification.