Как реализовать ожидание для API?

Стратегии ожидания в API-тестировании

Для эффективной реализации ожиданий в API-тестировании я применяю многоуровневый подход, комбинируя различные техники в зависимости от контекста.

Императивные (явные) ожидания

Наиболее простой и контролируемый подход — использование явных пауз с проверкой условий:

import time
import requests

def wait_for_api_response(url, max_attempts=10, delay=2):
    """Ожидание доступности API с экспоненциальной задержкой"""
    for attempt in range(max_attempts):
        try:
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                return response
        except requests.exceptions.RequestException:
            pass
        
        # Экспоненциальная задержка для снижения нагрузки
        sleep_time = delay * (2 ** attempt)
        time.sleep(min(sleep_time, 30))  # Ограничиваем максимальную задержку
    
    raise TimeoutError(f"API {url} не ответил за {max_attempts} попыток")

Умные ожидания с поллингом

Для асинхронных операций (генерация отчетов, обработка данных) реализую поллинг с интеллектуальными стратегиями:

import requests
from typing import Callable, Optional

class APIPoller:
    def __init__(self, initial_delay: float = 1.0, max_delay: float = 30.0):
        self.initial_delay = initial_delay
        self.max_delay = max_delay
    
    def poll_until(
        self,
        api_call: Callable,
        condition: Callable,
        timeout: float = 300,
        backoff_factor: float = 1.5
    ) -> Optional[dict]:
        """
        Полим API до выполнения условия с экспоненциальным откатом
        """
        import time
        
        start_time = time.time()
        current_delay = self.initial_delay
        
        while time.time() - start_time < timeout:
            try:
                response = api_call()
                
                if condition(response):
                    return response
                
                # Если ответ получен, но условие не выполнено
                if response.status_code == 202:  # Accepted - операция в процессе
                    time.sleep(current_delay)
                    current_delay = min(current_delay * backoff_factor, self.max_delay)
                else:
                    # Для других статусов возможна немедленная реакция
                    break
                    
            except requests.exceptions.RequestException:
                time.sleep(current_delay)
                current_delay = min(current_delay * backoff_factor, self.max_delay)
        
        return None

Ожидание по бизнес-условиям

Ключевой аспект — отделение технических ожиданий от бизнес-логики:

class BusinessConditionWaiter:
    def wait_for_order_status(self, order_id: str, expected_status: str, timeout: int = 60):
        """
        Ожидание конкретного статуса заказа в системе
        """
        poller = APIPoller()
        
        def check_order():
            return requests.get(f"/api/orders/{order_id}")
        
        def status_condition(response):
            if response.status_code == 200:
                data = response.json()
                return data.get('status') == expected_status
            return False
        
        result = poller.poll_until(check_order, status_condition, timeout)
        if not result:
            raise AssertionError(
                f"Статус заказа {order_id} не изменился на {expected_status} "
                f"в течение {timeout} секунд"
            )

Практические рекомендации по реализации

Конфигурируемые параметры ожидания

• Все временные параметры выношу в конфигурацию
• Использую разные таймауты для различных окружений (dev/stage/prod)
• Реализую декораторы для повторных попыток (retry logic)

from functools import wraps
import requests

def retry_on_failure(max_retries=3, delay=1, backoff=2):
    """Декоратор для повторных попыток вызова API"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            current_delay = delay
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (requests.exceptions.ConnectionError, 
                        requests.exceptions.Timeout) as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        time.sleep(current_delay)
                        current_delay *= backoff
            
            raise last_exception
        return wrapper
    return decorator

Мониторинг и логирование

• Добавляю детальное логирование всех этапов ожидания
• Реализую метрики времени отклика API
• Использую структурированные логи для анализа проблем

import logging
from contextlib import contextmanager

@contextmanager
def api_timing_context(api_name: str):
    """Контекстный менеджер для измерения времени выполнения API"""
    start_time = time.time()
    try:
        yield
    finally:
        duration = time.time() - start_time
        logging.info({
            'event': 'api_timing',
            'api': api_name,
            'duration_seconds': round(duration, 3),
            'threshold_exceeded': duration > 5.0
        })

Распространенные антипаттерны и их решения

1. Жесткие sleep() без условия → Заменяю на поллинг с условием
2. Отсутствие таймаутов → Всегда устанавливаю разумные лимиты
3. Игнорирование HTTP-статусов → Обрабатываю 202, 429, 503 отдельно
4. Ожидание в UI-тестах вместо API → Переношу логику ожидания на уровень API

Интеграция с фреймворками тестирования

Для проектов на pytest использую встроенные фикстуры и хуки:

import pytest

@pytest.fixture
def api_patiently():
    """Фикстура для терпеливого ожидания API"""
    poller = APIPoller()
    
    def _wait(api_call, condition, timeout=30):
        return poller.poll_until(api_call, condition, timeout)
    
    return _wait

Заключение

Эффективная стратегия ожидания для API включает:

• Комбинацию явных и неявных ожиданий
• Адаптивные алгоритмы с экспоненциальным откатом
• Четкое разделение технических и бизнес-ожиданий
• Детальное логирование и мониторинг
• Конфигурируемые параметры для разных окружений

Ключевой принцип — реализовывать "терпеливое", но не "бесконечное" ожидание, которое обеспечивает стабильность тестов без излишнего увеличения времени их выполнения. Оптимальный подход всегда зависит от конкретного API: его SLA, характера операций (синхронные/асинхронные) и требований бизнеса.