测试时创建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
相关产品推荐
相关产品推荐

