如何测试Python的google-cloud-tasks库是否使用已设置的环境变量?
我用google-cloud-tasks Python库在Dataflow工作节点创建任务。这个库默认会先查找本地凭证,找不到就调用GCP元数据服务器获取临时凭证。我已经编写代码从Secret Manager提取凭证文件存储到工作节点,还设置了GOOGLE_APPLICATION_CREDENTIALS环境变量,以此避开对元数据服务器的依赖。作为测试新手,我想知道能不能通过测试库验证这个云任务库确实使用了我设置的环境变量?
当然可以验证,这里有几个适合新手的实用测试思路:
Mock凭证加载逻辑,检查环境变量读取
用Python自带的unittest.mock或pytest-mock,mock掉google-cloud-tasks底层依赖的凭证加载函数(比如google.auth.default()),然后断言它是否优先读取了GOOGLE_APPLICATION_CREDENTIALS指向的文件。
示例代码:import os from unittest.mock import patch from google.cloud import tasks_v2 def test_uses_custom_credentials(): # 设置测试用环境变量 os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/tmp/test_creds.json" # Mock凭证加载方法 with patch("google.auth.default") as mock_auth: # 初始化云任务客户端 client = tasks_v2.CloudTasksClient() # 断言mock被调用时,未传入自定义凭证(即默认走环境变量路径) mock_auth.assert_called_once_with(scopes=None, credentials=None) # 若要更细致验证,可进一步mock`google.oauth2.service_account.Credentials.from_service_account_file`,检查是否传入了正确的文件路径直接检查客户端的凭证来源
初始化客户端后,查看其内部的凭证对象,确认它是从指定文件加载的服务账号凭证,而非元数据服务器的临时凭证:from google.cloud import tasks_v2 def test_credential_source(): client = tasks_v2.CloudTasksClient() # 获取客户端的凭证实例 creds = client._transport._credentials # 验证凭证为服务账号类型(元数据服务器返回的临时凭证无此属性) assert hasattr(creds, "_service_account_email")禁用元数据服务器,验证任务可正常创建
在测试环境中模拟元数据服务器不可用的场景(比如修改hosts屏蔽169.254.169.254),然后运行你的代码。如果能正常创建任务,说明代码确实使用了环境变量指定的凭证,而非依赖元数据服务器。这种方法更贴近真实运行场景,但需要测试环境支持网络配置修改。通过日志验证加载路径
开启Google Cloud Python客户端的调试日志,查看凭证加载过程的输出,里面会明确显示是从环境变量指定的文件加载凭证,还是调用元数据服务器。可通过设置以下环境变量开启调试日志:export GOOGLE_API_USE_CLIENT_CERTIFICATE=0 export PYTHONLOGLEVEL=DEBUG运行代码后,日志中会出现类似
Loading credentials from file...的条目,核对路径是否与你设置的一致即可。
内容的提问来源于stack exchange,提问作者Aishwary Shukla

