busya
f87c873027
- app.py split from 1966-line test_app.py into 7 files (68 tests, 92% coverage): - test_app_lifespan: lifespan, ensure_initial_admin_user - test_app_handlers: exception handlers, HSTS - test_app_middleware: log_requests, middleware chain - test_app_ws_auth: _authenticate_websocket - test_app_ws_endpoint: WS main loop, 5 endpoint handlers - test_app_ws_events: task/maintenance/dataset/translate WS streams - test_app_spa: SPA serving, TestClient integration - migration_engine: extended coverage with init, edge cases, error paths
92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
# #region Test.AppModule.Handlers [C:3] [TYPE Module] [SEMANTICS test,app,handlers,exceptions]
|
|
# @BRIEF Tests for app.py — global_exception_handler, network_error_handler.
|
|
# @RELATION BINDS_TO -> [AppModule]
|
|
# @TEST_EDGE: unhandled_exception -> returns 500 JSON with path
|
|
# @TEST_EDGE: network_error -> returns 503
|
|
# @TEST_INVARIANT: every_500_logged -> VERIFIED_BY: test_global_exception_handler_logs_and_returns_500
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import json
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
from fastapi import Request, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
|
|
|
|
# #region _make_mock_request [C:1] [TYPE Function]
|
|
def _make_mock_request(method="GET", path="/api/test", client_host="127.0.0.1"):
|
|
req = MagicMock(spec=Request)
|
|
req.method = method
|
|
req.url = MagicMock()
|
|
req.url.path = path
|
|
req.client = MagicMock()
|
|
req.client.host = client_host
|
|
req.query_params = {}
|
|
return req
|
|
# #endregion _make_mock_request
|
|
|
|
|
|
class TestExceptionHandlers:
|
|
"""global_exception_handler and network_error_handler."""
|
|
|
|
# #region test_global_handler_500 [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_global_handler_500(self):
|
|
from src.app import global_exception_handler
|
|
request = _make_mock_request(method="POST", path="/api/break")
|
|
response = await global_exception_handler(request, ValueError("broke"))
|
|
assert isinstance(response, JSONResponse)
|
|
assert response.status_code == 500
|
|
body = json.loads(response.body)
|
|
assert body["detail"] == "Internal server error"
|
|
assert body["path"] == "/api/break"
|
|
# #endregion test_global_handler_500
|
|
|
|
# #region test_global_handler_unknown_client [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_global_handler_unknown_client(self):
|
|
from src.app import global_exception_handler
|
|
req = _make_mock_request()
|
|
req.client = None
|
|
response = await global_exception_handler(req, RuntimeError("no client"))
|
|
assert response.status_code == 500
|
|
# #endregion test_global_handler_unknown_client
|
|
|
|
# #region test_network_error_handler [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_network_error_handler(self):
|
|
from src.app import network_error_handler
|
|
from src.core.utils.network import NetworkError
|
|
request = _make_mock_request()
|
|
response = await network_error_handler(request, NetworkError("down"))
|
|
assert isinstance(response, HTTPException)
|
|
assert response.status_code == 503
|
|
assert "Environment unavailable" in response.detail
|
|
# #endregion test_network_error_handler
|
|
|
|
# #region test_global_handler_with_query_params [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_global_handler_with_query_params(self):
|
|
from src.app import global_exception_handler
|
|
req = _make_mock_request()
|
|
req.query_params = {"env_id": "prod"}
|
|
response = await global_exception_handler(req, Exception("test"))
|
|
assert response.status_code == 500
|
|
# #endregion test_global_handler_with_query_params
|
|
|
|
# #region test_internal_server_error_returns_json [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_internal_server_error_returns_json(self):
|
|
from src.app import global_exception_handler
|
|
request = _make_mock_request(method="POST", path="/api/crash")
|
|
response = await global_exception_handler(request, RuntimeError("crash"))
|
|
assert response.status_code == 500
|
|
body = json.loads(response.body)
|
|
assert body["path"] == "/api/crash"
|
|
# #endregion test_internal_server_error_returns_json
|
|
# #endregion Test.AppModule.Handlers
|