Search code examples
pythonpytestboto3integration-testingpytest-fixtures

In-memory dynamodb for tests that run in parallel


I am using python 3.9 and in my project I have bunch of integration tests that runs with in-memory dynamo-db using moto 4.0.7. My configuration setup looks like this -

@pytest.fixture(scope="session", name="dynamodb_server")
def dynamodb_server_fixture() -> Emitter[str]:
    with Popen(["moto_server", "-p3000"]) as process:  # nosec
        yield "http://localhost:3000"
        process.kill()


@pytest.fixture(scope="session", name="dynamodb_resource")
def dynamodb_resource_fixture(dynamodb_server: str) -> Emitter[Any]:
    resource = boto3.resource("dynamodb", endpoint_url=dynamodb_server)
    original_getter = aws.resource

    def resource_override(name: str) -> Any:
        if name == "dynamodb":
            return resource
        return original_getter(name)

    with patch("core_aws.aws.resource") as function:
        function.side_effect = resource_override
        yield resource


@pytest.fixture(scope="session", name="dynamodb_client")
def dynamodb_client_fixture(dynamodb_server: str) -> Emitter[Any]:
    client = boto3.client("dynamodb", endpoint_url=dynamodb_server)
    original_getter = aws.client

    def client_override(name: str, endpoint_url: Optional[str] = None) -> Any:
        if name == "dynamodb":
            return client
        return original_getter(name, endpoint_url)

    with patch("core_aws.aws.client") as function:
        function.side_effect = client_override
        yield client


@pytest.fixture(scope="session", autouse=True)
def my_test_table(dynamodb_resource: Any, dynamodb_client: Any) -> Emitter[None]:
    original = MyTestTable.__metadata__.table
    MyTestTable.__metadata__.table = tables.create(dynamodb_resource, dynamodb_client, MyTestTable.__metadata__)
    yield
    MyTestTable.__metadata__.table = original

As you can understand that MyTestTable is yielded and made available for all my integration tests. So pytest -s tests/integration works perfectly. But if I want to run the tests parallely using pytest -n 10 -s tests/integration, then it start failing with the error -

botocore.errorfactory.ResourceInUseException: An error occurred (ResourceInUseException) when calling the CreateTable operation: Resource in use

Can someone please let me know what is wrong with my configuration? Is scope="session" the culprit here?


Solution

  • Simple solution is remove autouse=True from all fixtures responsible to create tables. I am now able to run all the tests in parallel.