Modern QA2026Тестирование бессерверных функций
Join

Course › 08 Infrastructure as Code Testing

Cutting-edge · Chapter 08

Тестирование бессерверных функций

Updated Jul 2026

Вызовы тестирования бессерверных архитектур

Бессерверные функции выполняются в окружениях, которые вы не контролируете. Вы не можете подключиться к машине по SSH, не можете проверить файловую систему и не можете присоединить отладчик в продакшене. Среда выполнения эфемерна — она существует только на время вызова.

Это означает, что ваша стратегия тестирования должна быть более строгой, чем при тестировании традиционных приложений. Если вы пропустите ошибку в бессерверной функции, вашими единственными инструментами отладки будут логи и трейсы. Нет никакого «давайте быстро проверим на сервере».

AWS Lambda с SAM Local

AWS SAM (Serverless Application Model) обеспечивает локальную эмуляцию функций Lambda. SAM Local запускает Docker-контейнер, имитирующий среду выполнения Lambda, что позволяет тестировать функции локально перед развёртыванием в AWS.

Локальный вызов

# Invoke a single function locally with a test event
sam local invoke MyFunction --event events/api-gateway.json

# Start a local API Gateway emulator
sam local start-api --port 3000

# Run tests against the local emulator
pytest tests/integration/ --base-url http://localhost:3000

# Generate a sample event for testing
sam local generate-event apigateway aws-proxy > events/api-gateway.json
sam local generate-event sqs receive-message > events/sqs-message.json
sam local generate-event s3 put > events/s3-put.json

Модульное тестирование обработчиков Lambda

Обработчики Lambda — это обычные функции. Тестируйте их как любые другие функции, мокая AWS-сервисы, которые они вызывают:

# tests/unit/test_handler.py
import json
import pytest
from unittest.mock import patch, MagicMock
from src.handlers.order_processor import lambda_handler

def test_order_processing_success():
    """Test successful order processing."""
    event = {
        "Records": [{
            "body": json.dumps({
                "orderId": "ORD-123",
                "items": [{"sku": "WIDGET-A", "qty": 2}],
                "customerId": "CUST-456"
            })
        }]
    }
    context = MagicMock()
    context.function_name = "order-processor"
    context.memory_limit_in_mb = 256
    context.get_remaining_time_in_millis.return_value = 30000

    result = lambda_handler(event, context)

    assert result["statusCode"] == 200
    body = json.loads(result["body"])
    assert body["orderId"] == "ORD-123"
    assert body["status"] == "processed"

def test_order_processing_invalid_payload():
    """Test handling of malformed input."""
    event = {"Records": [{"body": "not json"}]}
    context = MagicMock()

    result = lambda_handler(event, context)

    assert result["statusCode"] == 400
    body = json.loads(result["body"])
    assert "error" in body

def test_order_processing_missing_required_fields():
    """Test handling of missing required fields."""
    event = {
        "Records": [{
            "body": json.dumps({
                "orderId": "ORD-123"
                # Missing: items and customerId
            })
        }]
    }
    context = MagicMock()

    result = lambda_handler(event, context)

    assert result["statusCode"] == 400

@patch("src.handlers.order_processor.dynamodb_client")
def test_order_persisted_to_dynamodb(mock_dynamo):
    """Test that processed orders are saved to DynamoDB."""
    event = {
        "Records": [{
            "body": json.dumps({
                "orderId": "ORD-123",
                "items": [{"sku": "WIDGET-A", "qty": 2}],
                "customerId": "CUST-456"
            })
        }]
    }
    context = MagicMock()

    lambda_handler(event, context)

    mock_dynamo.put_item.assert_called_once()
    call_args = mock_dynamo.put_item.call_args
    item = call_args["Item"]
    assert item["orderId"]["S"] == "ORD-123"
    assert item["status"]["S"] == "processed"

Тестирование Google Cloud Functions

Google Cloud Functions можно тестировать локально с использованием Functions Framework:

# Install the Functions Framework for local testing
pip install functions-framework

# Run locally
functions-framework --target=process_order --port=8080

# Test with curl
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{"orderId": "ORD-123"}'

Тестирование Cloud Functions с pytest

# tests/test_cloud_function.py
import pytest
from unittest.mock import patch, MagicMock
from flask import Flask
import json

# Import the function
from main import process_order

@pytest.fixture
def app():
    return Flask(__name__)

@pytest.fixture
def client(app):
    return app.test_client()

def test_process_order_valid_request(app):
    """Test processing a valid order."""
    with app.test_request_context(
        method="POST",
        json={"orderId": "ORD-123", "amount": 99.99}
    ):
        from flask import request
        response = process_order(request)
        data = json.loads(response)
        assert data["status"] == "processed"

