Какие используются подходы для тестирования микросервисов?

Подходы для тестирования микросервисов

Контекст микросервисной архитектуры

Микросервисы — это небольшие независимые сервисы, которые работают вместе. Тестирование становится сложнее из-за распределенности.

1. Unit Testing (Модульное тестирование)

Назначение: тестировать отдельные функции/методы сервиса

Уровень: внутри сервиса

import pytest
from calculator_service import add

def test_add_positive_numbers():
    assert add(2, 3) == 5

def test_add_negative_numbers():
    assert add(-2, -3) == -5

Преимущества:

• Быстро
• Много покрытие (70%+)
• Дешево

2. Integration Testing (Интеграционное тестирование)

Назначение: тестировать взаимодействие между компонентами внутри сервиса

Уровень: один сервис, но несколько компонентов

import pytest
from flask import Flask
from calculator_service import app, db

@pytest.fixture
def client():
    app.config['TESTING'] = True
    with app.test_client() as client:
        with app.app_context():
            db.create_all()
            yield client
            db.session.remove()
            db.drop_all()

def test_create_and_get_user(client):
    # Создание
    response = client.post('/users',
        json={'name': 'John', 'email': 'john@test.com'}
    )
    assert response.status_code == 201
    user_id = response.json['id']
    
    # Получение
    response = client.get(f'/users/{user_id}')
    assert response.status_code == 200
    assert response.json['name'] == 'John'

3. Service-to-Service Testing

Назначение: тестировать взаимодействие между микросервисами

Уровень: несколько сервисов

import requests
import pytest

USER_SERVICE = "http://user-service:5000"
ORDER_SERVICE = "http://order-service:5001"

def test_create_order_with_user():
    # 1. Создаем пользователя в User Service
    user_response = requests.post(
        f"{USER_SERVICE}/users",
        json={'name': 'John', 'email': 'john@test.com'}
    )
    assert user_response.status_code == 201
    user_id = user_response.json['id']
    
    # 2. Создаем заказ в Order Service
    order_response = requests.post(
        f"{ORDER_SERVICE}/orders",
        json={'user_id': user_id, 'total': 100.00}
    )
    assert order_response.status_code == 201
    assert order_response.json['user_id'] == user_id

4. Contract Testing

Назначение: обеспечить совместимость между сервисами через контракты

Инструмент: Pact

from pact import Consumer, Provider

pact = Consumer('OrderService').has_state(
    'user with id 1 exists'
).upon_receiving(
    'a request for user 1'
).with_request(
    'get', '/users/1'
).will_respond_with(
    200, body={'id': 1, 'name': 'John'}
)

with pact:
    # Тест, что наше приложение использует этот контракт
    user = get_user(1)
    assert user['name'] == 'John'

5. End-to-End (E2E) Testing

Назначение: тестировать полный flow через все сервисы

Уровень: вся система

import pytest
import requests
import time

API_GATEWAY = "http://api-gateway:8000"

def test_complete_user_order_flow():
    # 1. Регистрация
    register = requests.post(
        f"{API_GATEWAY}/register",
        json={'name': 'John', 'email': 'john@test.com', 'password': 'pass'}
    )
    assert register.status_code == 201
    user_id = register.json['id']
    
    # 2. Логин
    login = requests.post(
        f"{API_GATEWAY}/login",
        json={'email': 'john@test.com', 'password': 'pass'}
    )
    assert login.status_code == 200
    token = login.json['token']
    
    # 3. Создание заказа
    order = requests.post(
        f"{API_GATEWAY}/orders",
        json={'items': [{'id': 1, 'qty': 2}]},
        headers={'Authorization': f'Bearer {token}'}
    )
    assert order.status_code == 201
    order_id = order.json['id']
    
    # 4. Проверка статуса заказа
    time.sleep(1)  # Дождаться обработки
    status = requests.get(
        f"{API_GATEWAY}/orders/{order_id}",
        headers={'Authorization': f'Bearer {token}'}
    )
    assert status.status_code == 200
    assert status.json['status'] == 'confirmed'

6. Load/Performance Testing

Назначение: проверить производительность под нагрузкой

Инструмент: Locust, JMeter

from locust import HttpUser, task, between

class UserBehavior(HttpUser):
    wait_time = between(1, 3)
    
    @task(1)
    def get_user(self):
        self.client.get("/users/1")
    
    @task(2)
    def create_order(self):
        self.client.post(
            "/orders",
            json={'user_id': 1, 'total': 100}
        )

7. Chaos Testing

Назначение: проверить отказоустойчивость при сбоях

Инструмент: Chaos Monkey, Gremlin

import random

def test_service_resilience():
    # Имитируем сбой одного сервиса
    with mock.patch('requests.get') as mock_get:
        mock_get.side_effect = requests.exceptions.Timeout()
        
        # Наше приложение должно обработать это
        response = get_order(1)
        assert response['status'] == 'pending'  # Fallback
        assert response.get('cached_data') is not None

Практический пример с Docker Compose

version: '3.8'

services:
  user-service:
    build: ./user-service
    ports:
      - "5000:5000"
    environment:
      DATABASE_URL: postgresql://user:pass@db:5432/users
  
  order-service:
    build: ./order-service
    ports:
      - "5001:5000"
    environment:
      USER_SERVICE_URL: http://user-service:5000
      DATABASE_URL: postgresql://user:pass@db:5432/orders
    depends_on:
      - user-service
  
  db:
    image: postgres:13
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass

Пирамида тестирования для микросервисов

         △ E2E (API Gateway) - 5-10%
        △ △ Service-to-Service - 15-20%
       △ △ △ Integration - 20-30%
      △ △ △ △ Unit - 50%

Best Practices

1. Используй контракты — Pact для гарантии совместимости
2. Изолируй тесты — каждый сервис тестируется независимо
3. Мокируй зависимости — используй WireMock для имитации
4. Тестируй асинхронность — очереди, события
5. Мониторинг — логируй, метрики, трейсинг
6. Откат — убедись, что сервис может graceful shutdown

# Пример использования WireMock
import requests
from responses import RequestsMock

def test_with_mocked_service():
    with RequestsMock() as rsps:
        rsps.add(
            responses.GET,
            'http://user-service:5000/users/1',
            json={'id': 1, 'name': 'John'},
            status=200
        )
        
        user = get_user(1)
        assert user['name'] == 'John'

Для QA в микросервисной архитектуре критично тестировать не только отдельные сервисы, но и их взаимодействие через контракты и E2E сценарии.