80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
|
|
import qdrant.main as module
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, response=None, error=None):
|
|
self.response = response
|
|
self.error = error
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
async def post(self, *_args, **_kwargs):
|
|
if self.error:
|
|
raise self.error
|
|
return self.response
|
|
|
|
|
|
class QdrantClipSemanticsTests(unittest.IsolatedAsyncioTestCase):
|
|
async def call(self, operation, response=None, error=None):
|
|
client = _FakeClient(response=response, error=error)
|
|
with patch.object(module.httpx, "AsyncClient", lambda **_kwargs: client):
|
|
return await operation()
|
|
|
|
async def test_clip_4xx_is_preserved(self):
|
|
for status in (400, 404, 422):
|
|
with self.subTest(status=status):
|
|
response = httpx.Response(status, json={"detail": "SECRET BODY"}, request=httpx.Request("POST", "http://clip/embed"))
|
|
with self.assertRaises(module.HTTPException) as raised:
|
|
await self.call(lambda: module._embed_url("https://cdn.example/image.webp"), response=response)
|
|
self.assertEqual(raised.exception.status_code, status)
|
|
self.assertEqual(raised.exception.detail, "CLIP image request rejected.")
|
|
|
|
async def test_clip_5xx_and_transport_remain_502(self):
|
|
for status in (500, 502, 504):
|
|
with self.subTest(status=status):
|
|
response = httpx.Response(status, text="SECRET BODY", request=httpx.Request("POST", "http://clip/embed"))
|
|
with self.assertRaises(module.HTTPException) as raised:
|
|
await self.call(lambda: module._embed_url("https://cdn.example/image.webp"), response=response)
|
|
self.assertEqual(raised.exception.status_code, 502)
|
|
with self.assertRaises(module.HTTPException) as raised:
|
|
await self.call(lambda: module._embed_url("https://cdn.example/image.webp"), error=httpx.ConnectError("secret transport detail"))
|
|
self.assertEqual(raised.exception.status_code, 502)
|
|
|
|
async def test_file_endpoint_preserves_4xx_and_maps_5xx(self):
|
|
for status, expected in ((400, 400), (422, 422), (500, 502)):
|
|
with self.subTest(status=status):
|
|
response = httpx.Response(status, text="SECRET BODY", request=httpx.Request("POST", "http://clip/embed/file"))
|
|
with self.assertRaises(module.HTTPException) as raised:
|
|
await self.call(lambda: module._embed_bytes(b"image-bytes"), response=response)
|
|
self.assertEqual(raised.exception.status_code, expected)
|
|
|
|
async def test_success_has_no_failure_warning(self):
|
|
records = []
|
|
handler = logging.Handler()
|
|
handler.emit = lambda record: records.append(record.getMessage())
|
|
module.logger.addHandler(handler)
|
|
try:
|
|
response = httpx.Response(200, json={"vector": [0.1]}, request=httpx.Request("POST", "http://clip/embed"))
|
|
result = await self.call(lambda: module._embed_url("https://cdn.example/image.webp"), response=response)
|
|
self.assertEqual(result, [0.1])
|
|
self.assertEqual(records, [])
|
|
finally:
|
|
module.logger.removeHandler(handler)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|