我试图为下面的函数编写单元测试(从snowflake获取一些数据),但我的单元测试失败了,我无法找出原因。
我的函数
import snowflake.connector
from service.kms_utils import decrypt
def get_data_from_snowflake(server_name):
sql_query = f"Select * from mytable where servername={server_name}"
server_details = []
try:
conn = snowflake.connector.connect(
user=decrypt(os.environ.get('sfusername')),
password=decrypt(os.environ.get('sfpwd')),
account=decrypt(os.environ.get('account')),
warehouse="wh1"
)
cs = conn.cursor().execute(sql_query)
server_details= cs.fetchall()
conn.close()
except Exception as e:
if "Could not connect to Snowflake Query backend after" in str(e):
logger.error(f"Snowflake: Snowflake query timeout. {e}")
elif "Failed to connect to DB" in str(e):
logger.error(f"Snowflake: Snowflake DB connection error. {e}")
else:
logger.error(f"Snowflake: Failed to get CI validationm data. {e}")
return server_details
我的单元测试如下
import pytest
from unittest.mock import patch, MagicMock
from my_module import get_data_from_snowflake
@patch('snowflake.connector.connect')
def test_get_data_from_snowflake_success(mock_connect):
#Mock the Snowflake connection object and cursor
mock_cursor = MagicMock()
mock_connect.return_value.cursor.return_value.execute.return_value = mock_cursor
mock_cursor.fetchall.return_value = [(1, 'Server1', 'Details1'), (2, 'Server2', 'Details2')]
# Call the function being tested
result = get_data_from_snowflake('Server1')
# Check that the result is as expected
assert result == [(1, 'Server1', 'Details1')]
但是Assert失败,因为结果是一个空列表[]。解密字符串时解密函数出错。必须指定一个区域。
1条答案
按热度按时间mwg9r5ms1#
这个问题源于你的
decrypt
调用Error in decrypt function when decrypting string. You must specify a region
。我在模块导入语句中看到它引用了KMS
,这是一个AWS服务。请求区域也是我以前见过很多次的AWS
回溯。你需要确保模拟decrypt
实际上到达AWS
的功能。如果您对模拟此功能不感兴趣,则可以在测试中设置以下内容:
我的建议是不要让它击中
AWS
,但如果没有关于decrypt
的更多信息,更详细的答案是不可能的。