[测试技术] MongoDB如何测试？脏数据、并发覆盖与主节点切换

原创内容，未获授权禁止转载、转发、抄袭。

MongoDB 接口返回写入成功，不能证明业务数据一定正确。字段类型漂移、重复业务单号、并发覆盖、从节点旧数据和主节点切换，都可能让“成功写入”变成错误结果。

本文不讲安装和基础 CRUD，而是用订单扣库存场景说明 MongoDB 应该怎么测：如何制造问题、保留证据，以及最终断言什么。

示例在 MongoDB 8.3.7、PyMongo 4.14.0 和本地 3 节点副本集上执行。这里的版本号用于说明实验环境，不代表生产升级建议。所有地址和业务数据均为测试数据，连接串为便于阅读省略了认证与 TLS，生产环境不能照搬。本地副本集只能验证 API 与选主行为，不能替代跨机房网络分区、磁盘故障和生产负载测试。

测试前先明确数据边界

MongoDB 的单文档写入具有原子性，但“单文档原子”不等于整条业务链原子。一次下单可能同时涉及订单、库存、流水和消息，测试前应先画清数据边界：

测试时至少保留orderNo、sku、业务版本号、MongoDB_id、请求traceId和事务结果。只比较集合总数不够，总数相同也可能同时存在一条漏数据和一条重复数据。

下面的片段假设使用隔离的全新测试库，统一连接副本集，并对关键写入使用多数派确认：

frompymongoimportMongoClientfrompymongo.write_concernimportWriteConcern client=MongoClient("mongodb://127.0.0.1:27117,127.0.0.1:27118,127.0.0.1:27119/""?replicaSet=rs0&retryWrites=true")db=client.get_database("blog_test",write_concern=WriteConcern("majority"))

w: "majority"表示写入得到计算出的多数数据承载投票成员确认，不代表随后从任意从节点读取都一定能立即看到最新值。写确认和读可见性要分开测试。客户端还应设置符合接口目标的选主、连接和写关注超时；发生超时只能说明客户端没有拿到确定结果，不能直接断言服务端没有写入，应先按业务键对账。

先让错误数据写不进去

MongoDB 的灵活 Schema 不等于没有 Schema。若金额有时写成 Decimal128、有时写成字符串，代码短期可能不报错，但查询、排序和聚合结果会逐渐失真。

可以用$jsonSchema限定核心字段，再给业务主键建立唯一索引：

db.createCollection("orders",{validator:{$jsonSchema:{bsonType:"object",required:["orderNo","amount"],properties:{orderNo:{bsonType:"string"},amount:{bsonType:"decimal"}}}}})db.orders.createIndex({orderNo:1},{unique:true})

先写入错误类型，再重复写入同一个订单号。下面用pytest捕获预期异常，避免第一条失败后脚本直接终止：

importpytestfrombson.decimal128importDecimal128frompymongo.errorsimportDuplicateKeyError,WriteErrorwithpytest.raises(WriteError)asvalidation_error:db.orders.insert_one({"orderNo":"O-invalid","amount":"99.00"})assertvalidation_error.value.code==121db.orders.insert_one({"orderNo":"O-001","amount":Decimal128("99.00")})withpytest.raises(DuplicateKeyError)asduplicate_error:db.orders.insert_one({"orderNo":"O-001","amount":Decimal128("100.00")})assertduplicate_error.value.code==11000assertdb.orders.count_documents({"orderNo":"O-001"})==1

本次实际结果如下：

错误金额类型 -> code=121 Document failed validation 重复 orderNo -> code=11000 DuplicateKey

自动化测试不要只断言“抛了异常”，还要检查：

• 错误码是否符合预期，失败文档是否确实没有入库；
• 合法边界值能否写入，例如 0、最大金额和可选字段缺失；
• 批量写入出现部分失败时，成功项、失败项和重试范围是否清楚；
• 重复请求被唯一索引拦截后，库存、积分和下游通知是否也没有重复执行。

唯一索引只能约束集合内的字段，不能自动提供整条接口的幂等性。业务还要使用稳定的幂等键，并保证副作用发生在正确的事务或补偿边界内。

存量集合补建唯一索引时，还要先准备重复数据。若索引创建失败，测试应输出冲突业务键并验证清理方案，而不是临时取消唯一约束后继续上线。

并发测试不能只发很多请求

库存初始值为 10，两个请求几乎同时购买 1 件。若应用采用“先读库存，再计算新值，最后$set”的方式，两个请求都可能读到 10，并都写回 9：

