单元测试Mock GCS

问题描述

我很难找到一种方法来对此类中的read和write方法进行单元测试。我正尝试使用模拟补丁库创建一个模拟，以避免调用Google Storage，但我很难找出方法。

from google.cloud import storage


class GCSObject(str):
    

    def __init__(self,uri=""):
        self.base,self.bucket,self.path = self.parse_uri(uri)


    def parse_uri(self,uri):
        uri = uri.lstrip("gs://").replace("//","/").split("/",1)
        if len(uri) > 1:
            return ("gs://",uri[0],uri[1])
        else:
            return ("gs://","")

    def read(self) -> bytes:
        storage_client = storage.Client()
        bucket = storage_client.bucket(self.bucket)
        return bucket.blob(self.path).download_as_string()

    def write(self,content: bytes):
        storage_client = storage.Client()
        bucket = storage_client.get_bucket(self.bucket)
        blob = bucket.blob(self.path)
        blob.upload_from_string(content)

我目前正在尝试做什么

@mock.patch("packages.pyGCP.pyGCP.gcs.storage.Client")
def test_upload(client):
    gcs_path = GCSObject("fake_path")
    reader = gcs_path.read()  
    #  confused what I should assert 

解决方法

我正在使用google-cloud-storage==1.32.0和python 3.7.5。这是单元测试解决方案：

gcs.py：

from google.cloud import storage


class GCSObject(str):

    def __init__(self,uri=""):
        self.base,self.bucket,self.path = self.parse_uri(uri)

    def parse_uri(self,uri):
        uri = uri.lstrip("gs://").replace("//","/").split("/",1)
        if len(uri) > 1:
            return ("gs://",uri[0],uri[1])
        else:
            return ("gs://","")

    def read(self) -> bytes:
        storage_client = storage.Client()
        bucket = storage_client.bucket(self.bucket)
        return bucket.blob(self.path).download_as_string()

    def write(self,content: bytes):
        storage_client = storage.Client()
        bucket = storage_client.get_bucket(self.bucket)
        blob = bucket.blob(self.path)
        blob.upload_from_string(content)

test_gcs.py：

import unittest
from unittest import mock
from unittest.mock import Mock
from gcs import GCSObject


class TestGCSObject(unittest.TestCase):
    @mock.patch('gcs.storage')
    def test_read(self,mock_storage):
        mock_gcs_client = mock_storage.Client.return_value
        mock_bucket = Mock()
        mock_bucket.blob.return_value.download_as_string.return_value = "teresa teng".encode('utf-8')
        mock_gcs_client.bucket.return_value = mock_bucket
        gcs = GCSObject('fake_path')
        actual = gcs.read()
        mock_storage.Client.assert_called_once()
        mock_gcs_client.bucket.assert_called_once_with('fake_path')
        mock_bucket.blob.assert_called_once_with('')
        mock_bucket.blob.return_value.download_as_string.assert_called_once()
        self.assertEqual(actual,"teresa teng".encode('utf-8'))

    @mock.patch('gcs.storage')
    def test_write(self,mock_storage):
        mock_gcs_client = mock_storage.Client.return_value
        mock_bucket = Mock()
        mock_gcs_client.get_bucket.return_value = mock_bucket
        gcs = GCSObject('fake_path')
        gcs.write(b'teresa teng')
        mock_storage.Client.assert_called_once()
        mock_gcs_client.get_bucket.assert_called_once_with('fake_path')
        mock_bucket.blob.assert_called_once_with('')
        mock_bucket.blob.return_value.upload_from_string.assert_called_once_with(b'teresa teng')


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

单元测试结果：

..
----------------------------------------------------------------------
Ran 2 tests in 0.004s

OK
Name                                     Stmts   Miss  Cover   Missing
----------------------------------------------------------------------
src/stackoverflow/64672497/gcs.py           18      1    94%   12
src/stackoverflow/64672497/test_gcs.py      29      0   100%
----------------------------------------------------------------------
TOTAL                                       47      1    98%