MongoDB Atlas Local测试容器与本地容器隔离异常问题求助

背景与问题 #

我们正在迁移至MongoDB Atlas，用mongodb/mongodb-atlas-local容器做本地开发（端口绑定27017:27017），更新集成测试改用该容器时出现异常：

• 测试生成的testdb会出现在本地开发容器中
• 连接测试容器的随机主机端口（比如37017），能看到本地开发容器里的数据库
• 本地开发容器关闭时，除非把测试容器绑定到27017:27017，否则无法访问测试数据库

之前正常运行的旧代码（基于mongo镜像） #

@pytest.fixture(autouse=True, scope="session")
def init_mongo_get_uri():
    mongo_container = DockerContainer("mongo:latest")
    mongo_bind_port = randint(a=30000, b=40000)

    mongo_container.with_name("python_testing_mongodb")
    mongo_container.with_exposed_ports(mongo_bind_port)
    mongo_container.with_bind_ports(container=mongo_bind_port, host=mongo_bind_port)
    mongo_container.with_command(f"--port={mongo_bind_port} --replSet=rs")
    mongo_container.start()

    # 等待副本集初始化完成
    mongo_uri = f"mongodb://localhost:{mongo_bind_port}/?replicaSet=rs"
    max_attempts = 10
    for attempt in range(max_attempts):
        exit_code, output = mongo_container.exec(
            "mongosh --quiet --eval=\"rs.initiate({_id:'rs',members:[{_id:0,host:'localhost:%s'}]})\" mongodb://localhost:%s" % (
                mongo_bind_port, mongo_bind_port)
        )

        if exit_code == 0:
            break
        sleep(1)

        if attempt == max_attempts - 1:
            raise RuntimeError("Replica set didn't get setup properly after 10 seconds")

    yield mongo_uri

    mongo_container.stop()


@pytest.fixture(scope="session")
def test_app(init_mongo_get_uri):
    if not init_mongo_get_uri:
        raise RuntimeError("MongoDB Testcontainer URI is None in test_app fixture.")

    MongoDBSingleton._instance = None
    MongoDBSingleton(uri=init_mongo_get_uri, db_name="testdb")
    # 启动flask app

改用Atlas Local后的新代码（出现问题的版本） #

@pytest.fixture(autouse=True, scope="session")
def init_mongo_get_uri():
    mongo_container = DockerContainer("mongodb/mongodb-atlas-local")
    mongo_bind_port = randint(a=30000, b=40000)

    mongo_container.with_name("python_testing_mongodb")
    mongo_container.with_bind_ports(container=27017, host=mongo_bind_port)
    mongo_container.start()

    mongo_uri = f"mongodb://localhost:{mongo_bind_port}"

    yield mongo_uri

    mongo_container.stop()


@pytest.fixture(scope="session")
def test_app(init_mongo_get_uri):
# 逻辑和之前一致

问题原因 #

MongoDB Atlas Local镜像默认启动了名为rs0的副本集，但副本集节点的配置里，host是容器内部的localhost:27017。当测试代码通过随机主机端口连接容器后，MongoDB客户端会自动执行副本集节点发现，拿到容器内部的27017端口地址，然后直接访问本地主机的27017端口（也就是本地开发容器），导致两个容器的数据互通，出现“污染”。

而之前用mongo镜像时，我们手动初始化副本集，指定了节点的host为localhost:随机主机端口，客户端发现节点时会正确指向测试容器的端口，所以没有问题。

解决方案 #

核心是修改Atlas Local容器的副本集配置，把节点的外部访问地址设置为localhost:随机主机端口，让客户端能正确指向测试容器。

修复后的完整代码 #

import time
from random import randint
from testcontainers.docker_client import DockerContainer
import pymongo
from pymongo.errors import ConnectionFailure

@pytest.fixture(autouse=True, scope="session")
def init_mongo_get_uri():
    mongo_container = DockerContainer("mongodb/mongodb-atlas-local")
    mongo_bind_port = randint(a=30000, b=40000)

    mongo_container.with_name("python_testing_mongodb")
    mongo_container.with_bind_ports(container=27017, host=mongo_bind_port)
    mongo_container.start()

    # 等待MongoDB服务就绪
    mongo_uri = f"mongodb://localhost:{mongo_bind_port}/?replicaSet=rs0"
    max_attempts = 15
    client = None
    for attempt in range(max_attempts):
        try:
            client = pymongo.MongoClient(mongo_uri)
            # 检查连接是否正常
            client.admin.command("ping")
            break
        except ConnectionFailure:
            time.sleep(1)
            if attempt == max_attempts - 1:
                raise RuntimeError("Failed to connect to MongoDB Atlas Local container after 15 seconds")

    # 修改副本集配置，将节点host设置为localhost:主机绑定端口
    # 先获取当前副本集配置
    config = client.admin.command("replSetGetConfig")
    # 修改第一个节点的host
    config["config"]["members"][0]["host"] = f"localhost:{mongo_bind_port}"
    # 应用新配置
    client.admin.command("replSetReconfig", config["config"])

    yield mongo_uri

    mongo_container.stop()


@pytest.fixture(scope="session")
def test_app(init_mongo_get_uri):
    if not init_mongo_get_uri:
        raise RuntimeError("MongoDB Testcontainer URI is None in test_app fixture.")

    MongoDBSingleton._instance = None
    MongoDBSingleton(uri=init_mongo_get_uri, db_name="testdb")
    # 启动flask app

关键修改点说明 #

1. 等待容器就绪：用pymongo连接并执行ping命令，确保MongoDB服务完全启动后再修改配置
2. 修改副本集节点地址：通过replSetGetConfig获取当前配置，把节点的host改成localhost:随机主机端口，再用replSetReconfig应用配置，这样客户端发现节点时会指向测试容器的端口
3. URI指定副本集：连接URI必须加上?replicaSet=rs0，确保客户端走副本集发现流程，使用我们修改后的节点地址

额外验证 #

• 启动测试后，用Compass连接测试容器的随机端口（比如37017），现在只能看到测试生成的testdb，不会看到本地开发容器的数据库
• 本地开发容器运行时，测试生成的数据不会再出现在开发容器中
• 关闭本地开发容器后，测试容器的数据库依然可以正常访问

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