Создание асинхронных модульных тестов с помощью PYTHON + AWS LAMBDA + PYTEST + MOTO

Здравствуйте, ребята, в сегодняшнем посте я покажу вам, как я выполняю модульные тесты на AWS LAMBDAS, используя pytest и moto async. Это очень сложная тема, потому что в Интернете не так много информации, особенно когда речь идет об асинхронном тестировании, у moto даже есть механизм для запуска асинхронного режима, но мне не понравилось его использовать.

Здесь я говорю не о том, почему мы должны тестировать, какой тест лучше, а о том, как мы можем запускать асинхронные тесты на наших лямбда-выражениях.

Пойдем! :)

Структура папок

Вот как я структурировал свой лямбда-проект на питоне, я использую монорепозиторий, который группирует контекст. (Это отлично работает для небольших команд, вскоре последует пост о монорепо и мультирепо).

AWS: эта папка отвечает за формирование облака AWS SAM и информацию о конфигурации. Когда мы строим наш процесс CI/CD на таких сервисах, как GitLab, Bitbucket или Github, содержащиеся здесь файлы конфигурации помогают нам загружать нашу инфраструктуру для любого региона aws с различными конфигурациями.
BUILD: папка сборки — это место, где у нас есть сценарии для запуска наших модульных тестов, цикломатических тестов сложности и создания конфигураций для развертывания нашего облачного формирования. Мне нравится эта папка, потому что здесь находится все, что связано со сценариями, которые будут выполняться на нашем листе развертывания.
SRC: Последнее, но не менее важное — это папка src, в которую мы помещаем нашу лямбду, примеры тестовых событий, которые мы будем использовать как для разработки, так и позже для развертывания.
SETUP.CFG: этот файл также очень актуален, так как он содержит все правила pytest, которые будут загружены, как показано ниже.

Запуск проекта

В моем примере я протестировал три сервиса AWS.

Ведро S3
Динамо
SQS

Чтобы начать наши тесты, давайте сначала создадим несколько важных файлов и папок, это поможет нам лучше понять, что входит в каждую папку.

setup.cfg

[tool:pytest]
env =
    POWERTOOLS_SERVICE_NAME = MOCK_TEST
    POWERTOOLS_METRICS_NAMESPACE = MOCK_TEST
    POWERTOOLS_TRACE_DISABLE = 1
    POWERTOOLS_LOGGER_RATE = 0.1
    AWS_XRAY_SDK_ENABLE = false

addopts = -p no:warnings
filterwarnings = ignore::pytest.PytestConfigWarning
asyncio_mode = auto
# log_format = %(asctime)s %(levelname)s %(message)s
# log_date_format = %Y-%m-%d %H:%M:%S

[flake8]
ignore = E203
max-line-length = 120
exclude =
    .git,
    __pycache__,
    docs/source/conf.py,
    old,
    build,
    dist,
    swagger,
    events,
    test

[tool.black]
line-length = 120
target-version = ['py39', 'py310']
include = '\.pyi?$'


[report]
exclude_lines =
    pragma: no cover
    def __repr__
    if self.debug
    if settings.DEBUG
    raise AssertionError
    raise NotImplementedError
    if 0:
    if __name__ == .__main__.:

[coverage:run]
omit =
    test/**/*.py
    test/*.py
    tests/**/*.py
    tests/*.py
    ~/.local/lib/python3.8/site-packages/six.py
    ~/.local/lib/python3.8/site-packages/pywintypes.py
    **/__init__.py
    venv/**/*
    .virtualenvs/**/*

Теперь разместим наш файл конфигурации покрытия

.coveragerc

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    if self.debug:
    if settings.DEBUG
    raise AssertionError
    raise NotImplementedError
    if 0:
    if __name__ == .__main__.:

[run]
omit =
    *.cfg
    test/**/*.py
    test/*.py
    tests/**/*.py
    tests/*.py
    ~/.local/lib/python3.9/site-packages/six.py
    **/__init__.py
    venv/**/*

Помня, что эти два файла должны быть в корне проекта

Теперь мы создадим папку AWS и добавим в нее два файла: config.toml и template.yaml. , потому что именно через них сам может определить в какую папку в s3 класть файлы помимо содержащих параметры для загрузки инфраструктуры .

config.toml

version = 0.1

[default]
stack_name = "MOCK_TEST"
s3_prefix = "MOCK_TEST"
capabilities = "CAPABILITY_IAM"

[aws_account_number.us-east-1]
s3_bucket = "aws-sam-cli-deploy-5841bfa89d62"
[aws_account_number.us-east-1.parameter_overrides]
Role = "arn:aws:iam::accountNumber:role/role-lambda"

Не забудьте изменить параметр aws_account_number на номер вашей учетной записи aws. Помните, что у вас может быть несколько учетных записей, настроенных с несколькими разными параметрами, это очень помогает при развертывании в других регионах или разных учетных записях.

template.yaml

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: This is service test

Parameters:
  Role:
    Type: String
    Description: Permission allowed services access

  buildVersion:
    Type: String
    Description: Build version run gitlab

Globals:
  Function:
    Runtime: python3.9
    Timeout: 30
    MemorySize: 128
    Architectures:
      - arm64
    Layers:
      - !Sub arn:aws:lambda:${AWS::Region}:017000801446:layer:AWSLambdaPowertoolsPython:36
    Tags:
      service: mock-test
      buildVersion: !Ref buildVersion