def test_process_order_missing_order_id(app):
    """Test handling of missing order ID."""
    with app.test_request_context(
        method="POST",
        json={"amount": 99.99}
    ):
        from flask import request
        response, status_code = process_order(request)
        assert status_code == 400

Вызовы и решения при тестировании бессерверных архитектур

Вызов Решение Инструмент/Подход
Задержка холодного старта Измерять p99-задержку в нагрузочных тестах Artillery, k6 с целевыми Lambda
Поведение при таймауте Устанавливать таймаут функции ниже таймаута теста Свойство Timeout в SAM-шаблоне
Ограничения памяти Профилировать использование памяти при ожидаемых размерах payload Lambda Power Tuning
Лимиты параллелизма Нагрузочное тестирование с зарезервированным параллелизмом k6 со сценариями нарастания нагрузки
Разрешения IAM Тестировать с ролью минимальных привилегий локально SAM local с профилями учётных данных
Маппинг источников событий Тестировать с реалистичными event payload Воспроизведение событий из CloudWatch
Идемпотентность Отправлять дублирующие события, проверять отсутствие двойной обработки Интеграционные тесты с логикой повторов

Тестирование холодного старта

# tests/performance/test_cold_start.py
import time
import boto3
import statistics

def measure_cold_start(function_name, num_invocations=10):
    """Measure cold start latency by forcing new execution environments."""
    client = boto3.client("lambda")
    latencies = []

    for i in range(num_invocations):
        # Update environment variable to force a new container
        client.update_function_configuration(
            FunctionName=function_name,
            Environment={
                "Variables": {
                    "COLD_START_TRIGGER": str(time.time())
                }
            }
        )
        # Wait for update to propagate
        time.sleep(5)

        start = time.time()
        response = client.invoke(
            FunctionName=function_name,
            Payload=b'{"test": true}',
        )
        elapsed_ms = (time.time() - start) * 1000
        latencies.append(elapsed_ms)

    return {
        "p50": statistics.median(latencies),
        "p95": sorted(latencies)[int(len(latencies) * 0.95)],
        "p99": sorted(latencies)[int(len(latencies) * 0.99)],
        "max": max(latencies),
    }

Тестирование таймаутов

def test_function_completes_before_timeout():
    """Verify the function completes within its configured timeout."""
    import time

    FUNCTION_TIMEOUT = 30  # seconds (from SAM template)
    SAFETY_MARGIN = 5      # seconds

    start = time.time()
    result = lambda_handler(large_event, context)
    elapsed = time.time() - start

    assert elapsed < (FUNCTION_TIMEOUT - SAFETY_MARGIN), \
        f"Function took {elapsed:.1f}s, dangerously close to {FUNCTION_TIMEOUT}s timeout"

Тестирование с LocalStack

LocalStack предоставляет локальный стек AWS для тестирования бессерверных функций с реальной эмуляцией AWS-сервисов:

# tests/integration/test_lambda_localstack.py
import boto3
import json
import pytest

@pytest.fixture(scope="module")
def aws_clients():
    """Create AWS clients pointing to LocalStack."""
    endpoint = "http://localhost:4566"
    return {
        "lambda": boto3.client("lambda", endpoint_url=endpoint,
                               region_name="us-east-1",
                               aws_access_key_id="test",
                               aws_secret_access_key="test"),
        "sqs": boto3.client("sqs", endpoint_url=endpoint,
                            region_name="us-east-1",
                            aws_access_key_id="test",
                            aws_secret_access_key="test"),
        "dynamodb": boto3.client("dynamodb", endpoint_url=endpoint,
                                 region_name="us-east-1",
                                 aws_access_key_id="test",
                                 aws_secret_access_key="test"),
    }

def test_lambda_processes_sqs_message(aws_clients):
    """Test the full flow: SQS message triggers Lambda, result in DynamoDB."""
    sqs = aws_clients["sqs"]
    dynamodb = aws_clients["dynamodb"]

    # Send message to SQS
    queue_url = sqs.create_queue(QueueName="orders")["QueueUrl"]
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({"orderId": "ORD-TEST", "amount": 42.00})
    )

    # Wait for Lambda to process (triggered by SQS event source mapping)
    import time
    time.sleep(5)

    # Verify the result in DynamoDB
    result = dynamodb.get_item(
        TableName="orders",
        Key={"orderId": {"S": "ORD-TEST"}}
    )
    assert "Item" in result
    assert result["Item"]["status"]["S"] == "processed"

Комбинация модульных тестов (быстрые, изолированные), локальной эмуляции (SAM Local, Functions Framework) и интеграционных тестов (LocalStack, Testcontainers) обеспечивает всестороннее покрытие бессерверных функций без необходимости развёрнутого облачного окружения для каждого запуска тестов.