From da443d1266615507f52a101b461c80e0265069ae Mon Sep 17 00:00:00 2001
From: Harry Qian <huan.qian@nxp.com>
Date: Tue, 4 Aug 2026 18:21:22 +0800
Subject: [PATCH] test(proxy): lock in query-param validation across fastapi
 param types

Guards _declared_query_params against a regression in the get_flat_params
migration: the flatten step returns path, query, header and cookie params
together, so a dropped ParamTypes.query filter would wrongly treat path or
header names as declared query params and accept unknown ones. Removing the
filter fails these tests.
---
 .../management_v1/test_common.py              | 95 +++++++++++++++++++
 1 file changed, 95 insertions(+)
 create mode 100644 tests/test_litellm/proxy/management_endpoints/management_v1/test_common.py

diff --git a/tests/test_litellm/proxy/management_endpoints/management_v1/test_common.py b/tests/test_litellm/proxy/management_endpoints/management_v1/test_common.py
new file mode 100644
index 00000000000..167a06ed551
--- /dev/null
+++ b/tests/test_litellm/proxy/management_endpoints/management_v1/test_common.py
@@ -0,0 +1,95 @@
+from typing import Annotated
+
+from fastapi import Depends, FastAPI, Header, Query, Request
+from fastapi.testclient import TestClient
+
+from litellm.proxy.management_endpoints.management_v1.common import (
+    ManagementProblem,
+    PROBLEM_CONTENT_TYPE,
+    _declared_query_params,
+    problem_response,
+    reject_unknown_query_params,
+)
+
+
+def _client() -> TestClient:
+    app = FastAPI()
+
+    @app.exception_handler(ManagementProblem)
+    async def _handle(_request: Request, exc: ManagementProblem):
+        return problem_response(exc.problem)
+
+    @app.get("/things/{thing_id}", dependencies=[Depends(reject_unknown_query_params)])
+    def _handler(
+        thing_id: str,
+        request: Request,
+        status: Annotated[str | None, Query(alias="filter[status]")] = None,
+        page: Annotated[int, Query(ge=1)] = 1,
+        x_trace: Annotated[str | None, Header()] = None,
+    ) -> dict[str, bool]:
+        return {"ok": True}
+
+    return TestClient(app, raise_server_exceptions=False)
+
+
+def test_a_declared_query_param_is_accepted_by_its_alias():
+    response = _client().get("/things/abc", params={"filter[status]": "active", "page": "2"})
+    assert response.status_code == 200, response.text
+
+
+def test_an_unknown_query_param_is_rejected_as_a_problem():
+    response = _client().get("/things/abc", params={"bogus": "x"})
+    assert response.status_code == 400
+    assert response.headers["content-type"].startswith(PROBLEM_CONTENT_TYPE)
+    assert "bogus" in response.json()["detail"]
+
+
+def test_a_path_param_name_is_not_a_declared_query_param():
+    """The flatten step returns path+query+header together; only query names count as declared.
+
+    If the ParamTypes.query filter were dropped, `thing_id` (a path param) would leak
+    into the declared set and this request would be wrongly accepted.
+    """
+    response = _client().get("/things/abc", params={"thing_id": "x"})
+    assert response.status_code == 400
+    assert "thing_id" in response.json()["detail"]
+
+
+def test_a_header_param_name_is_not_a_declared_query_param():
+    response = _client().get("/things/abc", params={"x-trace": "x"})
+    assert response.status_code == 400
+    assert "x-trace" in response.json()["detail"]
+
+
+def test_declared_query_params_isolates_query_aliases_from_other_param_types():
+    captured: dict[str, frozenset[str]] = {}
+    app = FastAPI()
+
+    @app.get("/things/{thing_id}")
+    def _handler(
+        thing_id: str,
+        request: Request,
+        status: Annotated[str | None, Query(alias="filter[status]")] = None,
+        page: Annotated[int, Query(ge=1)] = 1,
+        x_trace: Annotated[str | None, Header()] = None,
+    ) -> dict[str, bool]:
+        captured["declared"] = _declared_query_params(request)
+        return {"ok": True}
+
+    TestClient(app).get("/things/abc")
+    assert captured["declared"] == frozenset({"filter[status]", "page"})
+
+
+def test_declared_query_params_is_empty_when_the_route_has_no_dependant():
+    request = Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "scheme": "http",
+            "root_path": "",
+            "path": "/things/abc",
+            "query_string": b"",
+            "headers": [(b"host", b"testserver")],
+        }
+    )
+    assert _declared_query_params(request) == frozenset()