【MongoDB实战】第12章 测试与部署：从开发到生产（python实战完善版）

《MongoDB实战入门》第12章 测试与部署：从开发到生产（python实战完善版）

本章聚焦MongoDB从开发测试到生产落地的全流程python实操，补充可直接复用的代码示例，确保每个技术点都能落地执行，核心围绕单元测试、生产部署、监控运维三大模块展开。

12.1 MongoDB单元测试

单元测试的核心目标是脱离真实MongoDB服务依赖，通过模拟环境验证数据操作逻辑的正确性，避免测试污染真实业务数据，python生态中常用pytest（测试框架）+mongomock（MongoDB模拟工具）实现该需求。

12.1.1 环境准备

首先安装所需依赖包：

pipinstallpytest mongomock pymongo python-dotenv

12.1.2 核心代码实现（覆盖CRUD+聚合+事务）

第一步：编写待测试的MongoDB业务操作类

创建mongodb_service.py，封装核心业务操作：

# mongodb_service.pyfrompymongoimportMongoClientfromdatetimeimportdatetimefromtypingimportList,Dict,Optional,UnionclassMongoDBBusinessService:"""MongoDB业务操作封装类，包含CRUD、聚合、事务等核心操作"""def__init__(self,mongo_client:MongoClient,db_name:str="business_db"):self.db=mongo_client[db_name]self.collection=self.db["user_orders"]# 订单业务集合# 新增数据（C）：创建订单defcreate_order(self,order_info:Dict)->str:""" 创建订单 :param order_info: 订单信息字典，必须包含user_id、order_amount字段 :return: 新增订单的ID字符串 :raises ValueError: 缺少必填字段时抛出异常 """required_fields=["user_id","order_amount"]ifnotall(fieldinorder_infoforfieldinrequired_fields):raiseValueError("订单信息缺少必填字段（user_id/order_amount）")# 补充默认字段order_info.update({"create_time":datetime.now(),"order_status":"pending","update_time":datetime.now()})result=self.collection.insert_one(order_info)returnstr(result.inserted_id)# 查询数据（R）：按用户ID查询订单defget_orders_by_user(self,user_id:str,limit:int=10)->List[Dict]:""" 按用户ID查询订单列表 :param user_id: 用户唯一标识 :param limit: 返回订单数量上限 :return: 订单列表（去除_id字段） """query={"user_id":user_id}projection={"_id":0}# 不返回_id字段orders=list(self.collection.find(query,projection).limit(limit).sort("create_time",-1))returnorders# 更新数据（U）：更新订单状态defupdate_order_status(self,order_id:str,new_status:str)->int:""" 更新订单状态 :param order_id: 订单ID :param new_status: 新状态（pending/paid/shipped/completed） :return: 受影响的记录数 """query={"_id":order_id}update_data={"$set":{"order_status":new_status,"update_time":datetime.now()}}result=self.collection.update_one(query,update_data)returnresult.modified_count# 删除数据（D）：删除过期无效订单defdelete_invalid_orders(self,user_id:str)->int:""" 删除用户的无效订单（状态为pending且创建时间超过24小时，仅示例） :param user_id: 用户唯一标识 :return: 删除的记录数 """# 模拟24小时前的时间（实际可根据业务调整）expire_time=datetime.now()-timedelta(hours=24)query={"user_id":user_id,"order_status":"pending","create_time":{"$lt":expire_time}}result=self.collection.delete_many(query)returnresult.deleted_count# 聚合查询：按用户统计订单总金额defstat_user_order_total(self)->List[Dict]:""" 聚合查询：统计每个用户的订单总金额、订单数量 :return: 聚合结果列表 """pipeline=[{"$group":{"_id":"$user_id","total_amount":{"$sum":"$order_amount"},"order_count":{"$sum":1}}},{"$sort":{"total_amount":-1}},{"$limit":20# 仅返回前20名用户}]returnlist(self.collection.aggregate(pipeline))# 事务操作：批量创建订单（需MongoDB 4.0+支持，mongomock兼容事务模拟）defbatch_create_orders(self,order_list:List[Dict])->bool:""" 事务批量创建订单，要么全部成功，要么全部回滚 :param order_list: 订单列表 :return: 批量创建是否成功 """ifnotorder_list:returnFalsewithself.db.client.start_session()assession:session.start_transaction()try:fororderinorder_list:self.create_order(order)session.commit_transaction()returnTrueexceptExceptionase:session.abort_transaction()raiseException(f"批量创建订单失败：{str(e)}")

第二步：编写单元测试用例

创建test_mongodb_service.py，使用pytest+mongomock实现无依赖测试：

# test_mongodb_service.pyimportpytestimportmongomockfrompymongoimportMongoClientfromdatetimeimporttimedelta,datetimefrommongodb_serviceimportMongoDBBusinessService# 测试夹具1：创建模拟MongoDB客户端（无需真实服务，全局复用）@pytest.fixture(scope="function")defmock_mongo_client():"""创建模拟MongoDB客户端，测试完成后自动清理数据"""client=mongomock.MongoClient()# 完全兼容pymongo APIyieldclient# 测试结束后删除测试数据库，避免数据残留client.drop_database("business_db")# 测试夹具2：初始化业务服务实例@pytest.fixture(scope="function")deforder_service(mock_mongo_client):"""初始化MongoDB业务服务，依赖模拟客户端"""returnMongoDBBusinessService(mock_mongo_client)# -------------- 测试用例：覆盖CRUD核心操作 --------------deftest_create_order_success(order_service):"""测试创建订单（正常场景）"""order_data={"use