klevze
1713f0ff79
Map Qdrant 400-499 through as the same status with a short public detail, and keep 5xx and transport failures as sanitized 502. CLIP and other services still use the old helper default.
274 lines
12 KiB
Python
274 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from typing import Any, Dict, Optional
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
|
|
from tests.test_gateway_llm import load_gateway_module
|
|
|
|
|
|
QDRANT = "http://qdrant-svc:8000"
|
|
LEAK = "SECRET-QDRANT-BODY-DO-NOT-LEAK"
|
|
AUTH = {"X-API-Key": "test-key"}
|
|
|
|
|
|
class StubVectorClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
post_response: Optional[httpx.Response] = None,
|
|
get_response: Optional[httpx.Response] = None,
|
|
delete_response: Optional[httpx.Response] = None,
|
|
post_exception: Optional[Exception] = None,
|
|
get_exception: Optional[Exception] = None,
|
|
delete_exception: Optional[Exception] = None,
|
|
):
|
|
self.post_response = post_response
|
|
self.get_response = get_response
|
|
self.delete_response = delete_response
|
|
self.post_exception = post_exception
|
|
self.get_exception = get_exception
|
|
self.delete_exception = delete_exception
|
|
self.last_post: Dict[str, Any] = {}
|
|
|
|
async def post(self, url: str, **kwargs: Any) -> httpx.Response:
|
|
self.last_post = {"url": url, **kwargs}
|
|
if self.post_exception is not None:
|
|
raise self.post_exception
|
|
if self.post_response is None:
|
|
return httpx.Response(404, json={"detail": f"No stub for POST {url}"})
|
|
return self.post_response
|
|
|
|
async def get(self, url: str, **kwargs: Any) -> httpx.Response:
|
|
if self.get_exception is not None:
|
|
raise self.get_exception
|
|
if self.get_response is None:
|
|
return httpx.Response(404, json={"detail": f"No stub for GET {url}"})
|
|
return self.get_response
|
|
|
|
async def delete(self, url: str, **kwargs: Any) -> httpx.Response:
|
|
if self.delete_exception is not None:
|
|
raise self.delete_exception
|
|
if self.delete_response is None:
|
|
return httpx.Response(404, json={"detail": f"No stub for DELETE {url}"})
|
|
return self.delete_response
|
|
|
|
|
|
def _json_response(status: int, payload: Dict[str, Any], method: str = "POST", url: str = f"{QDRANT}/search") -> httpx.Response:
|
|
return httpx.Response(status, json=payload, request=httpx.Request(method, url))
|
|
|
|
|
|
class GatewayVectorErrorTests(unittest.IsolatedAsyncioTestCase):
|
|
async def _request(
|
|
self,
|
|
module: Any,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
json_payload: Optional[Dict[str, Any]] = None,
|
|
files: Any = None,
|
|
data: Any = None,
|
|
) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
return await client.request(
|
|
method,
|
|
path,
|
|
headers=AUTH,
|
|
json=json_payload,
|
|
files=files,
|
|
data=data,
|
|
)
|
|
|
|
async def _search(self, module: Any, stub: StubVectorClient, payload: Optional[Dict[str, Any]] = None) -> httpx.Response:
|
|
with patch.object(module, "get_http_client", return_value=stub):
|
|
return await self._request(
|
|
module,
|
|
"POST",
|
|
"/vectors/search",
|
|
json_payload=payload or {"url": "https://cdn.example/art.webp", "limit": 12},
|
|
)
|
|
|
|
def _assert_sanitized(self, response: httpx.Response) -> None:
|
|
body = response.text
|
|
self.assertNotIn(LEAK, body)
|
|
self.assertNotIn("qdrant-svc", body)
|
|
self.assertNotIn("http://qdrant-svc:8000", body)
|
|
|
|
async def test_search_success_unchanged(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
payload = {"results": [{"id": 7, "score": 0.91}]}
|
|
stub = StubVectorClient(post_response=_json_response(200, payload))
|
|
response = await self._search(module, stub)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json(), payload)
|
|
|
|
async def test_search_preserves_client_errors(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
cases = (400, 401, 403, 404, 408, 409, 413, 422, 429)
|
|
for status in cases:
|
|
with self.subTest(status=status):
|
|
stub = StubVectorClient(
|
|
post_response=_json_response(status, {"detail": LEAK}),
|
|
)
|
|
response = await self._search(module, stub)
|
|
self.assertEqual(response.status_code, status)
|
|
self._assert_sanitized(response)
|
|
self.assertEqual(response.json()["detail"], "Vector service rejected the request.")
|
|
|
|
async def test_search_maps_server_errors_to_502(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
for status in (500, 503):
|
|
with self.subTest(status=status):
|
|
stub = StubVectorClient(
|
|
post_response=_json_response(status, {"detail": LEAK}),
|
|
)
|
|
response = await self._search(module, stub)
|
|
self.assertEqual(response.status_code, 502)
|
|
self._assert_sanitized(response)
|
|
self.assertEqual(response.json()["detail"], "Vector service unavailable.")
|
|
|
|
async def test_search_transport_error_is_sanitized_502(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
stub = StubVectorClient(
|
|
post_exception=httpx.ConnectError("boom", request=httpx.Request("POST", f"{QDRANT}/search")),
|
|
)
|
|
response = await self._search(module, stub)
|
|
self.assertEqual(response.status_code, 502)
|
|
body = response.text
|
|
self.assertNotIn("boom", body)
|
|
self.assertNotIn("qdrant-svc", body)
|
|
self.assertNotIn("http://", body)
|
|
self.assertEqual(response.json()["detail"], "Vector service unavailable.")
|
|
|
|
async def test_search_timeout_is_sanitized_502(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
stub = StubVectorClient(
|
|
post_exception=httpx.ReadTimeout("timed out", request=httpx.Request("POST", f"{QDRANT}/search")),
|
|
)
|
|
response = await self._search(module, stub)
|
|
self.assertEqual(response.status_code, 502)
|
|
self.assertNotIn("timed out", response.text)
|
|
self.assertEqual(response.json()["detail"], "Vector service unavailable.")
|
|
|
|
async def test_search_file_success_and_errors(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
|
|
stub_ok = StubVectorClient(post_response=_json_response(200, {"results": []}, url=f"{QDRANT}/search/file"))
|
|
with patch.object(module, "get_http_client", return_value=stub_ok):
|
|
ok = await self._request(
|
|
module,
|
|
"POST",
|
|
"/vectors/search/file",
|
|
files={"file": ("query.webp", b"image-bytes", "image/webp")},
|
|
data={"limit": "5"},
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
self.assertIn("file", stub_ok.last_post.get("files", {}))
|
|
self.assertEqual(stub_ok.last_post.get("data", {}).get("limit"), "5")
|
|
|
|
stub_422 = StubVectorClient(post_response=_json_response(422, {"detail": LEAK}, url=f"{QDRANT}/search/file"))
|
|
with patch.object(module, "get_http_client", return_value=stub_422):
|
|
bad = await self._request(
|
|
module,
|
|
"POST",
|
|
"/vectors/search/file",
|
|
files={"file": ("query.webp", b"image-bytes", "image/webp")},
|
|
data={"limit": "5"},
|
|
)
|
|
self.assertEqual(bad.status_code, 422)
|
|
self._assert_sanitized(bad)
|
|
|
|
stub_500 = StubVectorClient(post_response=_json_response(500, {"detail": LEAK}, url=f"{QDRANT}/search/file"))
|
|
with patch.object(module, "get_http_client", return_value=stub_500):
|
|
server = await self._request(
|
|
module,
|
|
"POST",
|
|
"/vectors/search/file",
|
|
files={"file": ("query.webp", b"image-bytes", "image/webp")},
|
|
data={"limit": "5"},
|
|
)
|
|
self.assertEqual(server.status_code, 502)
|
|
self._assert_sanitized(server)
|
|
|
|
stub_net = StubVectorClient(
|
|
post_exception=httpx.ConnectError("boom", request=httpx.Request("POST", f"{QDRANT}/search/file")),
|
|
)
|
|
with patch.object(module, "get_http_client", return_value=stub_net):
|
|
net = await self._request(
|
|
module,
|
|
"POST",
|
|
"/vectors/search/file",
|
|
files={"file": ("query.webp", b"image-bytes", "image/webp")},
|
|
data={"limit": "5"},
|
|
)
|
|
self.assertEqual(net.status_code, 502)
|
|
self.assertNotIn("boom", net.text)
|
|
|
|
async def test_get_collection_404_and_500(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
stub_404 = StubVectorClient(
|
|
get_response=_json_response(404, {"detail": LEAK}, method="GET", url=f"{QDRANT}/collections/nonexistent"),
|
|
)
|
|
with patch.object(module, "get_http_client", return_value=stub_404):
|
|
missing = await self._request(module, "GET", "/vectors/collections/nonexistent")
|
|
self.assertEqual(missing.status_code, 404)
|
|
self._assert_sanitized(missing)
|
|
|
|
stub_500 = StubVectorClient(
|
|
get_response=_json_response(500, {"detail": LEAK}, method="GET", url=f"{QDRANT}/collections/nonexistent"),
|
|
)
|
|
with patch.object(module, "get_http_client", return_value=stub_500):
|
|
server = await self._request(module, "GET", "/vectors/collections/nonexistent")
|
|
self.assertEqual(server.status_code, 502)
|
|
self._assert_sanitized(server)
|
|
|
|
async def test_delete_collection_semantics(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
stub_ok = StubVectorClient(delete_response=_json_response(200, {"ok": True}, method="DELETE", url=f"{QDRANT}/collections/test"))
|
|
with patch.object(module, "get_http_client", return_value=stub_ok):
|
|
ok = await self._request(module, "DELETE", "/vectors/collections/test")
|
|
self.assertEqual(ok.status_code, 200)
|
|
self.assertEqual(ok.json(), {"ok": True})
|
|
|
|
stub_404 = StubVectorClient(delete_response=_json_response(404, {"detail": LEAK}, method="DELETE", url=f"{QDRANT}/collections/test"))
|
|
with patch.object(module, "get_http_client", return_value=stub_404):
|
|
missing = await self._request(module, "DELETE", "/vectors/collections/test")
|
|
self.assertEqual(missing.status_code, 404)
|
|
self._assert_sanitized(missing)
|
|
|
|
stub_500 = StubVectorClient(delete_response=_json_response(500, {"detail": LEAK}, method="DELETE", url=f"{QDRANT}/collections/test"))
|
|
with patch.object(module, "get_http_client", return_value=stub_500):
|
|
server = await self._request(module, "DELETE", "/vectors/collections/test")
|
|
self.assertEqual(server.status_code, 502)
|
|
self._assert_sanitized(server)
|
|
|
|
stub_net = StubVectorClient(
|
|
delete_exception=httpx.ConnectError("boom", request=httpx.Request("DELETE", f"{QDRANT}/collections/test")),
|
|
)
|
|
with patch.object(module, "get_http_client", return_value=stub_net):
|
|
net = await self._request(module, "DELETE", "/vectors/collections/test")
|
|
self.assertEqual(net.status_code, 502)
|
|
self.assertNotIn("boom", net.text)
|
|
|
|
async def test_clip_422_still_mapped_to_502(self):
|
|
module = load_gateway_module(llm_enabled=False)
|
|
stub = StubVectorClient(
|
|
post_response=httpx.Response(
|
|
422,
|
|
json={"detail": "SECRET-CLIP-BODY"},
|
|
request=httpx.Request("POST", "http://clip:8000/analyze"),
|
|
),
|
|
)
|
|
with patch.object(module, "get_http_client", return_value=stub):
|
|
response = await self._request(
|
|
module,
|
|
"POST",
|
|
"/analyze/clip",
|
|
json_payload={"url": "https://cdn.example/art.webp"},
|
|
)
|
|
self.assertEqual(response.status_code, 502)
|
|
self.assertNotEqual(response.status_code, 422)
|