Как протестировать взаимодействие с внешним сервисом?

Тестирование взаимодействия с внешними сервисами

Тестирование интеграции с внешними API — это отдельное искусство. Нельзя делать реальные запросы в тестах, так как это делает тесты медленными, хрупкими и зависимыми от доступности сервиса. Расскажу о надёжных методах.

Проблемы при тестировании внешних сервисов

• Скорость — реальный HTTP запрос занимает 500мс-2с
• Надёжность — сервис может быть недоступен
• Стоимость — каждый запрос может быть платным
• Недетерминированность — разные ответы при повторных запросах
• Зависимость от окружения — нужны тестовые ключи API

Метод 1: Mock (Мокирование)

from unittest.mock import Mock, patch
import requests

class PaymentService:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.payment.com"
    
    def charge_card(self, card_id: str, amount: float) -> dict:
        response = requests.post(
            f"{self.base_url}/charge",
            json={"card_id": card_id, "amount": amount},
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        return response.json()

def test_charge_card_with_mock():
    service = PaymentService(api_key="test_key")
    
    with patch("requests.post") as mock_post:
        mock_post.return_value.json.return_value = {
            "transaction_id": "txn_123",
            "status": "success",
            "amount": 100.00
        }
        
        result = service.charge_card("card_123", 100.00)
        
        assert result["status"] == "success"
        mock_post.assert_called_once()

Метод 2: MagicMock с side_effect

from unittest.mock import MagicMock, patch

class EmailService:
    def send_email(self, to: str, subject: str, body: str) -> bool:
        pass

def test_email_service_with_side_effect():
    with patch.object(EmailService, "send_email") as mock_send:
        mock_send.side_effect = [True, False, Exception("Server error")]
        
        service = EmailService()
        
        assert service.send_email("user@example.com", "Hi", "Hello") == True
        assert service.send_email("user@example.com", "Hi", "Hello") == False

Метод 3: VCR.py (запись HTTP ответов)

import vcr
import requests

class WeatherService:
    def get_temperature(self, city: str) -> float:
        response = requests.get(
            f"https://api.weather.com/temp",
            params={"city": city}
        )
        return response.json()["temperature"]

@vcr.VCR(
    cassette_library_dir="tests/fixtures/vcr_cassettes",
    record_mode="once"
)
def test_weather_service():
    service = WeatherService()
    temp = service.get_temperature("Moscow")
    
    assert temp > -50 and temp < 50

Метод 4: Responses библиотека

import responses
import requests

@responses.activate
def test_api_with_responses():
    responses.add(
        responses.GET,
        "https://api.example.com/users/123",
        json={"id": 123, "name": "John"},
        status=200,
    )
    
    response = requests.get("https://api.example.com/users/123")
    assert response.json()["name"] == "John"

Метод 5: Тестирование обработки ошибок

from unittest.mock import patch, Mock
import requests

class APIClient:
    def get_data(self, endpoint: str) -> dict:
        try:
            response = requests.get(endpoint, timeout=5)
            response.raise_for_status()
            return response.json()
        except requests.Timeout:
            return {"error": "Request timeout"}
        except requests.HTTPError as e:
            return {"error": f"HTTP error"}

def test_api_error_handling():
    client = APIClient()
    
    with patch("requests.get") as mock_get:
        mock_get.side_effect = requests.Timeout()
        result = client.get_data("https://api.example.com/data")
        assert result["error"] == "Request timeout"

Метод 6: AsyncIO с aioresponses

from aioresponses import aioresponses
import aiohttp

async def fetch_user(user_id: int) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as response:
            return await response.json()

async def test_async_external_service():
    with aioresponses() as mock_aio:
        mock_aio.get(
            "https://api.example.com/users/123",
            payload={"id": 123, "name": "John"}
        )
        
        result = await fetch_user(123)
        assert result["name"] == "John"

Best Practices

1. Не делай реальные запросы — используй мокирование для юнит-тестов
2. Используй VCR.py для интеграционных тестов — записывает и воспроизводит ответы
3. Тестируй обработку ошибок — сервис может быть недоступен
4. Проверяй параметры запроса — используй assert_called_with()
5. Изолируй внешние зависимости — используй dependency injection
6. Раздели unit-тесты и интеграционные — юниты быстрые и надёжные
7. Документируй mock-поведение — для сложных сценариев

Правильное мокирование делает тесты быстрыми, надёжными и независимыми от внешних сервисов.