MCP Market Statistics Server

"""데이터베이스 연결 및 관리 테스트""" 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