27 lines
1.3 KiB
Python
27 lines
1.3 KiB
Python
"""
|
||
Интеграционные тесты для новых эндпоинтов RAG-сервера.
|
||
"""
|
||
|
||
import pytest
|
||
import httpx
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
# Используем httpx.AsyncClient для запросов к приложению
|
||
# Предполагаем, что приложение запущено в тестовом режиме или используем TestClient из FastAPI.
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clear_endpoint():
|
||
# В реальном тесте нужно запустить приложение и использовать TestClient.
|
||
# Здесь приведён пример с моками.
|
||
async with httpx.AsyncClient() as client:
|
||
# Мокаем зависимость
|
||
with patch("rag.rag_server.get_orchestrator") as mock_orch:
|
||
mock_orch.return_value.clear_kb = AsyncMock(return_value={"status": "ok"})
|
||
response = await client.post(
|
||
"http://localhost:8080/rag/clear",
|
||
json={"user_jid": "test@domain", "room_jid": None, "is_global": False},
|
||
headers={"X-API-Key": "testkey"}
|
||
)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data.get("status") == "ok" |