scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

test(proxy): restore app routes after pass-through reload tests

Browse source 
The two pass-through reload tests register real FastAPI routes on the
shared proxy app and only clean up the internal registry, so any test
running after them in the same worker sees stray /v1/kept-* and
/v1/deleted-* routes. test_component_allowlists counts those as
uncovered and fails. Snapshot app.routes and the registry up front and
restore both in a finally block.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-09-18 19:44:18 +00:00
parent a774e71175
commit 4f904f69b6
1 changed files with 35 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
tests/test_litellm/proxy/test_proxy_server.py
View file

@ -7536,25 +7536,35 @@ async def test_deleting_the_stored_pass_through_row_takes_the_route_out_of_servi
deleted. The proxy's own registry of live pass-through routes is what decides whether
a request is routed upstream or falls through to the auth error, so it has to lose the
entry on the reload rather than at the next process restart."""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import InitPassThroughEndpointHelpers
from litellm.proxy.proxy_server import ProxyConfig
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
_registered_pass_through_routes,
)
from litellm.proxy.proxy_server import ProxyConfig, app
path: Final = f"/v1/deleted-{uuid.uuid4().hex[:8]}"
db_endpoint: Final = {"id": "db-1", "path": path, "target": "https://example.com/post"}
prior_routes: Final = list(app.routes)
prior_registry: Final = dict(_registered_pass_through_routes)
def live_routes() -> set[str]:
return {route for route in InitPassThroughEndpointHelpers.get_all_registered_pass_through_routes() if path in route}
settings: Final = patch("litellm.proxy.proxy_server.general_settings", {}) # test-quality-ok: the method reads this module global; no injection seam
yaml_endpoints: Final = patch("litellm.proxy.proxy_server.config_passthrough_endpoints", None) # test-quality-ok: module global holding the YAML endpoints; this case has none
with settings, yaml_endpoints:
pc = ProxyConfig()
await pc._update_general_settings(db_general_settings={"pass_through_endpoints": [db_endpoint]})
assert live_routes(), "the stored endpoint should be serving before the row is deleted"
try:
with settings, yaml_endpoints:
pc = ProxyConfig()
await pc._update_general_settings(db_general_settings={"pass_through_endpoints": [db_endpoint]})
assert live_routes(), "the stored endpoint should be serving before the row is deleted"
await pc._update_general_settings(db_general_settings={})
await pc._update_general_settings(db_general_settings={})
assert live_routes() == set()
assert live_routes() == set()
finally:
app.routes[:] = prior_routes
_registered_pass_through_routes.clear()
_registered_pass_through_routes.update(prior_registry)
@pytest.mark.asyncio
@ -7564,15 +7574,18 @@ async def test_a_stored_pass_through_row_never_disturbs_the_config_declared_rout
serving untouched. The stored entry never gets a route of its own."""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
_registered_pass_through_routes,
initialize_pass_through_endpoints,
)
from litellm.proxy.proxy_server import ProxyConfig
from litellm.proxy.proxy_server import ProxyConfig, app
marker: Final = uuid.uuid4().hex[:8]
config_path: Final = f"/v1/kept-{marker}"
db_path: Final = f"/v1/ignored-{marker}"
config_endpoint: Final = {"id": f"cfg-{marker}", "path": config_path, "target": "https://example.com/post"}
db_endpoint: Final = {"id": f"db-{marker}", "path": db_path, "target": "https://example.com/post"}
prior_routes: Final = list(app.routes)
prior_registry: Final = dict(_registered_pass_through_routes)
def live_paths() -> set[str]:
registered: Final = InitPassThroughEndpointHelpers.get_all_registered_pass_through_routes()
@ -7580,17 +7593,22 @@ async def test_a_stored_pass_through_row_never_disturbs_the_config_declared_rout
settings: Final = patch("litellm.proxy.proxy_server.general_settings", {"pass_through_endpoints": [config_endpoint]}) # test-quality-ok: the method reads this module global; no injection seam
yaml_endpoints: Final = patch("litellm.proxy.proxy_server.config_passthrough_endpoints", [config_endpoint]) # test-quality-ok: module global holding the YAML endpoints the reload merges in
with settings, yaml_endpoints:
await initialize_pass_through_endpoints(pass_through_endpoints=[config_endpoint])
assert live_paths() == {config_path}
try:
with settings, yaml_endpoints:
await initialize_pass_through_endpoints(pass_through_endpoints=[config_endpoint])
assert live_paths() == {config_path}
pc = ProxyConfig()
await pc._update_general_settings(db_general_settings={"pass_through_endpoints": [db_endpoint]})
assert live_paths() == {config_path}
pc = ProxyConfig()
await pc._update_general_settings(db_general_settings={"pass_through_endpoints": [db_endpoint]})
assert live_paths() == {config_path}
await pc._update_general_settings(db_general_settings={})
await pc._update_general_settings(db_general_settings={})
assert live_paths() == {config_path}
assert live_paths() == {config_path}
finally:
app.routes[:] = prior_routes
_registered_pass_through_routes.clear()
_registered_pass_through_routes.update(prior_registry)
def _fill_user_api_key_cache(cache: DualCache, count: int) -> None: