347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""Unit тесты для обработки ошибок в интерфейсах."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import aiohttp
|
|
import httpx
|
|
import pytest
|
|
|
|
from dataloader.interfaces.gmap2_brief.interface import (
|
|
Gmap2BriefConnectionError,
|
|
Gmap2BriefInterface,
|
|
)
|
|
from dataloader.interfaces.tenera.interface import (
|
|
SuperTeneraConnectionError,
|
|
SuperTeneraInterface,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGmap2BriefErrorHandling:
|
|
"""Тесты обработки ошибок в Gmap2BriefInterface."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_export_handles_http_status_error(self):
|
|
"""Тест обработки HTTP статус ошибки в start_export."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.APP_CONFIG") as mock_config:
|
|
mock_config.gmap2brief.base_url = "http://test.com"
|
|
mock_config.gmap2brief.start_endpoint = "/start"
|
|
mock_config.gmap2brief.timeout = 30
|
|
mock_config.app.local = False
|
|
|
|
interface = Gmap2BriefInterface(mock_logger)
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 500
|
|
mock_response.text = "Internal Server Error"
|
|
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
|
"Server error", request=MagicMock(), response=mock_response
|
|
)
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.post.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
with pytest.raises(Gmap2BriefConnectionError) as exc_info:
|
|
await interface.start_export()
|
|
|
|
assert "Failed to start export" in str(exc_info.value)
|
|
assert "500" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_export_handles_request_error(self):
|
|
"""Тест обработки сетевой ошибки в start_export."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.APP_CONFIG") as mock_config:
|
|
mock_config.gmap2brief.base_url = "http://test.com"
|
|
mock_config.gmap2brief.start_endpoint = "/start"
|
|
mock_config.gmap2brief.timeout = 30
|
|
mock_config.app.local = False
|
|
|
|
interface = Gmap2BriefInterface(mock_logger)
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.post.side_effect = httpx.ConnectError("Connection refused")
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
with pytest.raises(Gmap2BriefConnectionError) as exc_info:
|
|
await interface.start_export()
|
|
|
|
assert "Request error" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_handles_http_error(self):
|
|
"""Тест обработки HTTP ошибки в get_status."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.APP_CONFIG") as mock_config:
|
|
mock_config.gmap2brief.base_url = "http://test.com"
|
|
mock_config.gmap2brief.status_endpoint = "/status/{job_id}"
|
|
mock_config.gmap2brief.timeout = 30
|
|
mock_config.app.local = False
|
|
|
|
interface = Gmap2BriefInterface(mock_logger)
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.httpx.AsyncClient") as mock_client_cls:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
mock_response.text = "Not Found"
|
|
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
|
"Not found", request=MagicMock(), response=mock_response
|
|
)
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
with pytest.raises(Gmap2BriefConnectionError) as exc_info:
|
|
await interface.get_status("job123")
|
|
|
|
assert "Failed to get status" in str(exc_info.value)
|
|
assert "404" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_export_handles_error(self):
|
|
"""Тест обработки ошибки в download_export."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.APP_CONFIG") as mock_config:
|
|
mock_config.gmap2brief.base_url = "http://test.com"
|
|
mock_config.gmap2brief.download_endpoint = "/download/{job_id}"
|
|
mock_config.gmap2brief.timeout = 30
|
|
mock_config.app.local = False
|
|
|
|
interface = Gmap2BriefInterface(mock_logger)
|
|
|
|
with patch("dataloader.interfaces.gmap2_brief.interface.httpx.AsyncClient") as mock_client_cls:
|
|
# Create mock for stream context manager
|
|
mock_stream_ctx = MagicMock()
|
|
mock_stream_ctx.__aenter__ = AsyncMock(side_effect=httpx.TimeoutException("Timeout"))
|
|
mock_stream_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.stream = MagicMock(return_value=mock_stream_ctx)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
with pytest.raises(Gmap2BriefConnectionError) as exc_info:
|
|
from pathlib import Path
|
|
await interface.download_export("job123", Path("/tmp/test.zst"))
|
|
|
|
assert "Request error" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSuperTeneraErrorHandling:
|
|
"""Тесты обработки ошибок в SuperTeneraInterface."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_request_handles_http_status_error(self):
|
|
"""Тест обработки HTTP статус ошибки в _get_request."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.app.debug = False
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status = 503
|
|
mock_response.raise_for_status = MagicMock(side_effect=aiohttp.ClientResponseError(
|
|
request_info=MagicMock(),
|
|
history=(),
|
|
status=503,
|
|
message="Service Unavailable",
|
|
))
|
|
mock_response.json = AsyncMock(return_value={})
|
|
|
|
# Create context manager mock
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError) as exc_info:
|
|
await interface._get_request("/test")
|
|
|
|
assert "HTTP error 503" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_request_handles_connection_error(self):
|
|
"""Тест обработки ошибки подключения в _get_request."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.app.debug = False
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
# Create context manager mock that raises on __aenter__
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(side_effect=aiohttp.ClientConnectionError("Connection failed"))
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError) as exc_info:
|
|
await interface._get_request("/test")
|
|
|
|
assert "Connection error" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_request_handles_timeout(self):
|
|
"""Тест обработки таймаута в _get_request."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.app.debug = False
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
# Create context manager mock that raises on __aenter__
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(side_effect=TimeoutError("Request timeout"))
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError) as exc_info:
|
|
await interface._get_request("/test")
|
|
|
|
assert "Connection error" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_quotes_data_propagates_error(self):
|
|
"""Тест что get_quotes_data пробрасывает ошибки из _get_request."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.app.debug = False
|
|
mock_config.supertenera.quotes_endpoint = "/quotes"
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
# Create context manager mock that raises on __aenter__
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(side_effect=aiohttp.ServerTimeoutError("Server timeout"))
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError):
|
|
await interface.get_quotes_data()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_handles_client_response_error(self):
|
|
"""Тест обработки ClientResponseError в ping."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.supertenera.quotes_endpoint = "/quotes"
|
|
mock_config.supertenera.timeout = 10
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock(side_effect=aiohttp.ClientResponseError(
|
|
request_info=MagicMock(),
|
|
history=(),
|
|
status=500,
|
|
message="Internal Server Error",
|
|
))
|
|
|
|
# Create context manager mock
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError) as exc_info:
|
|
await interface.ping()
|
|
|
|
assert "HTTP error 500" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_handles_timeout_error(self):
|
|
"""Тест обработки TimeoutError в ping."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.supertenera.quotes_endpoint = "/quotes"
|
|
mock_config.supertenera.timeout = 10
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
# Create context manager mock that raises on __aenter__
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(side_effect=TimeoutError())
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError) as exc_info:
|
|
await interface.ping()
|
|
|
|
assert "Timeout" in str(exc_info.value)
|
|
assert "10s" in str(exc_info.value)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_handles_client_error(self):
|
|
"""Тест обработки общей ClientError в ping."""
|
|
mock_logger = MagicMock()
|
|
|
|
with patch("dataloader.interfaces.tenera.interface.APP_CONFIG") as mock_config:
|
|
mock_config.app.local = False
|
|
mock_config.supertenera.quotes_endpoint = "/quotes"
|
|
mock_config.supertenera.timeout = 10
|
|
|
|
interface = SuperTeneraInterface(mock_logger, "http://test.com")
|
|
|
|
mock_session = MagicMock()
|
|
interface._session = mock_session
|
|
|
|
# Create context manager mock that raises on __aenter__
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError("Generic client error"))
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_ctx)
|
|
|
|
with pytest.raises(SuperTeneraConnectionError) as exc_info:
|
|
await interface.ping()
|
|
|
|
assert "Connection error" in str(exc_info.value)
|