db.inventory.drop()db.inventory.create_index("sku",unique=True)db.inventory.insert_one({"sku":"SKU-1","stock":10,"version":1})a=db.inventory.find_one({"sku":"SKU-1"})b=db.inventory.find_one({"sku":"SKU-1"})db.inventory.update_one({"_id":a["_id"]},{"$set":{"stock":a["stock"]-1}})db.inventory.update_one({"_id":b["_id"]},{"$set":{"stock":b["stock"]-1}})

两次更新都成功，但本次实测库存为 9，而不是 8。这就是丢失更新：数据库没有报错，业务结果却错了。

简单扣减可以直接使用单文档原子更新，并把库存下限放进过滤条件：

db.inventory.update_one({"sku":"SKU-1"},{"$set":{"stock":10}})for_inrange(2):result=db.inventory.update_one({"sku":"SKU-1","stock":{"$gte":1}},{"$inc":{"stock":-1}})assertresult.modified_count==1assertdb.inventory.find_one({"sku":"SKU-1"})["stock"]==8

预期扣减成功时必须断言modified_count=1；返回 0 时，应转换为“库存不足”或并发冲突，不能继续创建成功订单。两次$inc后，本次实测库存从 10 正确变为 8。

当更新逻辑不能用单个操作表达时，可以增加业务版本号做乐观锁：

db.inventory.update_one({"sku":"SKU-1"},{"$set":{"stock":10,"version":1}})first=db.inventory.update_one({"sku":"SKU-1","version":1},{"$inc":{"stock":-1,"version":1}})stale=db.inventory.update_one({"sku":"SKU-1","version":1},{"$inc":{"stock":-1,"version":1}})assertfirst.matched_count==1assertstale.matched_count==0

两个请求都携带version=1时，本次结果为：

第一次 matched_count=1 第二次 matched_count=0

这里的核心断言不是接口都返回 200，而是：成功购买数等于库存减少量，库存永不小于 0，每个成功订单只产生一条流水，版本冲突有明确的重读、重算或失败策略。

读写一致性要拆开验证

副本集允许把读取分发到从节点，但从节点复制存在延迟。即使刚完成多数派写入，使用secondary读取偏好时，也不应把“立即读到最新值”当作无条件保证。

需要先按业务区分两类接口：

• 下单后立即查询、余额确认等强依赖新值的流程，通常读取主节点；
• 报表、历史列表等允许短暂延迟的流程，可以读取从节点，但要定义可接受的延迟和降级行为。

对于要求读取多数派已提交数据的查询，可显式设置readConcern: "majority"：

frompymongo.read_concernimportReadConcernfrompymongo.read_preferencesimportReadPreference orders=db.get_collection("orders",read_concern=ReadConcern("majority"),read_preference=ReadPreference.PRIMARY)assertorders.count_documents({"orderNo":"O-001"})==1

readConcern决定可以读取哪种确认级别的数据，readPreference决定从哪个成员读取，两者不是一回事。若业务必须从从节点实现“读到自己刚写的数据”，还要结合因果一致会话和多数派读写关注，并在真实拓扑中验证，不能仅靠切换secondaryPreferred推断正确性。

可以人为暂停从节点复制或注入网络延迟，然后分别从主、从节点按业务主键查询。测试应记录写入时间、读取节点、读取版本和复制延迟，并断言系统是等待、返回旧值、回退主节点还是明确提示延迟。

事务要验证回滚后的副作用

扣库存和写库存流水涉及两个文档，可以放进事务。测试重点不是“事务代码能运行”，而是在第二步失败后检查第一步是否真正回滚：

frompymongo.read_concernimportReadConcernfrompymongo.write_concernimportWriteConcern db.inventory.update_one({"sku":"SKU-1"},{"$set":{"stock":10}})db.ledger.delete_many({"orderNo":"O-TX"})withclient.start_session()assession:session.start_transaction(read_concern=ReadConcern("majority"),write_concern=WriteConcern("majority"))try:stock_result=db.inventory.update_one({"sku":"SKU-1","stock":{"$gte":1}},{"$inc":{"stock":-1}},session=session)ifstock_result.modified_count!=1:raiseRuntimeError("库存不足")db.ledger.insert_one({"orderNo":"O-TX","sku":"SKU-1"},session=session)raiseRuntimeError("模拟第二步之后失败")exceptRuntimeError:session.abort_transaction()assertdb.inventory.find_one({"sku":"SKU-1"})["stock"]==10assertdb.ledger.count_documents({"orderNo":"O-TX"})==0

初始库存为 10，本次强制中断后的实际结果是：

