ruwiki-test/tests/test_services.py

302 lines
11 KiB
Python

import asyncio
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from src.models import ProcessingResult, SimplifyCommand
from src.models.article_dto import ArticleDTO, ArticleStatus
from src.services import ArticleRepository, AsyncWriteQueue, DatabaseService, SimplifyService
class TestDatabaseService:
async def test_database_initialization(self, database_service: DatabaseService):
health = await database_service.health_check()
assert health is True
async def test_get_connection(self, database_service: DatabaseService):
async with await database_service.get_connection() as conn:
cursor = await conn.execute("SELECT 1")
result = await cursor.fetchone()
assert result[0] == 1
async def test_health_check_success(self, database_service: DatabaseService):
result = await database_service.health_check()
assert result is True
async def test_health_check_failure(self, test_config):
test_config.db_path = "/invalid/path/database.db"
service = DatabaseService(test_config)
result = await service.health_check()
assert result is False
class TestArticleRepository:
async def test_create_article(self, repository: ArticleRepository):
article = await repository.create_article(
url="https://ru.ruwiki.ru/wiki/Test",
title="Test Article",
raw_text="Test content",
)
assert article.id is not None
assert article.url == "https://ru.ruwiki.ru/wiki/Test"
assert article.title == "Test Article"
assert article.raw_text == "Test content"
assert article.status == ArticleStatus.PENDING
assert article.simplified_text is None
async def test_create_duplicate_article(self, repository: ArticleRepository):
url = "https://ru.ruwiki.ru/wiki/Test"
await repository.create_article(
url=url,
title="Test Article",
raw_text="Test content",
)
with pytest.raises(ValueError, match="уже существует"):
await repository.create_article(
url=url,
title="Duplicate",
raw_text="Different content",
)
async def test_get_by_id(self, repository: ArticleRepository, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
retrieved = await repository.get_by_id(article.id)
assert retrieved is not None
assert retrieved.id == article.id
assert retrieved.url == article.url
async def test_get_by_id_not_found(self, repository: ArticleRepository):
result = await repository.get_by_id(99999)
assert result is None
async def test_get_by_url(self, repository: ArticleRepository, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
retrieved = await repository.get_by_url(article.url)
assert retrieved is not None
assert retrieved.id == article.id
assert retrieved.url == article.url
async def test_get_by_url_not_found(self, repository: ArticleRepository):
result = await repository.get_by_url("https://ru.ruwiki.ru/wiki/NonExistent")
assert result is None
async def test_update_article(self, repository: ArticleRepository, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
article.simplified_text = "Simplified content"
article.status = ArticleStatus.SIMPLIFIED
updated = await repository.update_article(article)
assert updated.simplified_text == "Simplified content"
assert updated.status == ArticleStatus.SIMPLIFIED
assert updated.updated_at is not None
async def test_update_nonexistent_article(self, repository: ArticleRepository):
article = ArticleDTO(
id=99999,
url="https://ru.ruwiki.ru/wiki/Test",
title="Test",
raw_text="Content",
status=ArticleStatus.PENDING,
created_at=datetime.now(timezone.utc),
)
with pytest.raises(ValueError, match="не найдена"):
await repository.update_article(article)
async def test_get_articles_by_status(self, repository: ArticleRepository):
article1 = await repository.create_article(
url="https://ru.ruwiki.ru/wiki/Test1",
title="Test 1",
raw_text="Content 1",
)
article2 = await repository.create_article(
url="https://ru.ruwiki.ru/wiki/Test2",
title="Test 2",
raw_text="Content 2",
)
article2.status = ArticleStatus.SIMPLIFIED
await repository.update_article(article2)
pending = await repository.get_articles_by_status(ArticleStatus.PENDING)
assert len(pending) == 1
assert pending[0].id == article1.id
simplified = await repository.get_articles_by_status(ArticleStatus.SIMPLIFIED)
assert len(simplified) == 1
assert simplified[0].id == article2.id
async def test_count_by_status(self, repository: ArticleRepository):
count = await repository.count_by_status(ArticleStatus.PENDING)
assert count == 0
await repository.create_article(
url="https://ru.ruwiki.ru/wiki/Test1",
title="Test 1",
raw_text="Content 1",
)
await repository.create_article(
url="https://ru.ruwiki.ru/wiki/Test2",
title="Test 2",
raw_text="Content 2",
)
count = await repository.count_by_status(ArticleStatus.PENDING)
assert count == 2
async def test_delete_article(self, repository: ArticleRepository, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
deleted = await repository.delete_article(article.id)
assert deleted is True
retrieved = await repository.get_by_id(article.id)
assert retrieved is None
async def test_delete_nonexistent_article(self, repository: ArticleRepository):
deleted = await repository.delete_article(99999)
assert deleted is False
class TestAsyncWriteQueue:
@pytest_asyncio.fixture
async def write_queue(self, repository: ArticleRepository) -> AsyncWriteQueue:
queue = AsyncWriteQueue(repository, max_batch_size=2)
await queue.start()
yield queue
await queue.stop()
async def test_write_queue_startup_shutdown(self, repository: ArticleRepository):
queue = AsyncWriteQueue(repository)
await queue.start()
assert queue._worker_task is not None
assert not queue._worker_task.done()
await queue.stop()
assert queue._worker_task.done()
async def test_update_article_operation(self, write_queue: AsyncWriteQueue, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
article.simplified_text = "Updated content"
await write_queue.update_article(article)
await asyncio.sleep(0.2)
retrieved = await write_queue.repository.get_by_id(article.id)
assert retrieved.simplified_text == "Updated content"
async def test_update_from_result_success(self, write_queue: AsyncWriteQueue, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
result = ProcessingResult.success_result(
url=article.url,
title=article.title,
raw_text=article.raw_text,
simplified_text="Processed content",
token_count_raw=100,
token_count_simplified=50,
processing_time_seconds=1.0,
)
updated_article = await write_queue.update_from_result(result)
assert updated_article.simplified_text == "Processed content"
assert updated_article.status == ArticleStatus.SIMPLIFIED
async def test_update_from_result_failure(self, write_queue: AsyncWriteQueue, sample_article_in_db: ArticleDTO):
article = sample_article_in_db
result = ProcessingResult.failure_result(
url=article.url,
error_message="Processing failed",
)
updated_article = await write_queue.update_from_result(result)
assert updated_article.status == ArticleStatus.FAILED
async def test_queue_stats(self, write_queue: AsyncWriteQueue):
stats = write_queue.stats
assert "total_operations" in stats
assert "failed_operations" in stats
assert "success_rate" in stats
class TestSimplifyService:
@pytest_asyncio.fixture
async def simplify_service(self, test_config, repository: ArticleRepository) -> SimplifyService:
ruwiki_adapter = AsyncMock()
llm_adapter = AsyncMock()
write_queue = AsyncMock()
service = SimplifyService(
config=test_config,
ruwiki_adapter=ruwiki_adapter,
llm_adapter=llm_adapter,
repository=repository,
write_queue=write_queue,
)
return service
async def test_get_prompt_template(self, simplify_service: SimplifyService):
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.read_text", return_value="Test prompt"):
template = await simplify_service.get_prompt_template()
assert template == "Test prompt"
async def test_check_existing_article(self, simplify_service: SimplifyService):
existing_article = ArticleDTO(
id=1,
url="https://ru.ruwiki.ru/wiki/Test",
title="Test",
raw_text="Content",
simplified_text="Simplified",
status=ArticleStatus.SIMPLIFIED,
created_at=datetime.now(timezone.utc),
)
simplify_service.repository.get_by_url = AsyncMock(return_value=existing_article)
result = await simplify_service._check_existing_article("https://ru.ruwiki.ru/wiki/Test")
assert result is not None
assert result.success is True
assert result.simplified_text == "Simplified"
async def test_check_existing_article_not_found(self, simplify_service: SimplifyService):
simplify_service.repository.get_by_url = AsyncMock(return_value=None)
result = await simplify_service._check_existing_article("https://ru.ruwiki.ru/wiki/Test")
assert result is None
async def test_health_check(self, simplify_service: SimplifyService):
simplify_service.ruwiki_adapter.health_check = AsyncMock()
simplify_service.llm_adapter.health_check = AsyncMock()
checks = await simplify_service.health_check()
assert "ruwiki" in checks
assert "llm" in checks
assert "prompt_template" in checks