如何在GCP集成测试中模拟ConnectionError并验证事务回滚

一、本地开发阶段：用Mock工具主动触发错误 #

• 用对应语言的Mock框架替换Cloud Storage和Firestore的客户端实例，强制抛出目标错误：
  • 针对Cloud Storage上传步骤，在调用上传方法时抛出ConnectionError或TimeoutError，验证事务是否触发回滚（确认Mock上传标记为失败，无实际文件残留）。
  • 针对Firestore写入步骤，在调用set()/add()方法时抛出服务不可用类错误，检查Cloud Storage侧是否执行了文件删除操作。
• 示例（Python）：

  from unittest.mock import patch
  from google.cloud.exceptions import ServiceUnavailable
  
  def test_transaction_rollback_on_storage_timeout():
      with patch('google.cloud.storage.Blob.upload_from_file') as mock_upload:
          # 模拟Cloud Storage超时错误
          mock_upload.side_effect = ServiceUnavailable("Connection timed out")
          # 调用API处理器
          result = your_api_handler(test_file_obj, test_metadata)
          # 验证结果与回滚行为
          assert result["status"] == "failed"
          mock_upload.assert_called_once()
          # 额外校验Firestore无写入记录
  

二、集成测试阶段：利用云服务与网络工具复现真实故障 #

• 云服务内置故障注入：
  • 使用Google Cloud的故障注入功能，针对测试专用的存储桶/数据库临时配置故障规则：比如给测试存储桶设置强制断开连接的规则，或让Firestore测试库模拟服务不可用状态（操作前务必备份测试数据）。
• 网络层拦截模拟：
  • 在测试环境的网络节点（如Docker测试集群、测试机），用tc（Linux）或类似工具限制带宽、增加延迟，直接阻断Cloud Storage/Firestore的服务IP段，触发超时或连接错误：

    # 给测试机网卡添加10秒延迟，模拟超时
    tc qdisc add dev eth0 root netem delay 10000ms
    # 测试完成后清理规则
    tc qdisc del dev eth0 root netem
    

三、事务回滚逻辑的关键验证点 #

• 确保错误捕获后执行反向操作：如果Cloud Storage上传成功但Firestore写入失败，必须调用delete()方法删除已上传文件。测试时可故意让Firestore写入失败，检查存储桶中是否无残留文件。
• 测试幂等性：重复触发失败的请求，确认不会产生脏数据（比如重复上传文件、写入半完成的元数据）。

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