You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

测试时创建pymongo.MongoClient而非mongomock.MongoClient的问题

解决测试中MongoDB连接未被mongomock模拟的问题

运行测试时,程序会尝试创建真实的MongoDB连接,而非使用mongomock模拟;但移除__init__.py中的from .main import HashClient(注:根据项目结构推测应为from .hash_controller import HashClient)语句后,问题消失。需要保留该导入语句同时正常运行测试。

项目结构:

hash_controller/
|
|----> __init__.py
|----> hash_controller.py
|----> test_hash_controller.py

问题根源

核心问题是导入时机不匹配:
当测试模块执行时,from . import HashClient(通过__init__.py间接导入)会触发hash_controller.py的加载,而此时mongomock的patch还未生效,导致get_connection函数中导入的MongoClient是真实的客户端实例,而非mock版本。

解决方案

以下三种方案均可解决问题,按需选择:

方案1:确保mongomock Patch在所有导入前生效

调整测试文件的执行顺序,让mongomock.patch在任何涉及pymongo或HashClient的导入前生效:

修改test_hash_controller.py:

import os
import mongomock

# 先设置环境变量
os.environ["HASH_DB_USER"] = ""
os.environ["HASH_DB_PASS"] = ""
os.environ["HASH_DB_HOST"] = "localhost"
os.environ["HASH_DB_PORT"] = "27017"
os.environ["HASH_DB_NAME"] = "test"
os.environ["HASH_DB_COLLECTION"] = "test"

# 将所有依赖mock的导入操作包裹在patch上下文内
with mongomock.patch(servers=(('localhost', 27017),)):
    import pymongo
    # 直接导入或通过__init__导入HashClient,此时MongoClient已被mock
    from .hash_controller import HashClient
    client = pymongo.MongoClient("localhost")


def test_exist_false():
    hash_val = HashClient._get_hash("123", "user", {"user": "222"})
    client["test"]["test"].insert_one({"_id": hash_val})
    response = HashClient.exist("123", "user", {"user": "123"})
    assert response == False
    client.test.test.delete_many({})

方案2:重构业务代码实现依赖注入

修改hash_controller.py,让get_connection和HashClient.exist支持传入自定义连接/集合,从根源上解耦数据库依赖:

import hashlib
import json
import os
import time
from pymongo import MongoClient


def get_connection(client=None):
    if client:
        # 测试时传入mock client,跳过真实连接逻辑
        _database = client[os.environ["HASH_DB_NAME"]]
        return _database[os.environ["HASH_DB_COLLECTION"]]
    
    _db_host = os.environ["HASH_DB_HOST"]
    _db_port = os.environ["HASH_DB_PORT"]
    _db_name = os.environ["HASH_DB_NAME"]
    _db_collection = os.environ["HASH_DB_COLLECTION"]

    _client = MongoClient(f"mongodb://{_db_host}:{_db_port}/")
    _database = _client[_db_name]
    _collection = _database[_db_collection]
    return _collection


class HashClient:
    @staticmethod
    def exist(customer_uuid: str, type: str, obj, collection=None):
        HashClient._check_customer_and_type(customer_uuid, type)
        HashClient._check_obj(obj)
        id = HashClient._get_hash(customer_uuid, type, obj)
        # 优先使用传入的集合,否则调用默认连接
        collection = collection or get_connection()
        item = collection.find_one(id)
        return item is not None

对应的测试文件简化为:

import os
import mongomock

os.environ["HASH_DB_USER"] = ""
os.environ["HASH_DB_PASS"] = ""
os.environ["HASH_DB_HOST"] = "localhost"
os.environ["HASH_DB_PORT"] = "27017"
os.environ["HASH_DB_NAME"] = "test"
os.environ["HASH_DB_COLLECTION"] = "test"

# 创建mock集合
client = mongomock.MongoClient("localhost")
test_collection = client["test"]["test"]

# 此时导入不会触发真实连接
from . import HashClient


def test_exist_false():
    hash_val = HashClient._get_hash("123", "user", {"user": "222"})
    test_collection.insert_one({"_id": hash_val})
    # 传入mock集合执行测试
    response = HashClient.exist("123", "user", {"user": "123"}, collection=test_collection)
    assert response == False
    test_collection.delete_many({})

方案3:用pytest Monkeypatch替换连接函数

如果不想修改业务代码,可使用pytest的monkeypatch工具,在测试中替换get_connection函数为返回mock集合的版本:

import os
import mongomock
import pytest

os.environ["HASH_DB_USER"] = ""
os.environ["HASH_DB_PASS"] = ""
os.environ["HASH_DB_HOST"] = "localhost"
os.environ["HASH_DB_PORT"] = "27017"
os.environ["HASH_DB_NAME"] = "test"
os.environ["HASH_DB_COLLECTION"] = "test"

# 正常导入HashClient
from . import HashClient


def test_exist_false(monkeypatch):
    # 创建mock集合
    client = mongomock.MongoClient("localhost")
    test_collection = client["test"]["test"]
    
    # 替换get_connection函数
    def mock_get_connection():
        return test_collection
    
    monkeypatch.setattr("hash_controller.hash_controller.get_connection", mock_get_connection)
    
    hash_val = HashClient._get_hash("123", "user", {"user": "222"})
    test_collection.insert_one({"_id": hash_val})
    response = HashClient.exist("123", "user", {"user": "123"})
    assert response == False
    test_collection.delete_many({})

内容的提问来源于stack exchange,提问作者Ezequiel Marcel

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.07 03:54:51