feat: add alembic migrations and PostgreSQL CI support
Some checks failed
ci/woodpecker/pr/pipeline Pipeline failed
Some checks failed
ci/woodpecker/pr/pipeline Pipeline failed
- Add alembic dependency and initialize migration framework - Configure async alembic env.py for SQLAlchemy 2.0 - Create initial migration for PostORM table - Gate init_db() with SKIP_INIT_DB env var for CI/production - Add PostgreSQL service to Woodpecker CI pipeline - Create integration tests for migrations (TC-INT-001..002) - Create integration tests for SQLAlchemyPostRepository (TC-INT-003..009) - Add unit test for init_db skip behavior (TC-UNIT-901) - All 176 tests pass, coverage 72.59%
This commit is contained in:
45
tests/integration/conftest.py
Normal file
45
tests/integration/conftest.py
Normal file
@@ -0,0 +1,45 @@
|
||||
import asyncio
|
||||
import os
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Generator
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from app.infrastructure.config import settings
|
||||
from app.infrastructure.database.models import Base
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop() -> Generator[asyncio.AbstractEventLoop]:
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
async def setup_database() -> AsyncGenerator[None]:
|
||||
db_url = os.environ.get("DB_URL", settings.database_url)
|
||||
test_engine = create_async_engine(db_url)
|
||||
|
||||
async with test_engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
yield
|
||||
|
||||
async with test_engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
|
||||
await test_engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def db_session() -> AsyncGenerator[AsyncSession]:
|
||||
db_url = os.environ.get("DB_URL", settings.database_url)
|
||||
test_engine = create_async_engine(db_url)
|
||||
session_factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
||||
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
await test_engine.dispose()
|
||||
Reference in New Issue
Block a user