test(proxy): stop monkeypatch.undo re-planting fixture-mocked prisma_client

This commit is contained in:
mateo-berri 2026-08-13 20:11:06 -07:00
parent 86f2f16fd4
commit a36ba05882
3 changed files with 85 additions and 21 deletions

View file

@ -43,32 +43,51 @@ def disconnected_prisma() -> DisconnectedPrisma:
return DisconnectedPrisma()
@pytest.fixture(autouse=True)
def _isolate_proxy_module_globals():
"""
Snapshot and restore module-level globals on litellm.proxy.proxy_server
that tests sometimes mutate via raw setattr (not monkeypatch).
_MODULE_GLOBAL_MISSING = object()
_proxy_module_globals_snapshot = pytest.StashKey[Dict[str, object]]()
Without this, a leaked value — e.g. master_key set by a sibling test —
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(item):
"""
Snapshot module-level globals on litellm.proxy.proxy_server before any
fixture runs, and restore them in pytest_runtest_teardown after every
fixture finalizer has run.
Without this, a leaked value (e.g. master_key set by a sibling test)
flips the auth short-circuit in user_api_key_auth and causes unrelated
tests in the same xdist worker to return 401 instead of 200.
This must be a hook pair, not an autouse fixture: an autouse fixture in
the root conftest requests monkeypatch, so monkeypatch's undo stack
unwinds after every other fixture finalizer. A test that monkeypatches a
global while a fixture has it patched records the fixture's mock as the
"original", and monkeypatch.undo re-plants that mock after all restores
have run, poisoning the global for the rest of the xdist worker.
"""
from litellm.proxy import proxy_server
sentinel = object()
snapshot = {
name: getattr(proxy_server, name, sentinel)
item.stash[_proxy_module_globals_snapshot] = {
name: getattr(proxy_server, name, _MODULE_GLOBAL_MISSING)
for name in _PROXY_MODULE_GLOBALS_TO_ISOLATE
}
try:
yield
finally:
for name, value in snapshot.items():
if value is sentinel:
if hasattr(proxy_server, name):
delattr(proxy_server, name)
else:
setattr(proxy_server, name, value)
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_teardown(item, nextitem):
yield
snapshot = item.stash.get(_proxy_module_globals_snapshot, None)
if snapshot is None:
return
from litellm.proxy import proxy_server
for name, value in snapshot.items():
if value is _MODULE_GLOBAL_MISSING:
if hasattr(proxy_server, name):
delattr(proxy_server, name)
else:
setattr(proxy_server, name, value)
@pytest.fixture(autouse=True)

View file

@ -1688,9 +1688,15 @@ class TestTemporaryMCPSessionEndpoints:
expires_at=datetime.utcnow() - timedelta(seconds=30),
)
cache = {"expired": expired_entry}
with patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._temporary_mcp_servers",
cache,
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._temporary_mcp_servers",
cache,
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_prisma_client_or_none",
return_value=None,
),
):
result = await get_cached_temporary_mcp_server("expired")
@ -2274,6 +2280,10 @@ class TestTemporaryMCPSessionEndpoints:
"litellm.proxy.management_endpoints.mcp_management_endpoints._cache_temporary_mcp_server_in_redis",
AsyncMock(),
) as redis_cache_mock,
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_prisma_client_or_none",
return_value=None,
),
):
response = await add_session_mcp_server(
payload=payload,
@ -3419,6 +3429,10 @@ class TestTemporaryMCPSessionEndpoints:
"litellm.proxy.management_endpoints.mcp_management_endpoints.decrypt_value_helper",
return_value=serialized,
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_prisma_client_or_none",
return_value=None,
),
):
result = await get_cached_temporary_mcp_server("from-redis")
finally:

View file

@ -0,0 +1,31 @@
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@pytest.fixture
def fixture_planted_prisma_mock():
with patch("litellm.proxy.proxy_server.prisma_client", MagicMock()):
yield
def test_monkeypatch_over_fixture_patched_prisma_client(
fixture_planted_prisma_mock, monkeypatch
):
"""
Mirrors the flake in test_team_endpoints.py: an autouse fixture patches
prisma_client, the test monkeypatches the same global, and monkeypatch
records the fixture's MagicMock as the value to restore. Its undo runs
after every other finalizer, so without hook-level isolation the mock
leaks and every later no-database test on the worker fails awaiting it.
"""
import litellm.proxy.proxy_server as proxy_server
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", AsyncMock())
assert isinstance(proxy_server.prisma_client, AsyncMock)
def test_prisma_client_did_not_leak_from_previous_test():
import litellm.proxy.proxy_server as proxy_server
assert not isinstance(proxy_server.prisma_client, MagicMock)