Mocking multiple AWS services with moto

Question

I'm trying to mock the creation of a compute environment, which requires some other resources, namely an IAM instance profile and service role. However, when I create those IAM resources and then attempt to use them in the compute environment creation, things fail with:

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

The code is below:

@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,
    )

In the test, I can see the prints for the IAM objects, so I know they are being created. Are these just not shared across moto mocks?

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

I know this may not be the complete working example if we can get past the instance profile, but this is where it's stuck now.

Any insight is much appreciated. Thanks so much!

Answer 1

I was able to get past this, and I hope this can help others. Briefly, I created fixtures and passed services around where I needed them.

@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
):
    ...

I was then able to mock out test submitting jobs using these and other fixtures, created in the same way as the above.

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
        ...