如何在Pytest单元测试中Mock Google BigQuery客户端？

我正在用Python客户端库通过服务账号连接Google BigQuery，业务代码如下：

from google.cloud.bigquery import Client, LoadJobConfig, LoadJob
from google.oauth2 import service_account
import json  # 原代码漏导入json，此处补上

def fetch_last_modified_date_from_bq():
    service_account_json_string = "json to connect"
    service_account_json = json.loads(service_account_json_string)
    credentials = service_account.Credentials.from_service_account_info(service_account_json)
    client = Client(credentials=credentials)        
    # 执行表查询
    sql = """
                  SELECT last_modified_date FROM `project.dataset.table` 
                  order by last_modified_date desc LIMIT 1
          """
    query_job = client.query(sql)
    # 返回结果
    for row in query_job:
        return row.last_modified_date  # 原代码row.result应为笔误，修正为对应字段名

我尝试用以下方式Mock，但因为客户端需要传入凭证，测试没成功，请问这段测试代码是否正确？

@mock.patch('google.cloud.bigquery.Client')        
@mock.patch('google.oauth2.service_account.Credentials.from_service_account_info')
@mock.patch('google.auth.credentials.Credentials')
def test_fetch_last_modified_date_from_bq(self, mock_credentials ,mock_service_account, mock_client):
    # 准备测试数据
    mock_service_account.return_value = mock_credentials
    row={}
    row['last_modified_date'] = ''
    rows = [row]
    mock_client.return_value.query.return_value = rows
    # 执行被测试函数
    fetch_last_modified_date_from_bq()

问题分析与修正后的Mock代码 #

你的测试代码存在3个核心问题：

• Mock参数顺序错误：@mock.patch的参数传递顺序是反向的，最下方的装饰器对应第一个函数参数，你当前的参数顺序完全颠倒。
• QueryJob模拟不完整：client.query()返回的是QueryJob对象，并非直接可迭代的列表，必须MockQueryJob.result()方法才能返回行数据。
• 凭证Mock冗余：无需Mockgoogle.auth.credentials.Credentials，让服务账号凭证方法返回任意Mock对象即可，测试场景下BigQuery Client不会校验凭证有效性。

修正后的测试代码：

from unittest import mock
import json
from your_module import fetch_last_modified_date_from_bq  # 替换为你的业务代码模块名

def test_fetch_last_modified_date_from_bq():
    # 1. Mock服务账号凭证生成
    with mock.patch('google.oauth2.service_account.Credentials.from_service_account_info') as mock_sa_creds:
        mock_creds = mock.Mock()
        mock_sa_creds.return_value = mock_creds
        
        # 2. Mock BigQuery Client类及实例
        with mock.patch('google.cloud.bigquery.Client') as mock_client_cls:
            mock_client = mock.Mock()
            mock_client_cls.return_value = mock_client
            
            # 3. Mock QueryJob及查询结果
            mock_query_job = mock.Mock()
            # 构造支持属性访问的模拟行对象
            mock_row = mock.Mock()
            mock_row.last_modified_date = "2024-05-20"
            mock_query_job.result.return_value = [mock_row]
            
            # 让client.query返回Mock的QueryJob
            mock_client.query.return_value = mock_query_job
            
            # 4. 执行测试并断言结果
            result = fetch_last_modified_date_from_bq()
            assert result == "2024-05-20"
            
            # 验证调用逻辑符合预期
            mock_sa_creds.assert_called_once()
            mock_client_cls.assert_called_once_with(credentials=mock_creds)
            mock_client.query.assert_called_once()

关键注意事项 #

• Mock路径要精准：确保Mock的是业务代码中实际导入的模块路径，比如业务代码用from google.cloud.bigquery import Client，就必须Mockgoogle.cloud.bigquery.Client。
• 还原真实调用链：严格按照BigQuery SDK的调用流程模拟，query()返回QueryJob，必须通过result()获取数据。
• 避免过度Mock：只Mock必要的外部依赖，无需对底层无关类做冗余模拟。

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