"""데이터베이스 연결 및 관리 테스트"""
import pytest
import asyncio
from unittest.mock import Mock, patch, AsyncMock
from src.utils.database import DatabaseManager
from src.exceptions import DatabaseConnectionError
class TestDatabaseManager:
"""데이터베이스 매니저 테스트"""
@pytest.fixture
def db_config(self):
"""테스트용 데이터베이스 설정"""
return {
"host": "localhost",
"port": 5432,
"name": "test_db",
"user": "test_user",
"password": "test_pass"
}
@pytest.fixture
def db_manager(self, db_config):
"""테스트용 데이터베이스 매니저"""
return DatabaseManager(db_config)
def test_database_manager_initialization(self, db_manager, db_config):
"""데이터베이스 매니저 초기화 테스트"""
assert db_manager.config == db_config
assert db_manager.pool is None
assert not db_manager.is_connected
def test_connection_url_generation(self, db_manager):
"""연결 URL 생성 테스트"""
expected_url = "postgresql://test_user:test_pass@localhost:5432/test_db"
assert db_manager.connection_url == expected_url
@pytest.mark.asyncio
async def test_successful_connection(self, db_manager):
"""성공적인 데이터베이스 연결 테스트"""
mock_pool = AsyncMock()
mock_pool._closed = False
with patch('src.utils.database.asyncpg.create_pool', new_callable=AsyncMock) as mock_create_pool:
mock_create_pool.return_value = mock_pool
await db_manager.connect()
assert db_manager.pool == mock_pool
assert db_manager.is_connected
mock_create_pool.assert_called_once()
@pytest.mark.asyncio
async def test_connection_failure(self, db_manager):
"""데이터베이스 연결 실패 테스트"""
with patch('src.utils.database.asyncpg.create_pool', new_callable=AsyncMock) as mock_create_pool:
mock_create_pool.side_effect = Exception("Connection failed")
with pytest.raises(DatabaseConnectionError):
await db_manager.connect()
assert db_manager.pool is None
assert not db_manager.is_connected
@pytest.mark.asyncio
async def test_disconnect(self, db_manager):
"""데이터베이스 연결 해제 테스트"""
# 먼저 연결
mock_pool = AsyncMock()
mock_pool._closed = False
with patch('src.utils.database.asyncpg.create_pool', new_callable=AsyncMock) as mock_create_pool:
mock_create_pool.return_value = mock_pool
await db_manager.connect()
# 연결 해제
await db_manager.disconnect()
mock_pool.close.assert_called_once()
assert db_manager.pool is None
assert not db_manager.is_connected
@pytest.mark.asyncio
async def test_execute_query(self, db_manager):
"""쿼리 실행 테스트"""
with patch('src.utils.database.asyncpg.create_pool', new_callable=AsyncMock) as mock_create_pool:
mock_pool = AsyncMock()
mock_pool._closed = False
mock_connection = AsyncMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_connection
# Mock record objects that behave like dicts
mock_record = Mock()
mock_record.__iter__ = Mock(return_value=iter([("id", 1), ("name", "test")]))
mock_connection.fetch.return_value = [mock_record]
mock_create_pool.return_value = mock_pool
await db_manager.connect()
result = await db_manager.fetch_all("SELECT * FROM test")
assert result == [{"id": 1, "name": "test"}]
mock_connection.fetch.assert_called_once_with("SELECT * FROM test")
@pytest.mark.asyncio
async def test_execute_query_without_connection(self, db_manager):
"""연결 없이 쿼리 실행 시 예외 테스트"""
with pytest.raises(DatabaseConnectionError):
await db_manager.fetch_all("SELECT * FROM test")
@pytest.mark.asyncio
async def test_health_check_success(self, db_manager):
"""헬스체크 성공 테스트"""
with patch('src.utils.database.asyncpg.create_pool', new_callable=AsyncMock) as mock_create_pool:
mock_pool = AsyncMock()
mock_pool._closed = False
mock_connection = AsyncMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_connection
mock_connection.fetchval.return_value = 1
mock_create_pool.return_value = mock_pool
await db_manager.connect()
is_healthy = await db_manager.health_check()
assert is_healthy is True
mock_connection.fetchval.assert_called_once_with("SELECT 1")
@pytest.mark.asyncio
async def test_health_check_failure(self, db_manager):
"""헬스체크 실패 테스트"""
with patch('src.utils.database.asyncpg.create_pool', new_callable=AsyncMock) as mock_create_pool:
mock_pool = AsyncMock()
mock_pool._closed = False
mock_connection = AsyncMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_connection
mock_connection.fetchval.side_effect = Exception("DB error")
mock_create_pool.return_value = mock_pool
await db_manager.connect()
is_healthy = await db_manager.health_check()
assert is_healthy is False
