|
1 | 1 | """Tests for the `httpx` → `httpx2` migration shim in `mcp.shared._httpx`. |
2 | 2 |
|
3 | | -`mcp` prefers `httpx2` and falls back to `httpx` with an `MCPDeprecationWarning`. The |
4 | | -fallback is exercised when only `httpx` is installed (today this is always — `httpx2` is |
5 | | -not yet on PyPI). The warning is emitted once per process from HTTP-touching surfaces. |
| 3 | +`mcp` prefers `httpx2` and falls back to `httpx` with an `MCPDeprecationWarning` emitted at |
| 4 | +the shim's import time. Today the lockfile pins `httpx` (not `httpx2`), so importing the shim |
| 5 | +exercises the fallback. |
6 | 6 | """ |
7 | 7 |
|
8 | 8 | from __future__ import annotations |
9 | 9 |
|
| 10 | +import importlib |
| 11 | +import sys |
10 | 12 | import warnings |
11 | 13 |
|
12 | 14 | import pytest |
13 | 15 |
|
14 | | -from mcp.shared import _httpx as httpx_shim |
15 | | -from mcp.shared._httpx import MCPDeprecationWarning, emit_httpx_deprecation_warning |
16 | | -from mcp.shared._httpx_utils import create_mcp_http_client |
| 16 | +from mcp.shared._warnings import MCPDeprecationWarning |
17 | 17 |
|
18 | | -pytestmark = pytest.mark.anyio |
19 | 18 |
|
| 19 | +def _force_reimport_shim() -> None: |
| 20 | + """Drop the cached shim module so the next import re-runs its top-level code.""" |
| 21 | + sys.modules.pop("mcp.shared._httpx", None) |
20 | 22 |
|
21 | | -@pytest.fixture |
22 | | -def reset_warning_flag(monkeypatch: pytest.MonkeyPatch): |
23 | | - """Reset the once-per-process flag so each test gets a fresh emission state.""" |
24 | | - monkeypatch.setattr(httpx_shim, "_warning_emitted", False) |
25 | 23 |
|
| 24 | +def test_fallback_emits_warning_at_import(monkeypatch: pytest.MonkeyPatch) -> None: |
| 25 | + """With only `httpx` installed, importing the shim emits `MCPDeprecationWarning`.""" |
| 26 | + monkeypatch.delitem(sys.modules, "httpx2", raising=False) |
| 27 | + _force_reimport_shim() |
26 | 28 |
|
27 | | -def test_emit_warns_when_httpx_is_deprecated(monkeypatch: pytest.MonkeyPatch, reset_warning_flag: None) -> None: |
28 | | - monkeypatch.setattr(httpx_shim, "_HTTPX_IS_DEPRECATED", True) |
29 | | - with pytest.warns(MCPDeprecationWarning, match=r"install `httpx2` instead"): |
30 | | - emit_httpx_deprecation_warning() |
31 | | - |
32 | | - |
33 | | -def test_emit_silent_when_httpx2_is_used(monkeypatch: pytest.MonkeyPatch, reset_warning_flag: None) -> None: |
34 | | - monkeypatch.setattr(httpx_shim, "_HTTPX_IS_DEPRECATED", False) |
35 | | - with warnings.catch_warnings(record=True) as caught: |
36 | | - warnings.simplefilter("always", MCPDeprecationWarning) |
37 | | - emit_httpx_deprecation_warning() |
38 | | - assert [w for w in caught if issubclass(w.category, MCPDeprecationWarning)] == [] |
| 29 | + from collections.abc import Mapping, Sequence |
39 | 30 |
|
| 31 | + real_import = __import__ |
40 | 32 |
|
41 | | -def test_emit_only_warns_once_per_process(monkeypatch: pytest.MonkeyPatch, reset_warning_flag: None) -> None: |
42 | | - monkeypatch.setattr(httpx_shim, "_HTTPX_IS_DEPRECATED", True) |
43 | | - with warnings.catch_warnings(record=True) as caught: |
44 | | - warnings.simplefilter("always", MCPDeprecationWarning) |
45 | | - emit_httpx_deprecation_warning() |
46 | | - emit_httpx_deprecation_warning() |
47 | | - emit_httpx_deprecation_warning() |
48 | | - matching = [w for w in caught if issubclass(w.category, MCPDeprecationWarning)] |
49 | | - assert len(matching) == 1 |
50 | | - |
| 33 | + def fake_import( |
| 34 | + name: str, |
| 35 | + globals: Mapping[str, object] | None = None, |
| 36 | + locals: Mapping[str, object] | None = None, |
| 37 | + fromlist: Sequence[str] = (), |
| 38 | + level: int = 0, |
| 39 | + ) -> object: |
| 40 | + if name == "httpx2": |
| 41 | + raise ImportError("simulated: httpx2 not installed") |
| 42 | + return real_import(name, globals, locals, fromlist, level) |
51 | 43 |
|
52 | | -async def test_create_mcp_http_client_emits_warning(monkeypatch: pytest.MonkeyPatch, reset_warning_flag: None) -> None: |
53 | | - monkeypatch.setattr(httpx_shim, "_HTTPX_IS_DEPRECATED", True) |
| 44 | + monkeypatch.setattr("builtins.__import__", fake_import) |
54 | 45 | with pytest.warns(MCPDeprecationWarning, match=r"install `httpx2` instead"): |
55 | | - async with create_mcp_http_client(): |
56 | | - pass |
57 | | - |
58 | | - |
59 | | -async def test_create_mcp_http_client_silent_with_httpx2( |
60 | | - monkeypatch: pytest.MonkeyPatch, reset_warning_flag: None |
61 | | -) -> None: |
62 | | - monkeypatch.setattr(httpx_shim, "_HTTPX_IS_DEPRECATED", False) |
63 | | - with warnings.catch_warnings(record=True) as caught: |
64 | | - warnings.simplefilter("always", MCPDeprecationWarning) |
65 | | - async with create_mcp_http_client(): |
66 | | - pass |
67 | | - assert [w for w in caught if issubclass(w.category, MCPDeprecationWarning)] == [] |
68 | | - |
69 | | - |
70 | | -def test_shim_picks_up_httpx2_when_present(monkeypatch: pytest.MonkeyPatch) -> None: |
71 | | - """Aliasing `httpx2` to the installed `httpx` module exercises the preferred import path. |
| 46 | + importlib.import_module("mcp.shared._httpx") |
72 | 47 |
|
73 | | - `mcp`'s lockfile pins `httpx` (not `httpx2`), so the `import httpx2 as httpx` branch is |
74 | | - otherwise uncovered in CI. This test injects `httpx2` into `sys.modules` and reloads the |
75 | | - shim to cover that branch deterministically. |
76 | | - """ |
77 | | - import importlib |
78 | | - import sys |
79 | 48 |
|
| 49 | +def test_httpx2_present_is_silent(monkeypatch: pytest.MonkeyPatch) -> None: |
| 50 | + """When `httpx2` is importable, the shim selects it and emits no warning.""" |
80 | 51 | import httpx as real_httpx |
81 | 52 |
|
82 | 53 | monkeypatch.setitem(sys.modules, "httpx2", real_httpx) |
83 | | - monkeypatch.delitem(sys.modules, "mcp.shared._httpx", raising=False) |
84 | | - try: |
| 54 | + _force_reimport_shim() |
| 55 | + |
| 56 | + with warnings.catch_warnings(record=True) as caught: |
| 57 | + warnings.simplefilter("always", MCPDeprecationWarning) |
85 | 58 | reloaded = importlib.import_module("mcp.shared._httpx") |
86 | | - assert reloaded._HTTPX_IS_DEPRECATED is False |
87 | | - assert reloaded.httpx is real_httpx |
88 | | - finally: |
89 | | - # Restore the canonical shim module so subsequent tests see the real state. |
90 | | - sys.modules.pop("mcp.shared._httpx", None) |
91 | | - importlib.import_module("mcp.shared._httpx") |
| 59 | + |
| 60 | + assert reloaded.httpx is real_httpx |
| 61 | + assert [w for w in caught if issubclass(w.category, MCPDeprecationWarning)] == [] |
0 commit comments