test: increased cb test coverage
This commit is contained in:
committed by
Luijkx,S.O.H. (Storm)
parent
de2e56ffce
commit
7f7c658901
75
test/unit/test_main.py
Normal file
75
test/unit/test_main.py
Normal file
@@ -0,0 +1,75 @@
|
||||
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.VADAgent.start", new_callable=AsyncMock),
|
||||
patch("control_backend.main.VADAgent.reset_stream", new_callable=AsyncMock),
|
||||
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)
|
||||
Reference in New Issue
Block a user