9cc0e39955
The test still expects the VAD agent to be started in main, rather than in the RI Communication Agent. ref: N25B-356
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
import asyncio
|
||
import sys
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from control_backend.api.v1.router import api_router
|
||
from control_backend.main import app, lifespan
|
||
|
||
# Fix event loop on Windows
|
||
if sys.platform == "win32":
|
||
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
||
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
# Patch setup_logging so it does nothing
|
||
with patch("control_backend.main.setup_logging"):
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
|
||
def test_root_fast():
|
||
# Patch heavy startup code so it doesn’t slow down
|
||
with patch("control_backend.main.setup_logging"), patch("control_backend.main.lifespan"):
|
||
client = TestClient(app)
|
||
resp = client.get("/")
|
||
assert resp.status_code == 200
|
||
assert resp.json() == {"status": "ok"}
|
||
|
||
|
||
def test_cors_middleware_added():
|
||
"""Test that CORSMiddleware is correctly added to the app."""
|
||
from starlette.middleware.cors import CORSMiddleware
|
||
|
||
middleware_classes = [m.cls for m in app.user_middleware]
|
||
assert CORSMiddleware in middleware_classes
|
||
|
||
|
||
def test_api_router_included():
|
||
"""Test that the API router is included in the FastAPI app."""
|
||
|
||
route_paths = [r.path for r in app.routes]
|
||
for route in api_router.routes:
|
||
assert route.path in route_paths
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_lifespan_agent_start_exception():
|
||
"""
|
||
Trigger an exception during agent startup to cover the error logging branch.
|
||
Ensures exceptions are logged properly and re-raised.
|
||
"""
|
||
with (
|
||
patch(
|
||
"control_backend.main.RICommunicationAgent.start", new_callable=AsyncMock
|
||
) as ri_start,
|
||
patch("control_backend.main.setup_logging"),
|
||
patch("control_backend.main.threading.Thread"),
|
||
):
|
||
# Force RICommunicationAgent.start to raise an exception
|
||
ri_start.side_effect = Exception("Test exception")
|
||
|
||
with patch("control_backend.main.logger") as mock_logger:
|
||
with pytest.raises(Exception, match="Test exception"):
|
||
async with lifespan(app):
|
||
pass
|
||
|
||
# Verify the error was logged correctly
|
||
assert mock_logger.error.called
|
||
args, _ = mock_logger.error.call_args
|
||
assert isinstance(args[2], Exception)
|