pytest - 使用 moto 模拟多个 AWS 服务

Question

我正在尝试模拟计算环境的创建，这需要一些其他资源，即 IAM 实例配置文件和服务角色。但是，当我创建这些 IAM 资源然后尝试在计算环境创建中使用它们时，事情失败了:

<Message>Role arn:aws:iam::123456789012:instance-profile/InstanceProfile not found</Message>

代码如下:

@mock_batch
@mock_iam
def test_create_compute_environment(lims_objs):
    client = boto3.client("batch")
    iam = boto3.resource("iam")
    service_role = iam.create_role(
        RoleName="BatchServiceRole", AssumeRolePolicyDocument="AWSBatchServiceRole"
    )
    instance_profile = iam.create_instance_profile(
        InstanceProfileName="InstanceProfile"
    )
    instance_profile.add_role(RoleName=service_role.name)
    for elem in iam.instance_profiles.all():
        print(elem, elem.arn)

    for elem in iam.roles.all():
        print(elem)

    response = client.create_compute_environment(
        computeEnvironmentName="compute_environment",
        type="MANAGED",
        state="ENABLED",
        computeResources={
            "type": "EC2",
            "minvCpus": 0,
            "maxvCpus": 256,
            "desiredvCpus": 2,
            "instanceTypes": ["optimal"],
            "imageId": "test",
            "subnets": [],
            "securityGroupIds": [],
            "ec2KeyPair": "",
            "instanceRole": instance_profile.arn,
            "tags": {},
        },
        serviceRole=service_role.arn,
    )

在测试中，我可以看到 IAM 对象的输出，所以我知道它们正在创建中。这些不是在 moto 模拟之间共享吗？

iam.InstanceProfile(name='InstanceProfile') arn:aws:iam::123456789012:instance-profile/InstanceProfile
iam.Role(name='BatchServiceRole')

我知道如果我们可以通过实例配置文件，这可能不是完整的工作示例，但这就是它现在卡住的地方。

非常感谢任何见解。非常感谢!

Answer 1

我能够克服这个问题，我希望这可以帮助其他人。简而言之，我创建了固定装置并在我需要它们的地方传递服务。

@pytest.fixture()
def vpc():
    mock = mock_ec2()
    mock.start()
    ec2 = boto3.resource("ec2")
    vpc = ec2.create_vpc(CidrBlock="172.16.0.0/16")
    vpc.wait_until_available()
    ec2.create_subnet(CidrBlock="172.16.0.1/24", VpcId=vpc.id)
    ec2.create_security_group(
        Description="Test security group", GroupName="sg1", VpcId=vpc.id
    )
    yield vpc
    mock.stop()


@pytest.fixture()
def iam_resource():
    mock = mock_iam()
    mock.start()
    yield boto3.resource("iam")
    mock.stop()


@pytest.fixture()
def batch_client():
    mock = mock_batch()
    mock.start()
    yield boto3.client("batch")
    mock.stop()


@pytest.fixture()
def batch_roles(iam_resource) -> Tuple[Any, Any]:
    iam = iam_resource

    service_role = iam.create_role(
        RoleName="BatchServiceRole",
        AssumeRolePolicyDocument=json.dumps(
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {"Service": "batch.amazonaws.com"},
                        "Action": "sts:AssumeRole",
                    }
                ],
            }
        ),
    )
    instance_role = iam.create_role(
        RoleName="InstanceRole",
        AssumeRolePolicyDocument=json.dumps(
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {"Service": "ec2.amazonaws.com"},
                        "Action": "sts:AssumeRole",
                    }
                ],
            }
        ),
    )
    return service_role, instance_role

@pytest.fixture()
def batch_compute_environments(
    lims_objs, batch_client, batch_roles, vpc
):
    ...

然后我能够使用这些和其他固定装置模拟测试提交作业，这些固定装置以与上述相同的方式创建。

def test_submit_batch(
    lims_objs,
    batch_client,
    batch_compute_environments,
    batch_job_definitions,
    batch_queue,
    capsys,
):
    client = batch_client
    for (env, assay), lims in lims_objs.items():
        name = f"pytest_batch_job_{env.name}_{assay.name}"
        res = client.submit_job(
            jobName="pytest_" + name,
            jobQueue=lims.get_aws_name("job_queue"),
            jobDefinition=lims.get_aws_name("job_definition"),
            parameters={
                "assay": "...",
                "runid": name,
                "reqid": "pytest",
                "env": env.name,
            },
        )

        assert res["ResponseMetadata"]["HTTPStatusCode"] == requests.codes.ok
        ...