1b0d291ad3
- Implemented tests for player API endpoints including play, pause, stop, next, previous, seek, set volume, set mode, reload playlist, and get state. - Added mock player service for testing API interactions. - Created tests for player service methods including play, pause, stop playback, next, previous, seek, set volume, set mode, and reload playlist. - Ensured proper handling of success, error, and edge cases in both API and service tests. - Verified state management and serialization in player state tests.
72 lines
1.9 KiB
Python
72 lines
1.9 KiB
Python
from collections.abc import AsyncGenerator
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
|
|
from sqlmodel import SQLModel
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.core.logging import get_logger
|
|
from app.core.seeds import seed_all_data
|
|
from app.models import ( # noqa: F401
|
|
extraction,
|
|
plan,
|
|
playlist,
|
|
playlist_sound,
|
|
sound,
|
|
sound_played,
|
|
user,
|
|
user_oauth,
|
|
)
|
|
|
|
engine: AsyncEngine = create_async_engine(
|
|
settings.DATABASE_URL,
|
|
echo=settings.DATABASE_ECHO,
|
|
)
|
|
|
|
|
|
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
|
"""Get a database session for dependency injection."""
|
|
logger = get_logger(__name__)
|
|
async with AsyncSession(engine) as session:
|
|
try:
|
|
yield session
|
|
except Exception:
|
|
logger.exception("Database session error")
|
|
await session.rollback()
|
|
raise
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
def get_session_factory():
|
|
"""Get a session factory function for services."""
|
|
def session_factory():
|
|
return AsyncSession(engine)
|
|
return session_factory
|
|
|
|
|
|
async def init_db() -> None:
|
|
"""Initialize the database and create tables if they do not exist."""
|
|
logger = get_logger(__name__)
|
|
try:
|
|
logger.info("Initializing database tables")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.create_all)
|
|
logger.info("Database tables created successfully")
|
|
|
|
# Seed initial data
|
|
await seed_initial_data()
|
|
|
|
except Exception:
|
|
logger.exception("Failed to initialize database")
|
|
raise
|
|
|
|
|
|
async def seed_initial_data() -> None:
|
|
"""Seed initial data into the database."""
|
|
logger = get_logger(__name__)
|
|
logger.info("Starting initial data seeding")
|
|
|
|
async with AsyncSession(engine) as session:
|
|
await seed_all_data(session)
|