Bruno Charest
cc99fea20a
- Implement tests for database generator to ensure proper session handling. - Create tests for EXIF extraction and conversion functions. - Add tests for image-related endpoints, ensuring proper data retrieval and isolation between clients. - Develop tests for OCR functionality, including language detection and text extraction. - Introduce tests for the image processing pipeline, covering success and failure scenarios. - Validate rate limiting functionality and ensure independent counters for different clients. - Implement scraper tests to verify HTML content fetching and error handling. - Add unit tests for various services, including storage and filename generation. - Establish worker entry point for ARQ to handle background image processing tasks.
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
import pytest
|
|
import hashlib
|
|
from sqlalchemy import select
|
|
from app.database import init_db
|
|
from app.models.client import APIClient
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_default_client():
|
|
# Mock engine and AsyncSessionLocal
|
|
with patch("app.database.engine") as mock_engine, \
|
|
patch("app.database.AsyncSessionLocal") as mock_session_factory:
|
|
|
|
# Mock engine.begin() context manager
|
|
mock_conn = AsyncMock()
|
|
mock_engine.begin.return_value.__aenter__.return_value = mock_conn
|
|
|
|
# Mock session behavior
|
|
mock_session = AsyncMockSession()
|
|
mock_session_factory.return_value.__aenter__.return_value = mock_session
|
|
|
|
# Simulate empty table
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
await init_db()
|
|
|
|
# Check if add was called for bootstrap client
|
|
assert any(isinstance(call.args[0], APIClient) for call in mock_session.add.call_args_list)
|
|
|
|
class AsyncMockSession:
|
|
def __init__(self):
|
|
self.add = MagicMock()
|
|
self.commit = AsyncMock()
|
|
self.execute = AsyncMock()
|
|
self.close = AsyncMock()
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *args): pass
|
|
|
|
class AsyncMock(MagicMock):
|
|
async def __call__(self, *args, **kwargs):
|
|
return super(AsyncMock, self).__call__(*args, **kwargs)
|
|
def __await__(self):
|
|
return self().__await__()
|