30 lines
879 B
Python
30 lines
879 B
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import gateway.main as module
|
|
|
|
|
|
class GatewayPointSearchTests(unittest.TestCase):
|
|
def test_point_search_proxies_to_qdrant_point_endpoint(self):
|
|
async def run():
|
|
with patch.object(module, "_post_json", new_callable=AsyncMock) as post:
|
|
post.return_value = {"results": []}
|
|
result = await module.vectors_search_point({"id": 1284, "limit": 5})
|
|
|
|
post.assert_awaited_once_with(
|
|
f"{module.QDRANT_SVC_URL}/search/point",
|
|
{"id": 1284, "limit": 5},
|
|
preserve_client_errors=True,
|
|
upstream_name="qdrant",
|
|
)
|
|
self.assertEqual(result, {"results": []})
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|