并行运行pytest集成测试时In-memory DynamoDB（moto）报错排查

问题场景 #

使用Python 3.9 + moto 4.0.7搭建内存型DynamoDB做集成测试，单线程执行pytest -s tests/integration时测试完全正常，但通过pytest -n 10 -s tests/integration并行运行时，出现以下错误：

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

测试配置代码如下：

@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

问题根源 #

1. 共享moto服务实例：你用固定端口3000启动moto_server，且fixture的scope为session——在pytest-xdist并行模式下，主进程启动一个moto服务，所有工作进程都连接这个共享实例。当多个进程同时执行my_test_table fixture时，会同时调用CreateTable，而moto的内存数据库是全局共享的，第一个进程创建表后，其他进程再创建就会触发资源占用错误。
2. patch失效：pytest-xdist的工作进程是独立的Python进程，你在主进程中对core_aws.aws.resource/client的patch操作，不会同步到工作进程，这会导致测试逻辑混乱，进一步加剧资源冲突。

scope="session"本身不是问题，问题出在多进程共享同一个DynamoDB测试实例。

解决方案 #

1. 给每个工作进程分配独立的moto服务 #

使用随机空闲端口启动moto_server，确保每个工作进程拥有独立的测试数据库实例，避免资源竞争。可以借助port-for库自动获取空闲端口：

• 先安装依赖：pip install port-for

2. 修改fixture配置 #

调整后的代码如下：

import port_for
from subprocess import Popen
import boto3
from unittest.mock import patch
import core_aws.aws
from your_module import MyTestTable, tables
from typing import Any, Optional

@pytest.fixture(scope="session", name="dynamodb_server")
def dynamodb_server_fixture() -> str:
    # 随机选择空闲端口
    port = port_for.select_random()
    with Popen(["moto_server", f"-p{port}"]) as process:
        yield f"http://localhost:{port}"
        process.kill()

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

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

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

@pytest.fixture(scope="session", name="dynamodb_client")
def dynamodb_client_fixture(dynamodb_server: str) -> Any:
    client = boto3.client("dynamodb", endpoint_url=dynamodb_server)
    original_getter = core_aws.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", side_effect=client_override):
        yield client

@pytest.fixture(scope="session", autouse=True)
def my_test_table(dynamodb_resource: Any, dynamodb_client: Any) -> None:
    original = MyTestTable.__metadata__.table
    table_name = MyTestTable.__metadata__.table_name
    # 创建前先检查表是否存在，避免重复创建报错
    try:
        dynamodb_resource.Table(table_name).load()
    except dynamodb_resource.meta.client.exceptions.ResourceNotFoundException:
        MyTestTable.__metadata__.table = tables.create(dynamodb_resource, dynamodb_client, MyTestTable.__metadata__)
    yield
    MyTestTable.__metadata__.table = original

3. 额外注意事项 #

• 确保tables.create函数可以兼容表已存在的场景，或者像示例中那样提前检查表状态
• pytest-xdist的每个工作进程都会独立运行session级fixture，所以每个进程都会启动自己的moto服务，完全隔离测试数据
• 如果测试用例需要在不同进程间共享数据，这种方案不适用，但集成测试一般不需要跨进程共享测试数据

内容的提问来源于stack exchange，提问作者a13e