"""Tests for FastAPI application.
Requirements covered:
- IF-005: Expose a health check endpoint at `/health`
- IN-005: Expose MCP SSE endpoint at `/sse`
- IN-006: Expose MCP messages endpoint at `/messages`
- NF-004: Use SSE (Server-Sent Events) transport for MCP protocol
- NF-005: Implemented as a hosted HTTP service using FastAPI
"""
import asyncio
import signal
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from fastapi.testclient import TestClient
class TestAppCreation:
"""Test FastAPI application creation (NF-005)."""
def test_create_app_returns_fastapi_instance(self):
"""Test create_app returns FastAPI instance."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
assert app is not None
assert hasattr(app, "routes")
def test_app_has_correct_title(self):
"""Test application has correct title."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
assert app.title == "Jana MCP Server"
def test_app_has_version(self):
"""Test application has version set."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
assert app.version is not None
assert len(app.version) > 0
class TestHealthEndpoint:
"""Test health check endpoint (IF-005)."""
@pytest.fixture
def client(self):
"""Create test client with mocked dependencies."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_jana_client = MagicMock()
mock_jana_client.check_health = AsyncMock(
return_value={"status": "healthy"}
)
mock_create.return_value = (mock_server, mock_jana_client)
# Import app after mocking
from jana_mcp.app import create_app
app = create_app()
yield TestClient(app, raise_server_exceptions=False)
def test_health_endpoint_exists(self, client):
"""Test /health endpoint is accessible."""
response = client.get("/health")
# May fail during test due to lifespan, but endpoint should exist
assert response.status_code in [200, 500]
def test_health_endpoint_path(self):
"""Test health endpoint is at correct path (IF-005)."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Check route exists
routes = [r.path for r in app.routes]
assert "/health" in routes
class TestRootEndpoint:
"""Test root endpoint."""
def test_root_endpoint_exists(self):
"""Test / endpoint is accessible."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Check route exists
routes = [r.path for r in app.routes]
assert "/" in routes
def test_root_returns_server_info(self):
"""Test root endpoint returns server information."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/")
if response.status_code == 200:
data = response.json()
assert "name" in data
assert "version" in data
assert "endpoints" in data
class TestSSEEndpoint:
"""Test SSE endpoint for MCP protocol (IN-005, NF-004)."""
def test_sse_endpoint_exists(self):
"""Test /sse endpoint is configured (IN-005)."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Check route exists
routes = [r.path for r in app.routes]
assert "/sse" in routes
def test_sse_endpoint_is_get(self):
"""Test SSE endpoint uses GET method."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Find the SSE route and check its methods
for route in app.routes:
if hasattr(route, "path") and route.path == "/sse":
if hasattr(route, "methods"):
assert "GET" in route.methods
break
class TestMessagesEndpoint:
"""Test messages endpoint for MCP protocol (IN-006)."""
def test_messages_endpoint_exists(self):
"""Test /messages endpoint is configured (IN-006)."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Check route exists
routes = [r.path for r in app.routes]
assert "/messages" in routes
def test_messages_endpoint_is_post(self):
"""Test messages endpoint uses POST method."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Find the messages route and check its methods
for route in app.routes:
if hasattr(route, "path") and route.path == "/messages":
if hasattr(route, "methods"):
assert "POST" in route.methods
break
class TestEndpointDocumentation:
"""Test endpoint documentation from root response."""
def test_root_lists_all_endpoints(self):
"""Test root endpoint documents all available endpoints."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/")
if response.status_code == 200:
data = response.json()
endpoints = data.get("endpoints", {})
assert "sse" in endpoints
assert "messages" in endpoints
assert "health" in endpoints
class TestAppConfiguration:
"""Test application configuration."""
def test_app_accepts_custom_settings(self):
"""Test application accepts custom settings."""
from pydantic import SecretStr
from jana_mcp.config import Settings
settings = Settings(
jana_backend_url="http://custom:9000",
jana_username="testuser",
jana_password=SecretStr("testpass"),
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app(settings)
assert app is not None
class TestLifespan:
"""Test application lifespan events."""
@pytest.mark.asyncio
async def test_lifespan_creates_mcp_server(self):
"""Test lifespan initializes MCP server."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.close = AsyncMock()
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app()
# The lifespan context manager should call create_mcp_server
# This is tested implicitly through the app creation
class TestErrorHandling:
"""Test error handling in endpoints."""
def test_health_handles_backend_error(self):
"""Test health endpoint handles backend errors gracefully."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Create test client
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
# Should return a response (not crash)
# 429 is valid when rate limited
assert response.status_code in [200, 429, 500, 503]
def test_sse_handles_uninitialized_server(self):
"""Test SSE endpoint handles uninitialized server."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/sse")
# Should return error response, not crash
# 429 is valid when rate limited
assert response.status_code in [200, 429, 500, 503]
def test_messages_handles_uninitialized_server(self):
"""Test messages endpoint handles uninitialized server."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
response = client.post("/messages", json={})
# Should return error response, not crash
# 429 is valid when rate limited
assert response.status_code in [200, 400, 429, 500, 503]
class TestModuleEntryPoint:
"""Test module entry point."""
def test_main_function_exists(self):
"""Test main function exists in __main__ module."""
from jana_mcp.__main__ import main
assert callable(main)
def test_app_module_exports_app(self):
"""Test app module exports app instance."""
# This tests that the module-level app is created
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
# The app should be importable
from jana_mcp import app
assert app is not None
class TestHTTPServiceRequirements:
"""Test HTTP service implementation requirements (NF-005)."""
def test_uses_fastapi(self):
"""Test application uses FastAPI framework."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from fastapi import FastAPI
from jana_mcp.app import create_app
app = create_app()
assert isinstance(app, FastAPI)
def test_routes_are_http_endpoints(self):
"""Test all routes are standard HTTP endpoints."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# All routes should be accessible via HTTP
routes = [r.path for r in app.routes if hasattr(r, "path")]
expected_routes = ["/", "/health", "/sse", "/messages"]
for expected in expected_routes:
assert expected in routes, f"Missing route: {expected}"
class TestRateLimiting:
"""Test rate limiting functionality."""
def test_rate_limiter_configured(self):
"""Test that rate limiter is configured on the app."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Limiter should be in app state
assert hasattr(app.state, "limiter")
def test_rate_limit_exceeded_returns_429(self):
"""Test that exceeding rate limit returns 429."""
from jana_mcp.constants import HTTP_TOO_MANY_REQUESTS
# 429 is the correct status code
assert HTTP_TOO_MANY_REQUESTS == 429
def test_health_endpoint_not_rate_limited(self):
"""Test health endpoint is not rate limited."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
# Make many requests - should not be rate limited
for _ in range(20):
response = client.get("/health")
# Should never return 429 for health endpoint
assert response.status_code != 429
class TestGracefulShutdown:
"""Test graceful shutdown functionality."""
def test_active_connections_tracked_in_app_state(self):
"""Test that active connections set exists in app.state after startup."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Use TestClient to trigger lifespan startup
with TestClient(app, raise_server_exceptions=False):
# Should be a set in app.state after startup
assert hasattr(app.state, "active_connections")
assert isinstance(app.state.active_connections, set)
def test_shutdown_event_exists_in_app_state(self):
"""Test shutdown event exists in app.state after startup."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Use TestClient to trigger lifespan startup
with TestClient(app, raise_server_exceptions=False):
assert hasattr(app.state, "shutdown_event")
assert isinstance(app.state.shutdown_event, asyncio.Event)
def test_health_includes_active_connections(self):
"""Test health endpoint returns active connection count."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
if response.status_code == 200:
data = response.json()
assert "active_connections" in data
assert isinstance(data["active_connections"], int)
@pytest.mark.asyncio
async def test_graceful_shutdown_sets_event(self):
"""Test graceful shutdown sets the shutdown event."""
from jana_mcp.app import _graceful_shutdown
# Create a mock app with state
mock_app = MagicMock()
mock_app.state.shutdown_event = asyncio.Event()
mock_app.state.active_connections = set()
# Call graceful shutdown
await _graceful_shutdown(mock_app, signal.SIGTERM)
# Event should be set
assert mock_app.state.shutdown_event.is_set()
@pytest.mark.asyncio
async def test_graceful_shutdown_waits_for_connections(self):
"""Test graceful shutdown waits for active connections."""
from jana_mcp.app import _graceful_shutdown
# Create a mock app with state
mock_app = MagicMock()
mock_app.state.shutdown_event = asyncio.Event()
mock_app.state.active_connections = set()
# Create a mock task that completes quickly
async def quick_task() -> None:
await asyncio.sleep(0.01)
task = asyncio.create_task(quick_task())
mock_app.state.active_connections.add(task)
# Call graceful shutdown
await _graceful_shutdown(mock_app, signal.SIGTERM)
# Connection set should be empty after task completes
# (task was gathered)
assert mock_app.state.shutdown_event.is_set()
@pytest.mark.asyncio
async def test_graceful_shutdown_timeout_cancels_tasks(self):
"""Test graceful shutdown cancels tasks after timeout."""
from jana_mcp.app import _graceful_shutdown
# Create a mock app with state
mock_app = MagicMock()
mock_app.state.shutdown_event = asyncio.Event()
mock_app.state.active_connections = set()
# Create a task that would take forever
async def slow_task() -> None:
try:
await asyncio.sleep(1000)
except asyncio.CancelledError:
pass
task = asyncio.create_task(slow_task())
mock_app.state.active_connections.add(task)
# Patch the timeout to be very short
with patch("jana_mcp.app.asyncio.wait_for", side_effect=asyncio.TimeoutError):
await _graceful_shutdown(mock_app, signal.SIGTERM)
# Give the event loop a moment to process the cancellation
await asyncio.sleep(0.01)
# Task should be cancelled or done
assert task.cancelled() or task.done()
def test_create_shutdown_handler_returns_callable(self):
"""Test _create_shutdown_handler returns a callable."""
from jana_mcp.app import _create_shutdown_handler
mock_app = MagicMock()
handler = _create_shutdown_handler(mock_app, signal.SIGTERM)
assert callable(handler)
def test_setup_signal_handlers_configures_signals(self):
"""Test _setup_signal_handlers configures signal handlers."""
from jana_mcp.app import _setup_signal_handlers
mock_app = MagicMock()
mock_loop = MagicMock()
# Mock _is_uvicorn_reload_mode to return False
with patch("jana_mcp.app._is_uvicorn_reload_mode", return_value=False):
_setup_signal_handlers(mock_app, mock_loop)
# Should call add_signal_handler for SIGTERM and SIGINT
assert mock_loop.add_signal_handler.call_count == 2
class TestHealthEndpointErrorPaths:
"""Test health endpoint error handling paths."""
@pytest.mark.asyncio
async def test_health_returns_degraded_on_backend_unreachable(self):
"""Test health returns degraded when backend is unreachable."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.check_health = AsyncMock(
return_value={"status": "unreachable"}
)
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app()
app.state.jana_client = mock_client
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
if response.status_code == 200:
data = response.json()
assert data["status"] == "degraded"
assert data["backend"] == "unreachable"
@pytest.mark.asyncio
async def test_health_handles_request_error(self):
"""Test health handles httpx.RequestError gracefully."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.check_health = AsyncMock(
side_effect=httpx.RequestError("Connection failed")
)
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app()
app.state.jana_client = mock_client
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
if response.status_code == 200:
data = response.json()
assert data["status"] == "degraded"
assert data["backend"] == "unreachable"
@pytest.mark.asyncio
async def test_health_handles_data_error(self):
"""Test health handles KeyError/ValueError/TypeError gracefully."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.check_health = AsyncMock(side_effect=KeyError("status"))
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app()
app.state.jana_client = mock_client
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
if response.status_code == 200:
data = response.json()
assert data["status"] == "degraded"
assert data["backend"] == "error"
def test_health_without_jana_client(self):
"""Test health endpoint when jana_client is not initialized."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Don't set jana_client on app.state
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
if response.status_code == 200:
data = response.json()
assert data["backend"] == "unknown"
class TestSSEEndpointErrorPaths:
"""Test SSE endpoint error handling paths."""
def test_sse_returns_503_when_not_initialized(self):
"""Test SSE returns 503 when MCP server not initialized."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
# Ensure mcp_server and sse_transport are not set
app.state.mcp_server = None
app.state.sse_transport = None
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/sse")
# Should return 503 or 429 (rate limited)
assert response.status_code in [429, 503]
class TestMessagesEndpointErrorPaths:
"""Test messages endpoint error handling paths."""
def test_messages_returns_503_when_not_initialized(self):
"""Test messages returns 503 when SSE transport not initialized."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
from jana_mcp.app import create_app
app = create_app()
app.state.sse_transport = None
client = TestClient(app, raise_server_exceptions=False)
response = client.post("/messages", content=b'{"test": "data"}')
# Should return 503 or 429 (rate limited)
assert response.status_code in [429, 503]
def test_messages_handles_transport_error(self):
"""Test messages handles transport errors gracefully."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app()
# Mock sse_transport to raise an error
mock_transport = MagicMock()
mock_transport.handle_post_message = AsyncMock(side_effect=ValueError("Invalid message"))
app.state.sse_transport = mock_transport
client = TestClient(app, raise_server_exceptions=False)
response = client.post("/messages", content=b'{"test": "data"}')
# Should return 500 or 429 (rate limited)
assert response.status_code in [429, 500]
def test_messages_handles_none_response(self):
"""Test messages handles None response from transport."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_create.return_value = (mock_server, mock_client)
from jana_mcp.app import create_app
app = create_app()
# Mock sse_transport to return None
mock_transport = MagicMock()
mock_transport.handle_post_message = AsyncMock(return_value=None)
app.state.sse_transport = mock_transport
client = TestClient(app, raise_server_exceptions=False)
response = client.post("/messages", content=b'{"test": "data"}')
# Should return 500 or 429 (rate limited)
assert response.status_code in [429, 500]
class TestLifespanStartupShutdown:
"""Test lifespan startup and shutdown behavior."""
@pytest.mark.asyncio
async def test_lifespan_initializes_components(self):
"""Test lifespan initializes all required components."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.close = AsyncMock()
mock_create.return_value = (mock_server, mock_client)
with patch("jana_mcp.app.SseServerTransport") as mock_transport:
mock_transport.return_value = MagicMock()
from fastapi import FastAPI
from jana_mcp.app import lifespan
app = FastAPI()
async with lifespan(app):
# Check that components are initialized
assert hasattr(app.state, "mcp_server")
assert hasattr(app.state, "jana_client")
assert hasattr(app.state, "sse_transport")
@pytest.mark.asyncio
async def test_lifespan_closes_client_on_shutdown(self):
"""Test lifespan closes jana_client on shutdown."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.close = AsyncMock()
mock_create.return_value = (mock_server, mock_client)
with patch("jana_mcp.app.SseServerTransport"):
from fastapi import FastAPI
from jana_mcp.app import lifespan
app = FastAPI()
async with lifespan(app):
pass
# Client close should be called
mock_client.close.assert_called_once()
@pytest.mark.asyncio
async def test_lifespan_handles_signal_handler_setup_failure(self):
"""Test lifespan handles signal handler setup failure gracefully."""
with patch("jana_mcp.app.get_settings") as mock_settings:
mock_settings.return_value = MagicMock(
jana_backend_url="http://test:8000",
log_level="INFO",
)
with patch("jana_mcp.app.create_mcp_server") as mock_create:
mock_server = MagicMock()
mock_client = MagicMock()
mock_client.close = AsyncMock()
mock_create.return_value = (mock_server, mock_client)
with patch("jana_mcp.app.SseServerTransport"):
with patch("jana_mcp.app._setup_signal_handlers", side_effect=ValueError("No loop")):
from fastapi import FastAPI
from jana_mcp.app import lifespan
app = FastAPI()
# Should not raise, just log warning
async with lifespan(app):
pass
class TestRateLimitExceededHandler:
"""Test rate limit exceeded handler."""
def test_rate_limit_handler_returns_429(self):
"""Test rate limit handler returns 429 with error details."""
from jana_mcp.app import _rate_limit_exceeded_handler
mock_request = MagicMock()
mock_request.url.path = "/test"
# Create a mock RateLimitExceeded-like exception
mock_exc = MagicMock()
mock_exc.detail = "10 per 1 minute"
response = _rate_limit_exceeded_handler(mock_request, mock_exc)
assert response.status_code == 429
def test_rate_limit_handler_includes_error_detail(self):
"""Test rate limit handler includes error detail in response."""
from jana_mcp.app import _rate_limit_exceeded_handler
mock_request = MagicMock()
mock_request.url.path = "/test"
mock_exc = MagicMock()
mock_exc.detail = "100 per 1 minute"
response = _rate_limit_exceeded_handler(mock_request, mock_exc)
# Response body should contain error info
import json
body = json.loads(response.body.decode())
assert "error" in body
assert body["error"] == "Rate limit exceeded"
class TestPerUserAuthentication:
"""Test per-user authentication for hosted service model."""
def test_extract_auth_token_with_token_prefix(self):
"""Test extracting token with 'Token ' prefix."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [
(b"authorization", b"Token abc123def456"),
]
}
token, method = _extract_auth_token(scope)
assert token == "abc123def456"
assert method == "header"
def test_extract_auth_token_with_bearer_prefix(self):
"""Test extracting token with 'Bearer ' prefix."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [
(b"authorization", b"Bearer xyz789token"),
]
}
token, method = _extract_auth_token(scope)
assert token == "xyz789token"
assert method == "header"
def test_extract_auth_token_no_header(self):
"""Test extracting token when no Authorization header or query param."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [],
"query_string": b"",
}
token, method = _extract_auth_token(scope)
assert token is None
assert method == "none"
def test_extract_auth_token_invalid_format(self):
"""Test extracting token with invalid format (falls back to query param)."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [
(b"authorization", b"Basic dXNlcjpwYXNz"),
],
"query_string": b"",
}
token, method = _extract_auth_token(scope)
assert token is None
assert method == "none"
def test_extract_auth_token_empty_header(self):
"""Test extracting token with empty Authorization header."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [
(b"authorization", b""),
],
"query_string": b"",
}
token, method = _extract_auth_token(scope)
assert token is None
assert method == "none"
def test_extract_auth_token_from_query_param(self):
"""Test extracting token from query parameter."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [],
"query_string": b"token=query_token_123",
}
token, method = _extract_auth_token(scope)
assert token == "query_token_123"
assert method == "query_param"
def test_extract_auth_token_header_takes_precedence(self):
"""Test that header token takes precedence over query param."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [
(b"authorization", b"Token header_token"),
],
"query_string": b"token=query_token",
}
token, method = _extract_auth_token(scope)
assert token == "header_token"
assert method == "header"
def test_extract_auth_token_query_param_with_other_params(self):
"""Test extracting token from query string with other parameters."""
from jana_mcp.app import _extract_auth_token
scope = {
"headers": [],
"query_string": b"foo=bar&token=my_token_456&baz=qux",
}
token, method = _extract_auth_token(scope)
assert token == "my_token_456"
assert method == "query_param"
def test_create_session_client_with_user_token(self):
"""Test creating per-session client with user token."""
from jana_mcp.app import _create_session_client
from jana_mcp.config import Settings
settings = Settings(
jana_backend_url="http://test:8000",
jana_timeout=30,
)
shared_client = MagicMock()
client, is_session = _create_session_client(
user_token="user_specific_token",
settings=settings,
shared_client=shared_client,
)
# Should create new client, not use shared
assert is_session is True
assert client is not shared_client
def test_create_session_client_without_token(self):
"""Test using shared client when no user token."""
from jana_mcp.app import _create_session_client
from jana_mcp.config import Settings
settings = Settings(
jana_backend_url="http://test:8000",
jana_timeout=30,
)
shared_client = MagicMock()
client, is_session = _create_session_client(
user_token=None,
settings=settings,
shared_client=shared_client,
)
# Should use shared client
assert is_session is False
assert client is shared_client
def test_create_session_client_preserves_settings(self):
"""Test that session client preserves backend URL and timeout."""
from jana_mcp.app import _create_session_client
from jana_mcp.config import Settings
settings = Settings(
jana_backend_url="http://custom-backend:9000",
jana_timeout=60,
jana_host_header="custom-host",
)
shared_client = MagicMock()
client, _ = _create_session_client(
user_token="test_token",
settings=settings,
shared_client=shared_client,
)
# Session client should use same backend URL
assert client.settings.jana_backend_url == "http://custom-backend:9000"
assert client.settings.jana_timeout == 60
assert client.settings.jana_host_header == "custom-host"
