Other tests interfering with Moto mocks

[tomelliff]

I had some tests working fine when I was initialising the relevant clients in the functions that I was then testing. My tests looked something like this:

@mock_s3
def test_write_json_to_s3():
    j = {"foo": 1, "bar": True}
    s3 = boto3.resource("s3")
    bucket_name = "mock-s3-bucket-example"
    os.environ["S3_BUCKET"] = bucket_name
    bucket = s3.Bucket(bucket_name)
    bucket.create()
    assert len(list(bucket.objects.all())) == 0

    event_handler.write_json_to_s3(j)

    objects = list(bucket.objects.all())
    assert len(objects) == 1
    content = json.loads(objects[0].get()["Body"].read().decode("ascii"))
    assert content == j

Unfortunately this has quite a lot of overhead for Lambda functions so I moved the clients out to the global scope which then broke all my tests. I switched to importing my production code at the test function level and also followed the Pytest fixture examples in the README so now my tests look like this:

@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"
    os.environ["AWS_REGION"] = "us-east-1"
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"

@pytest.fixture(scope="function")
def ec2(aws_credentials):
    with mock_ec2():
        yield boto3.client("ec2", region_name="us-east-1")

@pytest.fixture(scope="function")
def s3(aws_credentials):
    with mock_s3():
        yield boto3.resource("s3", region_name="us-east-1")

# ...

def test_write_json_to_s3(s3):
    import event_handler

    now = datetime.utcnow()
    j = {
        "LaunchTime": now,
        "Time": f"{now}Z",
        "foo": 1,
        "bar": True,
    }
    bucket_name = "mock-s3-bucket-example"
    os.environ["S3_BUCKET"] = bucket_name
    bucket = s3.Bucket(bucket_name)
    bucket.create()
    assert len(list(bucket.objects.all())) == 0

    event_handler.write_json_to_s3(j, "foo_event")

    objects = list(bucket.objects.all())
    assert len(objects) == 1
    obj = objects[0]
    key_parts = obj.key.split("/")
    assert len(key_parts) == 5
    assert key_parts[0] == "foo_event"
    assert key_parts[1] == f"year={now.year}"
    assert key_parts[2] == f"month={now.month}"
    assert key_parts[3] == f"day={now.day}"
    assert key_parts[-1].endswith(".json")
    content = json.loads(obj.get()["Body"].read().decode("ascii"))
    assert content["foo"] == j["foo"]
    assert content["bar"] == j["bar"]
    assert content["LaunchTime"] == now.isoformat()

If I run this test by itself it passes. If I run it with other tests, including some others that use Moto, then it also passes. However if I attempt to run this test with one of a couple of other tests then it fails with NoCredentialsError which implies Moto is no longer working for some reason. The other tests that induce this behaviour pass as expected.

I've isolated the tests that cause this behaviour to these three:

def test_lambda_handler_throws_for_non_ec2_events():
    import event_handler

    ecs_event = {"source": "aws.ecs"}

    with pytest.raises(NotImplementedError):
        event_handler.lambda_handler(ecs_event, "context")

def test_spot_instance_interruption_event_is_handled(mocker):
    import event_handler

    event = example_spot_instance_interruption_event("i-123456")
    mocker.patch("event_handler.build_json_output")
    mocker.patch("event_handler.write_json_to_s3")

    event_handler.lambda_handler(event, "context")

    event_handler.write_json_to_s3.assert_called_with(unittest.mock.ANY, event_type="spot_instance_interruption")

def test_instance_state_change_event_is_handled(mocker):
    import event_handler

    event = example_instance_state_change_event("i-123456", "running")
    mocker.patch("event_handler.build_json_output")
    mocker.patch("event_handler.write_json_to_s3")

    event_handler.lambda_handler(event, "context")

    event_handler.write_json_to_s3.assert_called_with(unittest.mock.ANY, event_type="instance_state_change")

I can't see why these tests would cause the Moto mocks to fail to properly initialise but other tests are happy to run in the same run together and the Moto mocks work as expected so the tests pass.

I can also provide the whole production code and the full set of tests if that helps but I don't know how to recreate this as a more minimal example at this point because I've never seen this odd behaviour before.

I suspect this might be missing something about Pytest or fixtures but I'm completely stumped at this point. What am I missing here?

[bblommers]

Ignore my earlier (now-deleted) comment - I misread the question.

The new test-setup is due to the fact that moto has be initialized before any boto3-clients are created. Once a boto3-client is created, it's not possible to retro-actively mock it.

With this combination of mocked and patched tests, I imagine this might be what's happening:

• Test 1 runs, and initializes moto
• Test 1 creates a boto3-client (mocked)
• Test 1 destroys moto and any mocks after the test finishes
• Test 2 creates a boto3-client (patched - no moto involvement)
• Test 3 runs, and initialies moto
• Test 3 re-uses the previously created boto3-client that's somehow still cached
• Moto will never patch the boto3-client for test3, because the client was already initalized

Is it possible to change the order of your tests, and run the patched methods first, and all moto-tests afterwards? If that passes, it would validate the theory.

Another way could be to still initalize moto in the patch-tests. If everything is patched, the addition of a moto-fixture shouldn't change the behaviour of the test itself.

[tomelliff]

Passing any moto mock (either s3 or ec2) to just the test with pytest.raises seems to fix this without any other changes now.

Your explanation seemed to make sense to me for the other tests that were using the mocker but I'm confused why it's the pytest.raises one that should be the issue here.

The production code looks like this:

def lambda_handler(event, _):
    implemented_event_types = {
        "EC2 Spot Instance Interruption Warning": "spot_instance_interruption",
        "EC2 Instance State-change Notification": "instance_state_change",
    }

    if event["source"] != "aws.ec2":
        raise NotImplementedError
    detail_type = event["detail-type"]
    if detail_type not in implemented_event_types:
        raise NotImplementedError

    output = build_json_output(event)
    write_json_to_s3(output, event_type=implemented_event_types[detail_type])

So to me it seems like the assertion should be hit and the test exit before anything of interest with AWS clients happens. If it was just that the import was causing it because that triggers them to be initialised then I can't see why the other tests above or some other tests that have no mocking/patching in (they're testing methods with no side effect) don't cause this behaviour with the import.

[bblommers]

Might be the caching at work again. If the pytest.raises-test simply happens to be the first one that's executed, it will cache a mocked boto3-client in the event_handler. The cached client is then re-used in the mocker-instances.

If you forcefully reload the event_handler in the mocker-tests, it should re-initialize the boto3-client (without mocks), and cause the same failure again.

import eventhandler
import importlib
importlib.reload(eventhandler)

[bblommers]

Going to close this due to a lack of response, but feel free to let me know if this is still an issue.

[CynanX]

Came across this today and the issue was exactly as described.

Adding the moto import into my non-moto tests so it would be initialised before patching happened fixed my issues.