feat: add alembic migrations, integration tests, and CI support
- 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:
139
tests/integration/test_repositories.py
Normal file
139
tests/integration/test_repositories.py
Normal file
@@ -0,0 +1,139 @@
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.domain.entities import Post
|
||||
from app.domain.value_objects import Title
|
||||
from app.infrastructure.repositories.post import SQLAlchemyPostRepository
|
||||
|
||||
|
||||
class TestSQLAlchemyPostRepository:
|
||||
@pytest.fixture
|
||||
async def repo(self, db_session: AsyncSession) -> SQLAlchemyPostRepository:
|
||||
return SQLAlchemyPostRepository(db_session)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_creates_post(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
post = Post.create(
|
||||
title_str="Test Title",
|
||||
content_str="Test content with enough characters",
|
||||
author_id="user-1",
|
||||
tags=["test"],
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
result = await repo.get_by_id(post.id)
|
||||
assert result is not None
|
||||
assert result.title.value == "Test Title"
|
||||
assert result.slug.value == "test-title"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_by_id_returns_post(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
post = Post.create(
|
||||
title_str="Find Me",
|
||||
content_str="Content with enough characters for validation",
|
||||
author_id="user-1",
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
result = await repo.get_by_id(post.id)
|
||||
assert result is not None
|
||||
assert result.id == post.id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_by_id_returns_none_for_missing(self, repo: SQLAlchemyPostRepository) -> None:
|
||||
result = await repo.get_by_id(uuid4())
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_by_slug_returns_post(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
post = Post.create(
|
||||
title_str="Slug Test",
|
||||
content_str="Content with enough characters for validation",
|
||||
author_id="user-1",
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
result = await repo.get_by_slug(post.slug.value)
|
||||
assert result is not None
|
||||
assert result.id == post.id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_with_pagination(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
for i in range(5):
|
||||
post = Post.create(
|
||||
title_str=f"Post {i}",
|
||||
content_str="Content with enough characters for validation",
|
||||
author_id="user-1",
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
result = await repo.get_published(limit=3, offset=0)
|
||||
assert len(result) == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_modifies_post(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
post = Post.create(
|
||||
title_str="Original",
|
||||
content_str="Content with enough characters for validation",
|
||||
author_id="user-1",
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
post.update_title(Title("Updated"))
|
||||
await repo.update(post)
|
||||
await db_session.commit()
|
||||
|
||||
result = await repo.get_by_id(post.id)
|
||||
assert result is not None
|
||||
assert result.title.value == "Updated"
|
||||
assert result.slug.value == "updated"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_removes_post(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
post = Post.create(
|
||||
title_str="To Delete",
|
||||
content_str="Content with enough characters for validation",
|
||||
author_id="user-1",
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
await repo.delete(post.id)
|
||||
await db_session.commit()
|
||||
|
||||
result = await repo.get_by_id(post.id)
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_slug_exists_checks_uniqueness(
|
||||
self, repo: SQLAlchemyPostRepository, db_session: AsyncSession
|
||||
) -> None:
|
||||
post = Post.create(
|
||||
title_str="Unique Slug",
|
||||
content_str="Content with enough characters for validation",
|
||||
author_id="user-1",
|
||||
)
|
||||
await repo.add(post)
|
||||
await db_session.commit()
|
||||
|
||||
assert await repo.slug_exists(post.slug.value) is True
|
||||
assert await repo.slug_exists("nonexistent-slug") is False
|
||||
Reference in New Issue
Block a user