Sergey Vanyushkin
4c5b6bd0a7
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%
140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
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
|