"""Unit tests for database module.""" import pytest from sqlalchemy.orm import Session from app.database import get_db, init_db, SessionLocal, Base, engine class TestDatabase: """Test database initialization and session management.""" def test_init_db_creates_tables(self): """Test that init_db creates all model tables.""" init_db() # Check that tables exist from app.models import User, Project, Task, LogEntry, ProjectMembership # Verify table names are in metadata assert "users" in Base.metadata.tables assert "projects" in Base.metadata.tables assert "tasks" in Base.metadata.tables assert "log_entries" in Base.metadata.tables assert "project_memberships" in Base.metadata.tables def test_session_local_returns_session(self): """Test SessionLocal creates a valid session.""" session = SessionLocal() assert isinstance(session, Session) session.close() def test_get_db_yields_session(self): """Test get_db generator yields a session and closes it.""" gen = get_db() db = next(gen) assert isinstance(db, Session) # Exhaust generator to trigger cleanup try: next(gen) except StopIteration: pass def test_get_db_closes_on_exception(self): """Test get_db closes session even on exception.""" gen = get_db() db = next(gen) # Simulate exception by closing manually try: gen.throw(Exception("test error")) except Exception: pass # Session should be closed (no error means cleanup happened) class TestDatabaseEngine: """Test database engine configuration.""" def test_engine_is_sqlite(self): """Test engine is configured for SQLite.""" assert "sqlite" in str(engine.url) def test_engine_allows_multithread(self): """Test SQLite is configured for multi-threaded access.""" # check_same_thread=False is set in database.py assert engine.pool is not None