FastAPI中GCS上传方法单元测试：Mock Storage Client后仍触发凭证查找错误的求助

问题背景 #

我最近在给FastAPI项目里的GCS文件上传功能写单元测试，结果碰上个棘手的问题——明明已经用unittest.mock把Google Storage的Client给Mock掉了，跑测试的时候还是会触发真实的凭证校验逻辑，直接抛出No key could be detected的错误，完全绕不开真实的GCS客户端初始化流程。

我的GCS上传实现（cloud_service.py） #

from io import BytesIO
from google.auth.exceptions import DefaultCredentialsError
from google.cloud import storage
from fastapi import HTTPException

# 模块级别初始化Storage Client
storage_client = storage.Client()
bucket_name = "stupid-bucket-name-1234567890"

async def upload_to_gcs(file, file_name: str) -> str:
    try:
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(file_name)
        binary_data = await file.read()
        blob.upload_from_file(BytesIO(binary_data), content_type=file.content_type)
        return blob.public_url
    except DefaultCredentialsError as exe:
        raise HTTPException(status_code=500, detail=str(exe))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

我的单元测试代码 #

import sys
import pytest
from unittest.mock import patch, MagicMock, AsyncMock
from fastapi import HTTPException
from google.auth.exceptions import DefaultCredentialsError

# 尝试先Patch再导入模块
with patch("package.svc.cloud.storage.Client") as mock_client:
    mock_storage_client = mock_client.return_value
    mock_bucket = MagicMock()
    mock_blob = MagicMock()
    mock_storage_client.bucket.return_value = mock_bucket
    mock_bucket.blob.return_value = mock_blob
    mock_blob.public_url = "https://fake-url.com/file.txt"

    import importlib
    cloud_service = importlib.import_module("package.svc.cloud")

@pytest.mark.asyncio
async def test_upload_file_to_gcs_success():
    mock_file = AsyncMock()
    mock_file.read.return_value = b"test data"
    mock_file.content_type = "text/plain"
    # 这里函数名和实际定义有笔误，不过当前先不管这个
    url = await cloud_service.upload_file_to_gcs(mock_file, "file.txt")
    assert url == "https://fake-url.com/file.txt"
    cloud_service.storage_client.bucket.return_value.blob.return_value.upload_from_file.assert_called_once()

错误栈关键片段 #

E google.auth.exceptions.MalformedError: No key could be detected.

...

requestprocessor/services/cloud_service.py:11: in <module>
    storage_client = storage.Client()
    ^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/cloud/storage/client.py:247: in __init__
    super(Client, self).__init__(
...
E google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials.

问题排查 #

盯着错误栈看了半天，终于发现问题出在模块加载的时机：当我用importlib.import_module导入cloud_service的时候，模块里的storage_client = storage.Client()这行代码会立即执行，而我之前的Patch路径根本不对，导致Mock没有覆盖到这个调用，直接触发了真实的GCS Client初始化，自然就去查凭证了。

另外还发现两个小问题：

1. Patch的路径写得不对：我Patch的是package.svc.cloud.storage.Client，但实际我的模块路径是requestprocessor.services.cloud_service，完全不匹配，Mock根本没生效。
2. 测试里调用的函数名和实际定义的不一致：测试里写的是upload_file_to_gcs，但实际函数是upload_to_gcs，这会导致后续测试找不到函数（不过当前的凭证错误是先出现的）。

修复方案 #

1. 修正Patch路径（最关键） #

要Mock目标模块里实际引用storage.Client的位置。因为在cloud_service.py里我用的是from google.cloud import storage，然后调用storage.Client()，所以正确的Patch路径应该是：
"requestprocessor.services.cloud_service.storage.Client"（把这个换成你自己的真实模块路径）

2. 调整测试结构，确保Mock时机正确 #

保持先Patch再导入模块的逻辑，但把Patch放在测试函数内部的上下文管理器里，这样更清晰，也能避免模块导入的副作用。

3. 修正函数名不一致的问题 #

测试里调用的函数名要和实际定义的一致，把upload_file_to_gcs改成upload_to_gcs。

修复后的完整测试代码 #

import sys
import pytest
from unittest.mock import patch, MagicMock, AsyncMock
from fastapi import HTTPException
from google.auth.exceptions import DefaultCredentialsError

# 替换成你自己的真实模块路径
TARGET_MODULE = "requestprocessor.services.cloud_service"
# 正确的Patch路径：目标模块中的storage.Client引用
PATCH_TARGET = f"{TARGET_MODULE}.storage.Client"

@pytest.mark.asyncio
async def test_upload_to_gcs_success():
    # 上下文管理器包裹Patch，确保导入模块前Mock已生效
    with patch(PATCH_TARGET) as mock_client_class:
        # 配置Mock的各个层级
        mock_storage_client = mock_client_class.return_value
        mock_bucket = MagicMock()
        mock_blob = MagicMock()
        mock_storage_client.bucket.return_value = mock_bucket
        mock_bucket.blob.return_value = mock_blob
        mock_blob.public_url = "https://fake-url.com/file.txt"
        
        # 导入模块（此时Mock已经生效，不会触发真实的Client初始化）
        import importlib
        cloud_service = importlib.import_module(TARGET_MODULE)
        
        # 创建Mock的FastAPI文件对象
        mock_file = AsyncMock()
        mock_file.read.return_value = b"test data"
        mock_file.content_type = "text/plain"
        
        # 调用待测试的函数
        url = await cloud_service.upload_to_gcs(mock_file, "file.txt")
        
        # 断言结果
        assert url == "https://fake-url.com/file.txt"
        # 验证调用链是否正确
        mock_storage_client.bucket.assert_called_once_with("stupid-bucket-name-1234567890")
        mock_bucket.blob.assert_called_once_with("file.txt")
        mock_blob.upload_from_file.assert_called_once()

额外的可测试性优化 #

如果想彻底避免模块级别实例化带来的测试麻烦，可以把storage_client的创建封装成一个函数，用依赖注入的方式管理：

# 优化后的cloud_service.py
from io import BytesIO
from google.auth.exceptions import DefaultCredentialsError
from google.cloud import storage
from fastapi import HTTPException

bucket_name = "stupid-bucket-name-1234567890"

def get_storage_client():
    """封装Client创建，方便测试时替换"""
    return storage.Client()

async def upload_to_gcs(file, file_name: str) -> str:
    try:
        storage_client = get_storage_client()
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(file_name)
        binary_data = await file.read()
        blob.upload_from_file(BytesIO(binary_data), content_type=file.content_type)
        return blob.public_url
    except DefaultCredentialsError as exe:
        raise HTTPException(status_code=500, detail=str(exe))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

这样测试时可以直接Patchget_storage_client函数，不用再纠结模块级别的实例化问题，代码的可测试性会提升很多。

内容来源于stack exchange