d4c-service-geo-assistant/tests/test_api.py

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from uuid import uuid4

from geo_assistant.api.app import app
from geo_assistant.agent.graph import create_graph


@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