flask-pymongo+pytest测试隔离问题：如何控制MongoDB可用性？

• pymongo 4.3.3
• python 3.10.12
• flask-pymongo 2.3.0
• pytest 7.3.1

问题描述 #

需要解决Flask-PyMongo集成测试中模拟MongoDB宕机的隔离问题：当前测试受执行顺序与fixture使用影响，无法独立验证MongoDB可用/不可用两种场景。

现有测试代码 #

<other imports>
from app.app import app_module

def test_available(app_instance):
    assert app_module.check_mongodb_availability(app_instance) == True

def test_unavailable():
    test_app = app_module.create_app({'MONGO_URI': 'mongodb://invalid_uri:27017'})
    result = app_module.check_mongodb_availability(test_app)
    assert result is False

问题现象 #

• setup_environment（session作用域fixture）会启动MongoDB与Flask容器，供test_available等测试使用
• 若test_unavailable之前有测试使用过mongo_client fixture，该测试会意外失败（ping返回True）
• 若不提前使用mongo_client，test_available会失败；若完全不启动Mongo容器，test_unavailable能通过但会破坏其他测试

上下文代码 #

Flask应用核心代码 #

from flask_pymongo import PyMongo 
from pymongo.errors import ConnectionFailure 
<other imports>

mongo = PyMongo()

def check_mongodb_availability(app: Flask, max_retries: int = 5, delay: int = 2) -> bool:
    for attempt in range(max_retries): 
        try: 
            with app.app_context():             
                mongo.db.command('ping') 
                return True 
        except ConnectionFailure: 
            time.sleep(delay) 
    return False

def create_app(test_config=None):
    # Setup 
    app = Flask(__name__, instance_relative_config=True)
    settings = AppSettings()
    app.config['MONGO_URI'] = settings.get('mongo_uri')
    <other configs>

    # Apply test config if provided
    if test_config:
        app.config.update(test_config)
    
    # Initialize Pymongo
    mongo.init_app(app)
    
    <routes>
    
    return app

conftest.py fixture代码 #

<other imports>
import pytest
from dotenv import load_dotenv
from pymongo import MongoClient

@pytest.fixture(scope='session')
def setup_environment():
    # Set the mongo uri as an environment variable
    <read in params to build mongo uri>
    os.environ['MONGO_URI'] = f"mongodb://{uname}:{pword}@localhost:{port}/{db}"

    # Run the flask and mongo services
    subprocess.run(<spins up flask and mongodb docker containers>)

    yield

    # Clean up
    os.environ.pop('MONGO_URI', None)
    subprocess.run(<tears down flask and mongodb docker containers>)

@pytest.fixture(scope='function')
def mongo_client(setup_environment, directConnection=True): 
    # Note: I tried directConnection=False as well
    mongo_uri = os.environ['MONGO_URI']
    client = MongoClient(mongo_uri)
    try:
        yield client
    finally:
        client.close()

@pytest.fixture(scope='function')
def app_instance(request):
    from app.app import create_app
    return create_app()

<more fixtures for app context and app client>

问题原因分析 #

1. 全局PyMongo实例的客户端复用：Flask-PyMongo的PyMongo是全局单例，当之前的测试用有效URI初始化过该实例后，后续即使给新app传入无效URI，该实例可能复用之前创建的有效客户端，而非为新app创建新客户端。
2. pymongo连接池缓存：pymongo的MongoClient会缓存连接池，即使创建新的无效URI客户端，也可能复用之前的有效连接池，导致ping操作成功。
3. 上下文残留：测试间的app上下文未完全清理，导致全局mongo实例仍绑定之前的有效app配置。

解决方法 #

方法1：为每个app创建独立PyMongo实例（彻底隔离） #

修改Flask应用代码，移除全局mongo实例，在create_app内为每个app创建独立实例：

from flask_pymongo import PyMongo 
from pymongo.errors import ConnectionFailure 
<other imports>

def check_mongodb_availability(app: Flask, max_retries: int = 5, delay: int = 2) -> bool:
    # 从当前app的extensions获取专属PyMongo实例
    mongo = app.extensions['pymongo'][0]
    for attempt in range(max_retries): 
        try: 
            with app.app_context():             
                mongo.db.command('ping') 
                return True 
        except ConnectionFailure: 
            time.sleep(delay) 
    return False

def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)
    settings = AppSettings()
    app.config['MONGO_URI'] = settings.get('mongo_uri')
    
    if test_config:
        app.config.update(test_config)
    
    # 为当前app创建独立PyMongo实例
    mongo = PyMongo(app)
    
    <routes>
    
    return app

方法2：重置全局PyMongo实例的客户端缓存 #

在conftest.py添加fixture，清空全局mongo实例的客户端缓存：

@pytest.fixture(scope='function')
def reset_mongo_instance():
    from app.app import mongo
    # 清空PyMongo实例的客户端缓存
    mongo._clients.clear()
    yield

在测试中使用该fixture：

def test_available(reset_mongo_instance, app_instance):
    assert app_module.check_mongodb_availability(app_instance) == True

def test_unavailable(reset_mongo_instance):
    test_app = app_module.create_app({'MONGO_URI': 'mongodb://invalid_uri:27017'})
    result = app_module.check_mongodb_availability(test_app)
    assert result is False

方法3：隔离不可用场景的测试环境 #

为test_unavailable添加标记，跳过session级别的容器启动：

import pytest

@pytest.mark.skip_setup_environment
def test_unavailable():
    test_app = app_module.create_app({'MONGO_URI': 'mongodb://invalid_uri:27017'})
    result = app_module.check_mongodb_availability(test_app)
    assert result is False

修改setup_environment fixture：

@pytest.fixture(scope='session')
def setup_environment(request):
    # 检查是否有测试需要跳过容器启动
    skip_setup = any(item.name == 'skip_setup_environment' for item in request.node.items)
    if skip_setup:
        yield
        return

    # 原容器启动逻辑
    <read in params to build mongo uri>
    os.environ['MONGO_URI'] = f"mongodb://{uname}:{pword}@localhost:{port}/{db}"
    subprocess.run(<spins up flask and mongodb docker containers>)

    yield

    # 原清理逻辑
    os.environ.pop('MONGO_URI', None)
    subprocess.run(<tears down flask and mongodb docker containers>)

方法4：使用Mock模拟连接失败（推荐单元测试） #

无需依赖真实环境，直接Mock连接失败场景：

from unittest.mock import patch
from pymongo.errors import ConnectionFailure

def test_unavailable():
    test_app = app_module.create_app({'MONGO_URI': 'mongodb://invalid_uri:27017'})
    # Mock ping命令抛出ConnectionFailure
    with patch.object(test_app.extensions['pymongo'][0].db, 'command') as mock_command:
        mock_command.side_effect = ConnectionFailure("MongoDB unavailable")
        result = app_module.check_mongodb_availability(test_app)
        assert result is False

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