inventory.stock=10 ledger.count=0

还应继续覆盖以下故障点：

• 第一条更新条件不匹配，事务是否停止而不是继续写流水；
• 事务执行中主节点切换，驱动和业务如何处理可重试错误；
• 提交结果不确定时，是否按业务键查询后再决定重试；
• 重试整个事务后，是否产生重复订单、流水或库存扣减；
• 事务超时或并发冲突时，接口状态与最终数据是否一致。

PyMongo 的with_transaction()会根据TransientTransactionError重跑整个事务，并在UnknownTransactionCommitResult时重试提交；一次调用可能多次执行回调，因此回调不能包含不可重复的外部副作用。该方法最多重试 120 秒且不可配置，需要更短边界时应使用应用自己的事务处理逻辑。不能看到异常就盲目重放，也不能把 MongoDB 事务扩大到外部 HTTP 调用或消息发送。外部副作用需要 Outbox、幂等消费或补偿机制，并专门测试“数据库成功、通知失败”。

主节点切换要看恢复后的数据

本文使用 3 个具备投票权的数据节点验证副本集选主，生产测试拓扑还应按实际容灾目标设计。测试步骤如下：

1. 使用副本集 URI 写入订单，并确认多数派写入成功。
2. 记录当前主节点，通过进程终止或网络隔离让其不可用。
3. 等待驱动发现新主节点，不手工改成单节点连接串。
4. 继续写入新订单，再查询切换前后的业务主键。
5. 恢复旧节点，确认其重新加入副本集并追平数据。

本地测试先在127.0.0.1:27117写入O-001，随后停止该主节点；新主节点产生后写入O-AFTER-FAILOVER，实际结果如下：

old_primary=127.0.0.1:27117 new_primary=127.0.0.1:27118 post_failover_inserted=True old_and_new_orders=2

本次输出只覆盖前四步，没有记录旧节点恢复后的追平过程，因此只能证明驱动通过副本集地址发现了新主节点，并在选主后完成多数派写入。生产级测试还应关注：

• 选主窗口内接口是重试、失败还是长时间阻塞；
• 客户端超时是否大于故障恢复目标，连接池是否出现堆积；
• 重试写是否造成重复业务副作用；
• 未达到多数派确认的写入在故障后是否被回滚；
• 旧主节点恢复后是否出现数据回滚记录和告警；
• 跨机房延迟、网络分区和连续节点故障是否符合容灾目标。

retryWrites=true只会重试驱动支持的单文档写操作，不能替业务重试整个下单流程，也不能防止 HTTP、消息等外部副作用重复。测试最终要回到订单、库存和流水：切换前已确认的数据不能丢，切换期间结果不确定的请求可以对账，恢复后新请求能够继续处理。

查询性能要断言执行计划

接口响应变慢时，只看平均耗时很难判断是索引、数据量还是环境波动。对核心查询执行explain("executionStats")，至少记录返回数、扫描键数和扫描文档数：

db.inventory.explain("executionStats").find({sku:"SKU-1"})

给sku建立索引后，本次查询 101 条测试数据中的目标记录，结果为：

nReturned=1 totalKeysExamined=1 totalDocsExamined=1

测试中可以为固定数据集设置合理阈值，但不要把某个执行计划节点写死为所有版本都必须相同。更有价值的断言是扫描量没有随总数据量线性增长、排序没有意外落到内存、返回条数与业务条件一致。还要覆盖组合条件、排序、分页、空结果和低选择性字段，防止单字段样例通过而真实查询仍走全表扫描。

性能测试应使用接近生产的数据量和字段分布，并关注慢查询、CPU、磁盘、缓存命中、连接池和复制延迟。几百条本地数据只能验证索引是否被使用，不能得出容量结论。

本文未展开分片集群。若生产使用 Sharding，还应单独验证分片键分布、热点、Chunk 迁移、跨分片查询和分布式事务，不能用副本集结果替代这些场景。

一套最小回归集

总结

MongoDB 测试的重点不是 CRUD 能否成功，而是成功之后的数据是否满足业务约束。先用 Schema Validation 和唯一索引挡住脏数据，再通过并发覆盖、事务中断、从节点延迟和主节点切换主动制造失败，最后沿着订单、库存、流水和外部副作用完成对账。

真正有效的断言通常落在三个问题上：错误数据有没有被拒绝，并发结果有没有被覆盖，故障恢复后业务状态能不能解释清楚。把这三点测透，比堆几十条普通增删改查用例更接近 MongoDB 的真实风险。