litellm/tests/code_coverage_tests/router_code_coverage.py
RachelHuangZW bf9f59d76b
fix(scheduler): remove a request's queue entry once it stops waiting (#43061)
* fix(scheduler): remove a request's queue entry once it stops waiting

Requests admitted while a healthy deployment existed never left the priority
queue, and neither did requests cancelled while waiting. The stale entries
blocked later requests during cooldown and, with Redis, made add_request
raise a TypeError on queues read back as JSON lists

Both scheduling paths now share one polling helper that removes the entry in
a finally block, whether the request was admitted, timed out or cancelled

Related to #43059

* test(router): allowlist _wait_for_scheduler_turn in the router coverage check

The coverage script only counts direct calls in test files. The helper is
exercised through prioritized acompletion and atext_completion in
test_router.py, like the other allowlisted entries

* fix(scheduler): admit healthy requests before reading the queue, and clean up cancelled enqueues

poll() raised on an empty queue before checking for healthy deployments. With
the cleanup now rewriting the queue after every admission, a concurrent write
from another replica can erase a waiting request's entry, and that request
then failed while a deployment was healthy. poll() now admits as soon as a
deployment is healthy and only reads the queue during cooldown

add_request also moved inside the try block, so a request cancelled while its
queue write is in flight still has its entry removed

* refactor(scheduler): move the wait loop into Scheduler.wait_for_turn

The router passes a healthy-deployments callable into the scheduler, so tests
inject a Scheduler directly instead of replacing the router's scheduler
attribute. The cancelled-mid-enqueue test moves to test_scheduler.py with the
other scheduler tests, and poll() takes the deployments as a Sequence since it
only checks whether any are healthy

* test(scheduler): move scheduler tests into tests/unit

* fix(scheduler): finish the queue removal when cancellation is delivered again during cleanup

A second cancel, or an anyio cancel scope that re-cancels on every await,
interrupted remove_request mid-write and left the entry in Redis.

* fix(scheduler): read queue entries back from redis as tuples

* fix(scheduler): admit a request whose queue entry vanished during cooldown

* fix(scheduler): re-enqueue a request whose queue entry vanished during cooldown

* test(integration): cover priority scheduler queue cleanup across instances

Adds tests/integration/routing/test_priority_scheduler_queue_cleanup.py: 27 cells
against the real proxy (two workers, Postgres, Redis, a scripted upstream) for every
prioritized surface (/v1/chat/completions, /v1/completions, /queue/chat/completions,
streaming and not, OpenAI SDK sync and async, raw httpx), the non-integer priority
pass-through, the in-memory queue on one proxy, two proxies sharing a Redis queue
(served requests leave no entry, a dead replica's entry is skipped or expires, a
waiter that times out or disconnects removes only itself), a Redis outage mid burst
and a SIGKILLed worker. Each cell asserts the caller's response, the upstream's
requests by marker and the Redis queue contents. On the merge base the cross-instance
cells fail with the list-of-lists TypeError and the in-memory cell with 408s behind
the leaked entry; on this branch every cell passes

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
2026-10-07 17:34:38 -07:00

154 lines
8 KiB
Python

