mirror of
https://github.com/dataforcanada/d4c-service-geo-assistant.git
synced 2026-06-13 22:41:01 +02:00
Wei Ji
e3530cefd2
* Use pyupgrade (UP) rule * Use pyflakes (F) rule * Use isort (I) rule * Use ruff-specific (RUF) rules * Use flake8-commas (COM) rule * Fix UP043 Unnecessary default type arguments
54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from geo_assistant.agent.graph import create_graph
|
|
from geo_assistant.api.app import app
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def initialized_app():
|
|
"""Initialize the app's chatbot before testing"""
|
|
# Manually initialize the chatbot as the lifespan would
|
|
app.state.chatbot = await create_graph()
|
|
yield app
|
|
# Cleanup if needed
|
|
if hasattr(app.state, "chatbot"):
|
|
del app.state.chatbot
|
|
|
|
|
|
@pytest.mark.xfail
|
|
async def test_hello_world(initialized_app):
|
|
"""Hello world test for the API"""
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=initialized_app),
|
|
base_url="http://test",
|
|
) as client:
|
|
thread_id = uuid4()
|
|
response = await client.post(
|
|
"/chat",
|
|
json={
|
|
"agent_state_input": {
|
|
"messages": [
|
|
{
|
|
"content": "Find the Neighbourhood Cafe in Lisbon and buffer 0.5km around it",
|
|
"type": "human",
|
|
},
|
|
],
|
|
"place": None,
|
|
"search_area": None,
|
|
},
|
|
"thread_id": str(thread_id),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/x-ndjson; charset=utf-8"
|
|
|
|
# Read the streaming response
|
|
content = response.text
|
|
assert content is not None
|
|
assert len(content) > 0
|