无法成功修补Azure ContainerClient的功能

Question

我一直在尝试修补ContainerClient的list_blobs()函数，没有能够成功地做到这一点，这段代码输出一个MagicMock()函数-但该函数没有修补，因为我希望它是(试图修补列表['Blob1'， 'Blob2']。

#################Script File
import sys
from datetime import datetime, timedelta
import pyspark
import pytz
import yaml
# from azure.storage.blob import BlobServiceClient, ContainerClient
from pyspark.dbutils import DBUtils as dbutils
import azure.storage.blob
# Open Config
def main():
spark_context = pyspark.SparkContext.getOrCreate()
spark_context.addFile(sys.argv[1])
stream = None
stream = open(sys.argv[1], "r")
config = yaml.load(stream, Loader=yaml.FullLoader)
stream.close()
account_key = dbutils.secrets.get(scope=config["Secrets"]["Scope"], key=config["Secrets"]["Key Name"])
target_container = config["Storage Configuration"]["Container"]
target_account = config["Storage Configuration"]["Account"]
days_history_to_keep = config["Storage Configuration"]["Days History To Keep"]
connection_string = (
"DefaultEndpointsProtocol=https;AccountName="
+ target_account
+ ";AccountKey="
+ account_key
+ ";EndpointSuffix=core.windows.net"
)
blob_service_client: azure.storage.blob.BlobServiceClient = (
azure.storage.blob.BlobServiceClient.from_connection_string(connection_string)
)
container_client: azure.storage.blob.ContainerClient = (
blob_service_client.get_container_client(target_container)
)
blobs = container_client.list_blobs()
print(blobs)
print(blobs)
utc = pytz.UTC
delete_before_date = utc.localize(
datetime.today() - timedelta(days=days_history_to_keep)
)
for blob in blobs:
if blob.creation_time < delete_before_date:
print("Deleting Blob: " + blob.name)
container_client.delete_blob(blob, delete_snapshots="include")

if __name__ == "__main__":
main()
#################Test File
import unittest
from unittest import mock
import DeleteOldBlobs

class DeleteBlobsTest(unittest.TestCase):
def setUp(self):
pass
@mock.patch("DeleteOldBlobs.azure.storage.blob.ContainerClient")
@mock.patch("DeleteOldBlobs.azure.storage.blob.BlobServiceClient")
@mock.patch("DeleteOldBlobs.dbutils")
@mock.patch("DeleteOldBlobs.sys")
@mock.patch('DeleteOldBlobs.pyspark')
def test_main(self, mock_pyspark, mock_sys, mock_dbutils, mock_blobserviceclient, mock_containerclient):
# mock setup
config_file = "Delete_Old_Blobs_UnitTest.yml"
mock_sys.argv = ["unused_arg", config_file]
mock_dbutils.secrets.get.return_value = "A Secret"
mock_containerclient.list_blobs.return_value = ["ablob1", "ablob2"]
# execute test
DeleteOldBlobs.main()
# TODO assert actions taken
# mock_sys.argv.__get__.assert_called_with()
# dbutils.secrets.get(scope=config['Secrets']['Scope'], key=config['Secrets']['Key Name'])

if __name__ == "__main__":
unittest.main()

输出:

<MagicMock name='BlobServiceClient.from_connection_string().get_container_client().list_blobs()' id='1143355577232'>

我在这里做错了什么?

Answer 1

我无法执行您的代码在这一刻，但我已经尝试模拟它。为此，我在路径中创建了以下3个文件:/<path-to>/pkg/sub_pkg1(其中pkg和sub_pkg1是包)。

文件ContainerClient.py

def list_blobs(self):
return "blob1"

文件DeleteOldBlobs.py

from pkg.sub_pkg1 import ContainerClient
# Open Config
def main():
blobs = ContainerClient.list_blobs()
print(blobs)
print(blobs)

文件DeleteBlobsTest.py

import unittest
from unittest import mock
from pkg.sub_pkg1 import DeleteOldBlobs
class DeleteBlobsTest(unittest.TestCase):
def setUp(self):
pass
def test_main(self):
mock_containerclient = mock.MagicMock()
with mock.patch("DeleteOldBlobs.ContainerClient.list_blobs", mock_containerclient.list_blobs):
mock_containerclient.list_blobs.return_value = ["ablob1", "ablob2"]
DeleteOldBlobs.main()
if __name__ == '__main__':
unittest.main()

如果您执行测试代码，您将获得输出:

['ablob1', 'ablob2']
['ablob1', 'ablob2']

此输出表示mock_containerclient.list_blobs模拟了list_blobs()函数。

我不知道这篇文章的内容是否对你有用，但我现在无法更好地模拟你的代码。我希望你能启发我的代码，找到你真正的解决方案。

Answer 2

答案的结构与我的解决方案不匹配，也许两者都可以工作，但对我来说修补pyspark很重要，即使我从不调用它，或者当我的代码试图与spark交互时抛出异常。

也许这对某人有用:

@mock.patch("DeleteOldBlobs.azure.storage.blob.BlobServiceClient")
@mock.patch("DeleteOldBlobs.dbutils")
@mock.patch("DeleteOldBlobs.sys")
@mock.patch('DeleteOldBlobs.pyspark')
def test_list_blobs_called_once(self, mock_pyspark, mock_sys, mock_dbutils, mock_blobserviceclient):
# mock setup
config_file = "Delete_Old_Blobs_UnitTest.yml"
mock_sys.argv = ["unused_arg", config_file]
account_key = 'Secret Key'
mock_dbutils.secrets.get.return_value = account_key
bsc_mock: mock.Mock = mock.Mock()
container_client_mock = mock.Mock()
blob1 = Blob('newblob', datetime.today())
blob2 = Blob('oldfile', datetime.today() - timedelta(days=20))
container_client_mock.list_blobs.return_value = [blob1, blob2]
bsc_mock.get_container_client.return_value = container_client_mock
mock_blobserviceclient.from_connection_string.return_value = bsc_mock
# execute test
DeleteOldBlobs.main()
#Assert Results
container_client_mock.list_blobs.assert_called_once()