79 lines
1.9 KiB
Python
79 lines
1.9 KiB
Python
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.repository.auditlog_repository import AuditLogRepository
|
|
|
|
|
|
def _result_with_scalars(items):
|
|
result = MagicMock()
|
|
scalars = MagicMock()
|
|
scalars.first.return_value = items[0] if items else None
|
|
scalars.all.return_value = items
|
|
result.scalars.return_value = scalars
|
|
return result
|
|
|
|
|
|
def _result_with_scalar_one(value):
|
|
result = MagicMock()
|
|
result.scalar_one.return_value = value
|
|
return result
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db():
|
|
return AsyncMock(spec=AsyncSession)
|
|
|
|
|
|
@pytest.fixture
|
|
def repository(mock_db):
|
|
return AuditLogRepository(mock_db)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_returns_audit_log(repository, mock_db):
|
|
row = MagicMock(id=10, user_id=1, event_type="WRITE")
|
|
mock_db.execute.return_value = _result_with_scalars([row])
|
|
|
|
result = await repository.get(10)
|
|
|
|
assert result is row
|
|
mock_db.execute.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_returns_none_for_missing_id(repository, mock_db):
|
|
mock_db.execute.return_value = _result_with_scalars([])
|
|
|
|
result = await repository.get(999999)
|
|
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_all_applies_filters(repository, mock_db):
|
|
row = MagicMock(id=11, user_id=2, org_unit_id=7, event_type="UPDATE")
|
|
mock_db.execute.side_effect = [
|
|
_result_with_scalars([row]),
|
|
_result_with_scalar_one(1),
|
|
]
|
|
|
|
result, count = await repository.get_all(
|
|
limit=10,
|
|
offset=5,
|
|
user_id=2,
|
|
org_unit_id=7,
|
|
task_id=3,
|
|
form_id=4,
|
|
event_type="UPDATE",
|
|
date_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
|
date_to=datetime(2026, 12, 31, tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert result == [row]
|
|
assert count == 1
|
|
assert mock_db.execute.await_count == 2
|
|
|