Resources:
  s3:
    Type: 'AWS::S3::Bucket'
    Properties:
      BucketName: BUCKET_TEST
      AccessControl: Private
      VersioningConfiguration:
        Status: Enabled

  dynamo:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: DYNAMO_TABLE
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      BillingMode: PAY_PER_REQUEST

  sqsQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: SQS_QUEUE
      DelaySeconds: 0
      VisibilityTimeout: 30
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt sqsDeadLatter.Arn
        maxReceiveCount: 5

  sqsDeadLatter:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: SQS-QUEUE-DEAD-LATTER

  lambda:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: MOCK-TEST-LAMBDA
      CodeUri: ../src/mock_test
      Handler: app.app.lambda_handler
      Role: !Ref Role
      Environment:
        Variables:
          SQS_URL_PUBLISH: !Ref sqsQueue
          TABLE: !Ref dynamo
          BUCKET: !Ref s3

Готовый! Наши два файла, которые нам нужно развернуть, записаны!

Теперь давайте создадим еще одну папку с именем BUILD и добавим в нее 4 файла: generate_samconfig.py, radon.sh, requirements .txt, test.sh они отвечают за создание файла конфигурации и выполнение наших тестов.

generate_samconfig.py

import argparse
import logging

import toml


def format_parameter_overrides(parameters_overrides):
    template = lambda parameter, value: f'{parameter}="{value}"'
    parameters = [template(*item) for item in parameters_overrides.items()]
    return " ".join(parameters)


def format_parameters(parameters):
    p = parameters.copy()
    if "parameter_overrides" not in p:
        return p

    parameter_overrides = p.get("parameter_overrides")
    p["parameter_overrides"] = format_parameter_overrides(parameter_overrides)
    return p


def format_samconfig(parameters, version=0.1):
    base = {
        "version": version,
        "default": {"deploy": {"parameters": format_parameters(parameters)}},
    }
    return base


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Generate samconfig from config file")
    parser.add_argument("--account-id", type=str, required=True)
    parser.add_argument("--region", type=str, required=True)
    parser.add_argument("--config-file", type=str, required=True)
    parser.add_argument("--output", type=str, required=True)
    parser.add_argument("--extra-parameter", action="append", default=[])

    args = parser.parse_args()
    region = args.region
    account_id = args.account_id

    extra_parameters = dict()
    for parameter in args.extra_parameter:
        parameter_split = parameter.split("=")
        if len(parameter_split) < 2:
            raise f"Missing value of key {parameter_split[0]}"

        extra_parameters.update({parameter_split[0]: parameter_split[1]})

    config = toml.load(args.config_file)
    version = config.get("version", 0.1)

    default_config = config.get("default", {})
    account_config = config.get(account_id, {})
    region_config = account_config.get(region, {})
    config_output = default_config.copy()
    config_output.update(region_config)
    config_output["region"] = region

    parameter_overrides = config_output.get("parameter_overrides", {}).copy()
    parameter_overrides.update(extra_parameters)

    if len(parameter_overrides) > 0:
        config_output["parameter_overrides"] = parameter_overrides

    samconfig = format_samconfig(config_output, version=version)

    with open(args.output, "w") as f:
        toml.dump(samconfig, f)

    logging.info("File samconfig generate was success.")

radon.sh

#!/bin/bash

set -e

level=$1
if [ -z "$level" ]
then
  echo "Complexity level has not been defined. It must be between 'a' e 'f'."
  exit 1
fi
cyclomatic_complexity_result=$(radon cc -n$level --exclude "venv*" .)
if [ -z "$cyclomatic_complexity_result" ]
then
  echo "All files have low cyclomatic complexity."
else
  echo "Some files were identified with complexity higher than '$level'."
  printf "$cyclomatic_complexity_result\n"
  exit 1
fi

radon.sh

aioboto3==10.1.0
aws_lambda_powertools==1.29.2
moto==4.0.5
pre-commit==2.20.0
pytest==7.1.3
pytest-aiohttp==1.0.4
pytest-asyncio==0.19.0
pytest-cov==3.0.0
pytest-env==0.6.2
radon==5.1.0

test.sh

#!/bin/bash

for folder in src/*; do
  if [ "$folder" != "src/aws" ] && [ "$folder" != "src/tests" ] && [ "$folder" != "src/__init__.py" ] && [ "$folder" != "src/scripts" ] && [ "$folder" != "src/tests@tmp" ] && [ "$folder" != "src/__pycache__" ] && [ "$folder" != "src/scripts@tmp" ]; then
    cd "$folder"
    set -e;
    set -x;
    echo "Testing $folder"
    pytest --cov=app --cov-report term-missing --cov-fail-under 70 test/ -v;
    cd ..;
    cd ..;
  fi
done

# pytest --cov=app --cov-report term-missing --cov-fail-under 70 test/
# radon cc -na --exclude "venv*" .
# pytest -o log_cli=true --log-cli-level=10

Это была первая часть создания файлов и папок для нашего проекта. Во второй части мы создадим лямбду и проведем несколько тестов.

→ ЧАСТЬ 2

Если вы хотите скачать весь проект.
https://github.com/vieirinhasantana/aws-lambda-pytest-moto-async

Создание асинхронных модульных тестов с помощью PYTHON + AWS LAMBDA + PYTEST + MOTO — Часть 1

Структура папок

Запуск проекта

Вопросы по теме