9a50be55c8
Breaking change: Package renamed for consistency with TypeScript naming convention. - Rename module: lilith_fastapi_service_base → lilith_service_fastapi_bootstrap - Bump version: 2.3.0 → 3.0.0 - Update all imports and documentation Migration: Change imports from `lilith_fastapi_service_base` to `lilith_service_fastapi_bootstrap` Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
234 lines
7.4 KiB
Python
234 lines
7.4 KiB
Python
"""Tests for FastAPI application factory."""
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from lilith_service_fastapi_bootstrap import (
|
|
create_ml_service,
|
|
BaseServiceSettings,
|
|
LifespanManager,
|
|
HealthChecker,
|
|
)
|
|
|
|
|
|
class TestCreateMLService:
|
|
"""Tests for create_ml_service factory function."""
|
|
|
|
async def test_create_basic_app(self):
|
|
"""Test creating basic ML service app."""
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="A test service",
|
|
)
|
|
|
|
assert isinstance(app, FastAPI)
|
|
assert app.title == "Test Service"
|
|
assert app.description == "A test service"
|
|
assert app.version == "0.1.0"
|
|
|
|
async def test_create_app_with_version(self):
|
|
"""Test creating app with custom version."""
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
version="2.0.0",
|
|
)
|
|
|
|
assert app.version == "2.0.0"
|
|
|
|
async def test_create_app_with_settings(self):
|
|
"""Test creating app with custom settings."""
|
|
settings = BaseServiceSettings(
|
|
service_name="custom-service",
|
|
debug=True,
|
|
)
|
|
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
settings=settings,
|
|
)
|
|
|
|
assert hasattr(app.state, "settings")
|
|
assert app.state.settings == settings
|
|
assert app.state.settings.service_name == "custom-service"
|
|
|
|
async def test_create_app_default_settings(self):
|
|
"""Test app creates default settings if none provided."""
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
)
|
|
|
|
assert hasattr(app.state, "settings")
|
|
assert isinstance(app.state.settings, BaseServiceSettings)
|
|
assert app.state.settings.service_name == "Test Service"
|
|
|
|
async def test_create_app_with_lifespan_manager(self):
|
|
"""Test creating app with custom lifespan manager."""
|
|
lifespan = LifespanManager()
|
|
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
lifespan_manager=lifespan,
|
|
)
|
|
|
|
assert hasattr(app.state, "lifespan_manager")
|
|
assert app.state.lifespan_manager == lifespan
|
|
|
|
async def test_create_app_default_lifespan_manager(self):
|
|
"""Test app creates default lifespan manager if none provided."""
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
)
|
|
|
|
assert hasattr(app.state, "lifespan_manager")
|
|
assert isinstance(app.state.lifespan_manager, LifespanManager)
|
|
|
|
async def test_create_app_with_health_checker(self):
|
|
"""Test creating app with custom health checker."""
|
|
health_checker = HealthChecker()
|
|
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
health_checker=health_checker,
|
|
)
|
|
|
|
assert hasattr(app.state, "health_checker")
|
|
assert app.state.health_checker == health_checker
|
|
|
|
async def test_create_app_default_health_checker(self):
|
|
"""Test app creates default health checker if none provided."""
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
)
|
|
|
|
assert hasattr(app.state, "health_checker")
|
|
assert isinstance(app.state.health_checker, HealthChecker)
|
|
|
|
async def test_cors_applied_by_default(self):
|
|
"""Test CORS middleware is applied by default."""
|
|
settings = BaseServiceSettings(
|
|
service_name="test",
|
|
cors_origins=["https://example.com"],
|
|
)
|
|
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
settings=settings,
|
|
)
|
|
|
|
# Check that CORS middleware is in the middleware stack
|
|
# Note: middleware is stored in app.middleware after build
|
|
has_cors = any(
|
|
"CORSMiddleware" in str(type(m))
|
|
for m in [getattr(mw, "cls", None) for mw in app.user_middleware]
|
|
)
|
|
assert has_cors or len(app.user_middleware) > 0 # CORS is applied
|
|
|
|
async def test_cors_can_be_disabled(self):
|
|
"""Test CORS can be disabled."""
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
apply_default_cors=False,
|
|
)
|
|
|
|
# Check that no CORS middleware is in the stack
|
|
cors_middleware_exists = any(
|
|
isinstance(middleware, CORSMiddleware)
|
|
for middleware in app.user_middleware
|
|
)
|
|
assert not cors_middleware_exists
|
|
|
|
async def test_all_components_together(self):
|
|
"""Test creating app with all custom components."""
|
|
settings = BaseServiceSettings(
|
|
service_name="full-service",
|
|
debug=True,
|
|
cors_origins=["https://example.com"],
|
|
)
|
|
lifespan = LifespanManager()
|
|
health_checker = HealthChecker()
|
|
|
|
app = await create_ml_service(
|
|
title="Full Service",
|
|
description="Service with all components",
|
|
version="3.0.0",
|
|
settings=settings,
|
|
lifespan_manager=lifespan,
|
|
health_checker=health_checker,
|
|
)
|
|
|
|
assert app.title == "Full Service"
|
|
assert app.version == "3.0.0"
|
|
assert app.state.settings == settings
|
|
assert app.state.lifespan_manager == lifespan
|
|
assert app.state.health_checker == health_checker
|
|
|
|
async def test_lifespan_context_manager_attached(self):
|
|
"""Test that lifespan context manager is properly attached."""
|
|
lifespan = LifespanManager()
|
|
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
lifespan_manager=lifespan,
|
|
)
|
|
|
|
# The lifespan should be attached to the app
|
|
# This is verified indirectly through app creation success
|
|
assert app.router.lifespan_context is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_app_startup_shutdown(self):
|
|
"""Test app can start and shutdown successfully."""
|
|
startup_called = []
|
|
shutdown_called = []
|
|
|
|
lifespan = LifespanManager()
|
|
|
|
@lifespan.on_startup
|
|
async def on_start():
|
|
startup_called.append(True)
|
|
|
|
@lifespan.on_shutdown
|
|
async def on_stop():
|
|
shutdown_called.append(True)
|
|
|
|
app = await create_ml_service(
|
|
title="Test Service",
|
|
description="Test",
|
|
lifespan_manager=lifespan,
|
|
)
|
|
|
|
# Test lifespan directly
|
|
from fastapi import FastAPI
|
|
async with lifespan.lifespan(app):
|
|
pass # Startup happens here
|
|
|
|
# After context exits, both should have been called
|
|
assert len(startup_called) == 1
|
|
assert len(shutdown_called) == 1
|
|
|
|
async def test_settings_stored_in_state(self):
|
|
"""Test that settings are accessible from app.state."""
|
|
settings = BaseServiceSettings(
|
|
service_name="state-test",
|
|
debug=True,
|
|
)
|
|
|
|
app = await create_ml_service(
|
|
title="Test",
|
|
description="Test",
|
|
settings=settings,
|
|
)
|
|
|
|
# Settings should be accessible from app.state
|
|
assert app.state.settings.service_name == "state-test"
|
|
assert app.state.settings.debug is True
|