Python单元测试中Mock Google Cloud Logging报403权限错误如何解决

问题根因 #

你当前的Mock失败主要有两个核心原因：

1. Patch时机晚于真实对象初始化：rss_crawler.py在模块导入阶段就直接执行了logger = get_logger()，会立即调用google.cloud.logging.Client()创建真实的GCP日志客户端，而测试用例的@patch是在测试函数运行前才执行，此时真实客户端已经初始化完成，Patch没有生效。
2. Patch目标错误：你当前Patch的是rss_crawler.logger这个已经生成的日志对象，但真实的GCP API调用是在get_logger()函数执行时初始化的Client内部触发的，即使你替换了logger对象，之前初始化的GCP日志Handler已经挂载到logging体系中，仍然会尝试上报日志到GCP。

解决方法 #

方案一：调整日志初始化逻辑为懒加载（推荐，一劳永逸） #

修改rss_crawler.py的日志初始化逻辑，避免模块导入时就创建GCP客户端，只在第一次使用日志时初始化：

from cloud_logger import get_logger
# 去掉模块级别的logger = get_logger()，改成懒加载
_logger = None
def get_module_logger():
    global _logger
    if not _logger:
        _logger = get_logger()
    return _logger
def crawl_rss_source(source_crawling_details):
    logger = get_module_logger()
    # 原有业务逻辑不变
    brand_name = source_crawling_details[constants.BRAND_NAME]
    # 剩余代码和之前一致

此时测试用例的Patch就可以正常生效，直接Patch rss_crawler.get_module_logger返回标准库的logging对象即可：

@patch("start_crawl.fetch_source_crawling_fields")
@patch("rss_crawler.get_module_logger")
def test_crawl_rss_source_raises_exception(
    self, mocked_logger_getter, mocked_source_fetcher
):
    mocked_logger_getter.return_value = logging.getLogger(__name__)
    # 原有测试逻辑不变

方案二：直接Patch GCP日志Client（无需修改业务代码） #

如果不想改动业务代码，就需要Patch cloud_logger模块中使用的google.cloud.logging.Client，并且要保证Patch在rss_crawler模块导入前生效，修改测试代码如下：

import unittest
from unittest.mock import patch, MagicMock
# 先Patch Client，再导入业务模块
@patch("cloud_logger.google.cloud.logging.Client")
class TestRssCrawler(unittest.TestCase):
    def setUp(self, mocked_gcp_client):
        # Mock Client的所有方法，避免真实调用
        mocked_client_instance = MagicMock()
        mocked_client_instance.get_default_handler.return_value = MagicMock()
        mocked_client_instance.setup_logging.return_value = None
        mocked_gcp_client.return_value = mocked_client_instance
        # 导入业务模块必须放在Patch之后
        from rss_crawler import crawl_rss_source
        self.crawl_rss_source = crawl_rss_source
    @patch("start_crawl.fetch_source_crawling_fields")
    def test_crawl_rss_source_raises_exception(
        self, mocked_source_fetcher
    ):
        # 原有测试逻辑不变
        self.test_source[constants.SCRAPE] = {
            "white_list": ["https://buffer.com/blog/(\\w|\\d|\\-)+/$"]
        }
        details = set_content_source_details(
            self.brand_name,
            self.source_name,
            self.agent_args,
            self.source,
            **self.key_word_argument,
        )
        self.assertRaises(CustomException, self.crawl_rss_source, details)
        # 剩余测试逻辑不变

额外验证方案 #

如果还是有上报问题，可以在测试启动的最开头添加如下代码，直接清空所有已挂载的GCP日志Handler：

import logging
# 移除所有GCP日志Handler，避免意外上报
for handler in logging.root.handlers[:]:
    if "google.cloud.logging" in str(handler.__class__):
        logging.root.removeHandler(handler)

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