python - 单元测试 Mock GCS
问题描述
我很难找到一种方法来对此类中存在的read
和方法进行单元测试。write
我正在尝试使用模拟补丁库创建一个模拟,以避免调用谷歌存储,但我很难弄清楚如何去做。
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://", uri[0], "")
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://", uri[0], "")
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%
推荐阅读
- python - 将“conda activate env_name”的输出保存到批处理文件中的变量
- php - 这些文件位于 sleay32.dll 和 libeay32.dll 的什么位置?
- javascript - Gulp 和 Webpack 流不适用于 Imports
- sql - 修剪 SQL 中的前导 0
- java - JPA中使用persist()后类似executeUpdate的方法
- aws-sdk - 无法从假设角色 aws 获得凭证
- php - 使用 cURL 获取响应更新 Laravel 数据库表
- pandas - pytorch闪电海王星上传pandas数据框错误
- powershell - 使用 PowerShell 选择特定的 Excel 列以导出到另一个电子表格
- bash - 当管道输出到文件并且需要清理时,svn commit 挂起