Python unittest.mock 中异常方法调用计数问题详解与解决方案

# upload_service.py
import json
import logging

# 假设 GoogleCloudError 和 Blob 是外部库的类
class GoogleCloudError(Exception):
    pass

class Blob:
    def __init__(self, name, bucket):
        self.name = name
        self.bucket = bucket

    def upload_from_string(self, data, content_type):
        print(f"Uploading data: {data} to {self.name} in {self.bucket}")
        # 模拟实际的上传逻辑，这里简化
        if "error" in data: # 示例：模拟特定数据触发异常
            raise GoogleCloudError("Simulated upload error")
        return True

class UploadService:
    def __init__(self, name, gcs_bucket):
        self.name = name
        self.gcs_bucket = gcs_bucket

    def upload(self, data):
        try:
            gcs_blob = Blob(self.name, self.gcs_bucket)
            gcs_blob.upload_from_string(data=json.dumps(data), content_type="application/json")
            return "Upload successful"
        except GoogleCloudError as e:
            logging.exception("Error uploading file")
            return f"Upload failed: {e}"

# test_upload_service.py
import unittest
from unittest.mock import patch
from upload_service import UploadService, GoogleCloudError, Blob # 导入实际的Blob和GoogleCloudError

class TestUploadService(unittest.TestCase):
    def test_upload_failure(self):
        us = UploadService("my_file", "my_bucket")
        with patch("upload_service.Blob") as mocked_blob_class:
            # mocked_blob_class 是 Blob 类本身的 Mock
            # gcs_blob 是 Blob 实例的 Mock
            gcs_blob_instance = mocked_blob_class.return_value
            gcs_blob_instance.upload_from_string.side_effect = GoogleCloudError("Google Cloud error")

            result = us.upload({"status": "error"}) # 调用会触发异常
            self.assertIn("Upload failed", result)
            # 错误的断言方式：
            # self.assertEqual(1, mocked_blob_class.upload_from_string.call_count) # ❌ 实际会是 0

# test_upload_service.py (续)

class TestUploadService(unittest.TestCase):
    def test_upload_failure_corrected(self):
        us = UploadService("my_file", "my_bucket")
        with patch("upload_service.Blob") as mocked_blob_class:
            gcs_blob_instance = mocked_blob_class.return_value
            gcs_blob_instance.upload_from_string.side_effect = GoogleCloudError("Google Cloud error")

            result = us.upload({"status": "error"})
            self.assertIn("Upload failed", result)

            # 正确的断言方式：
            self.assertEqual(1, gcs_blob_instance.upload_from_string.call_count)
            # 或者直接通过 mocked_blob_class().upload_from_string.call_count 访问
            self.assertEqual(1, mocked_blob_class().upload_from_string.call_count) # 这两种方式等价