import ast
import os
def get_function_names_from_file(file_path):
"""
Extracts all function names from a given Python file.
"""
with open(file_path, "r", encoding="utf-8") as file:
tree = ast.parse(file.read())
function_names = []
for node in tree.body:
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
# Top-level functions
function_names.append(node.name)
elif isinstance(node, ast.ClassDef):
# Functions inside classes
for class_node in node.body:
if isinstance(class_node, (ast.FunctionDef, ast.AsyncFunctionDef)):
function_names.append(class_node.name)
return function_names
def get_all_functions_called_in_tests(base_dir):
"""
Returns a set of function names that are called in test functions
inside 'local_testing' and 'router_unit_test' directories,
specifically in files containing the word 'router'.
"""
called_functions = set()
test_dirs = ["local_testing", "router_unit_tests", "test_litellm", "unit"]
for test_dir in test_dirs:
dir_path = os.path.join(base_dir, test_dir)
if not os.path.exists(dir_path):
print(f"Warning: Directory {dir_path} does not exist.")
continue
print("dir_path: ", dir_path)
for root, _, files in os.walk(dir_path):
for file in files:
if file.endswith(".py") and ("router" in file.lower() or test_dir == "unit"):
print("file: ", file)
file_path = os.path.join(root, file)
with open(file_path, "r", encoding="utf-8") as f:
try:
tree = ast.parse(f.read())
except SyntaxError:
print(f"Warning: Syntax error in file {file_path}")
continue
if file == "test_router_validate_fallbacks.py":
print(f"tree: {tree}")
for node in ast.walk(tree):
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
called_functions.add(node.func.id)
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
called_functions.add(node.func.attr)
return called_functions
def get_functions_from_router(file_path):
"""
Extracts all functions defined in router.py.
"""
return get_function_names_from_file(file_path)
ignored_function_names = [
"_acancel_batch",
"_acreate_batch",
"_acreate_file",
"__init__",
"avector_store_create", # Tested via proxy vector_store_endpoints (files lack "router" in name)
"_override_vector_store_methods_for_router", # No-op placeholder, called during Router init
"_merge_tools_from_deployment", # Tested indirectly via _update_kwargs_with_deployment (test files lack "router" in name)
"_invalidate_access_groups_cache", # Tested indirectly via set_model_list, upsert_model etc. (test files lack "router" in name)
"has_buffered_provider_output", # Property, so its reads in test_router.py are never an ast.Call
"chunks", # Property on FallbackAwareAnthropicMessagesStream, so its reads in tests are never an ast.Call
"messages", # Property on FallbackAwareAnthropicMessagesStream, so its reads in tests are never an ast.Call
"model", # Property on FallbackAwareAnthropicMessagesStream, so its reads in tests are never an ast.Call
"_routing_groups", # Property getter and setter reads are never ast.Call nodes
"_request_header", # Tested through Claude Code session routing in test_router.py
"_claude_code_session_router_cache_key", # Tested through Claude Code session routing in test_router.py
"_delete_claude_code_session_router_binding", # Tested through Redis cleanup failure in test_router.py
"_resolve_claude_code_session_router", # Tested through Claude Code session routing in test_router.py
"_get_claude_code_session_router_binding", # Tested through the two-worker session routing test in test_router.py
"_apply_updated_routing_strategy_args", # Tested via update_settings in test_lowest_latency.py (file lacks "router" in name)
"_wait_for_scheduler_turn", # Tested through prioritized acompletion and atext_completion in test_router.py
"arm_routing_read_prefetch", # Tested in tests/unit/caching/test_request_redis_batch_pre_call.py (file lacks "router" in name)
"_configured_model_info", # Tested through get_configured_service_tiers in test_router.py
"_routable_deployments", # Tested through get_configured_service_tiers and get_routable_upstream_model in test_router.py
"_async_get_available_deployment", # Body of the `route {model}` phase wrapper, exercised through async_get_available_deployment in test_router.py
"_async_get_available_deployment_for_pass_through", # Same, through async_get_available_deployment_for_pass_through in test_router.py
"_embedding",
"_aembedding",
"_anthropic_stream_pre_content_error", # Tested through the non-retriable retry error tests in test_router.py
"_deployment_num_retries", # Tested through the deployment num_retries mid-stream budget test in test_router.py
"_request_fallback_list", # Tested through every mid-stream retry test in test_router.py
"_request_model_group", # Tested through test_anthropic_messages_retry_budget_precedence_direct_call
"_mid_stream_retry_trigger", # Tested through the retry policy mid-stream budget test in test_router.py
"_anthropic_messages_group_retry_policy", # Tested through the retry budget precedence test in test_router.py
"_anthropic_messages_resolved_retry_policy", # Tested through the malformed retry policy tests in test_router.py
"_anthropic_messages_plain_retry_budget", # Tested through the retry budget precedence test in test_router.py
"_anthropic_messages_should_retry", # Tested through every mid-stream retry test in test_router.py
"_aanthropic_messages_retry_same_group", # Tested through the dropped-before-content retry tests in test_router.py
"_aanthropic_messages_yield_recovered", # Tested through every mid-stream retry and fallback test in test_router.py
"_anthropic_messages_policy_retries", # Tested through the retry budget precedence test in test_router.py
"_get_wildcard_deployments", # Tested through the get_model_list_of_routed_group wildcard test in test_router.py
]
def main():
router_file = [
"./litellm/router.py",
"./litellm/router_utils/batch_utils.py",
"./litellm/router_utils/pattern_match_deployments.py",
]
# router_file = [
# "../../litellm/router.py",
# "../../litellm/router_utils/pattern_match_deployments.py",
# "../../litellm/router_utils/batch_utils.py",
# ] ## LOCAL TESTING
tests_dir = "./tests/" # Update this path if your tests directory is located elsewhere
# tests_dir = "../../tests/" # LOCAL TESTING
router_functions = []
for file in router_file:
router_functions.extend(get_functions_from_router(file))
print("router_functions: ", router_functions)
called_functions_in_tests = get_all_functions_called_in_tests(tests_dir)
untested_functions = [fn for fn in router_functions if fn not in called_functions_in_tests]
if untested_functions:
all_untested_functions = []
for func in untested_functions:
if func not in ignored_function_names:
all_untested_functions.append(func)
untested_perc = (len(all_untested_functions)) / len(router_functions)
print("untested_perc: ", untested_perc)
if untested_perc > 0:
print("The following functions in router.py are not tested:")
raise Exception(
f"{untested_perc * 100:.2f}% of functions in router.py are not tested: {all_untested_functions}"
)
else:
print("All functions in router.py are covered by tests.")
if __name__ == "__main__":
main()