添加到moto默认aws ssm参数

Question

我正在尝试运行一组测试，在这些测试中，对boto3.client('ssm')的调用是使用莫托模拟的。

Moto提供了一组默认的aws参数。https://github.com/spulec/moto/blob/master/moto/ssm/models.py#L59，但正在阻止添加更多：https://github.com/spulec/moto/blob/master/moto/ssm/models.py#L858试图主动添加任何aws前缀参数将按照boto3.py#L397中的测试返回一个错误

由于我的lambda依赖于下面的内容，所以我的测试失败了：/aws/service/ecs/optimized-ami/amazon-linux-2/recommended

我想试着给这个被嘲弄的ssm客户端打补丁，但我对moto的内部结构了解甚少。

我一直在跟踪这个示例，但为了我的需要修改了它(调用SSM而不是调用SQS和S3)。对于ref，我的代码如下所示，因为我试图修改put_parameter方法，但没有成功。

app.py

import boto3
from loguru import logger


@logger.catch()
def lambda_handler(event, context):
    ssm_client = boto3.client("ssm", "eu-west-1")

    ami_param_name = "/aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-ebs"
    ami_param_value = ssm_client.get_parameter(Name=ami_param_name)

    ecs_param_name = "/aws/service/ecs/optimized-ami/amazon-linux-2/recommended"
    ecs_param_value = ssm_client.get_parameter(Name=ecs_param_name)

    return [ami_param_value, ecs_param_value]

test.py

import os
from unittest import mock

import boto3
import pytest
from moto import mock_ssm

from src.app import lambda_handler

AWS_REGION = 'eu-west-1'


@pytest.fixture(scope="function")
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"


@pytest.fixture(scope="function")
def mock_ssm_client(aws_credentials):
    with mock_ssm():
        client = boto3.client("ssm", region_name=AWS_REGION)

        # already present in moto
        # client.put_parameter(
        #     Name='/aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-ebs',
        #     Type='String',
        #     Value='ami-stdparam'
        # )

        # What the lambda requires
        # client.put_parameter(
        #     Name='/aws/service/ecs/optimized-ami/amazon-linux-2/recommended',
        #     Type='String',
        #     Value='{"image_id": "ami-ecsparam"}'
        # )

        def side_effect(path):
            if path == "/aws/service/ecs/optimized-ami/amazon-linux-2/recommended":
                return_value = {
                    "Parameter": {
                        "Name": "/aws/service/ecs/optimized-ami/amazon-linux-2/recommended",
                        "Type": "String",
                        "Value": "{\"ecs_agent_version\":\"1.63.1\",\"ecs_runtime_version\":\"Docker version 20.10.13\",\"image_id\":\"ami-002e2fef4b94f8fd0\",\"image_name\":\"amzn2-ami-ecs-hvm-2.0.20220921-x86_64-ebs\",\"image_version\":\"2.0.20220921\",\"os\":\"Amazon Linux 2\",\"schema_version\":1,\"source_image_name\":\"amzn2-ami-minimal-hvm-2.0.20220912.1-x86_64-ebs\"}",
                        "Version": 94,
                        "LastModifiedDate": 1664230158.399,
                        "ARN": "arn:aws:ssm:eu-west-1::parameter/aws/service/ecs/optimized-ami/amazon-linux-2/recommended",
                        "DataType": "text"
                    }
                }

                return return_value
            else:
                return client.get_parameter(path)

        client.get_parameter = mock.patch(
            'boto3.client.get_parameter',
            side_effect=side_effect
        )
        yield client


def test_lambda_handler(mock_ssm_client):
    # Arrange

    # Act
    results = lambda_handler('', 'test')

    # Assert
    assert len(results) == 2

Answer 1

您可以使用Moto的内部API来存储参数，作为模拟/修补Moto的解决办法。

请参阅下面的代码，以添加名为/aws/test的自定义参数

@mock_ssm
def test_default_param():
    client = boto3.client("ssm", region_name="us-east-1")

    from moto.ssm.models import ssm_backends, Parameter
    ssm_backends["123456789012"]["us-east-1"]._parameters["/aws/test"].append(Parameter(
        account_id="123456789012",
        name="/aws/test",
        value="val",
        parameter_type="String",
        description="...",
        allowed_pattern=None,
        keyid=None,
        last_modified_date=1664230158.399,
        version=None,
        tags=[],
        data_type="text",
    ))

    response = client.get_parameters(Names=["/aws/test"])
    print(response)

请注意，这在Moto的最新版本(4.0.6)中有效，但是由于它是一个内部API，它可能会更改。