"""Tests for acloud.internal.lib.gstorage_client."""
|
|
import io
|
import time
|
|
import unittest
|
import mock
|
|
import apiclient
|
|
from acloud import errors
|
from acloud.internal.lib import driver_test_lib
|
from acloud.internal.lib import gstorage_client
|
|
|
# pylint: disable=protected-access, no-member
|
class StorageClientTest(driver_test_lib.BaseDriverTest):
|
"""Test StorageClient."""
|
|
LOCAL_SRC = "/fake/local/path"
|
BUCKET = "fake_bucket"
|
OBJECT = "fake_obj"
|
MIME_TYPE = "fake_mimetype"
|
|
def setUp(self):
|
"""Set up test."""
|
super(StorageClientTest, self).setUp()
|
self.Patch(gstorage_client.StorageClient, "InitResourceHandle")
|
self.client = gstorage_client.StorageClient(mock.MagicMock())
|
self.client._service = mock.MagicMock()
|
|
def testGet(self):
|
"""Test Get."""
|
mock_api = mock.MagicMock()
|
resource_mock = mock.MagicMock()
|
self.client._service.objects = mock.MagicMock(
|
return_value=resource_mock)
|
resource_mock.get = mock.MagicMock(return_value=mock_api)
|
self.client.Get(self.BUCKET, self.OBJECT)
|
resource_mock.get.assert_called_with(
|
bucket=self.BUCKET, object=self.OBJECT)
|
self.assertTrue(mock_api.execute.called)
|
|
def testList(self):
|
"""Test List."""
|
mock_items = ["fake/return"]
|
self.Patch(
|
gstorage_client.StorageClient,
|
"ListWithMultiPages",
|
return_value=mock_items)
|
resource_mock = mock.MagicMock()
|
self.client._service.objects = mock.MagicMock(
|
return_value=resource_mock)
|
items = self.client.List(self.BUCKET, self.OBJECT)
|
self.client.ListWithMultiPages.assert_called_once_with(
|
api_resource=resource_mock.list,
|
bucket=self.BUCKET,
|
prefix=self.OBJECT)
|
self.assertEqual(mock_items, items)
|
|
def testUpload(self):
|
"""Test Upload."""
|
# Create mocks
|
mock_file = mock.MagicMock()
|
mock_file_io = mock.MagicMock()
|
mock_file_io.__enter__.return_value = mock_file
|
mock_media = mock.MagicMock()
|
mock_api = mock.MagicMock()
|
mock_response = mock.MagicMock()
|
|
self.Patch(io, "FileIO", return_value=mock_file_io)
|
self.Patch(
|
apiclient.http, "MediaIoBaseUpload", return_value=mock_media)
|
resource_mock = mock.MagicMock()
|
self.client._service.objects = mock.MagicMock(
|
return_value=resource_mock)
|
resource_mock.insert = mock.MagicMock(return_value=mock_api)
|
mock_api.execute = mock.MagicMock(return_value=mock_response)
|
|
# Make the call to the api
|
response = self.client.Upload(self.LOCAL_SRC, self.BUCKET, self.OBJECT,
|
self.MIME_TYPE)
|
|
# Verify
|
self.assertEqual(response, mock_response)
|
io.FileIO.assert_called_with(self.LOCAL_SRC, mode="rb")
|
apiclient.http.MediaIoBaseUpload.assert_called_with(
|
mock_file, self.MIME_TYPE)
|
resource_mock.insert.assert_called_with(
|
bucket=self.BUCKET, name=self.OBJECT, media_body=mock_media)
|
|
def testUploadOSError(self):
|
"""Test Upload when OSError is raised."""
|
self.Patch(io, "FileIO", side_effect=OSError("fake OSError"))
|
self.assertRaises(errors.DriverError, self.client.Upload,
|
self.LOCAL_SRC, self.BUCKET, self.OBJECT,
|
self.MIME_TYPE)
|
|
def testDelete(self):
|
"""Test Delete."""
|
mock_api = mock.MagicMock()
|
resource_mock = mock.MagicMock()
|
self.client._service.objects = mock.MagicMock(
|
return_value=resource_mock)
|
resource_mock.delete = mock.MagicMock(return_value=mock_api)
|
self.client.Delete(self.BUCKET, self.OBJECT)
|
resource_mock.delete.assert_called_with(
|
bucket=self.BUCKET, object=self.OBJECT)
|
self.assertTrue(mock_api.execute.called)
|
|
def testDeleteMultipleFiles(self):
|
"""Test Delete multiple files."""
|
fake_objs = ["fake_obj1", "fake_obj2"]
|
mock_api = mock.MagicMock()
|
resource_mock = mock.MagicMock()
|
self.client._service.objects = mock.MagicMock(
|
return_value=resource_mock)
|
resource_mock.delete = mock.MagicMock(return_value=mock_api)
|
deleted, failed, error_msgs = self.client.DeleteFiles(
|
self.BUCKET, fake_objs)
|
self.assertEqual(deleted, fake_objs)
|
self.assertEqual(failed, [])
|
self.assertEqual(error_msgs, [])
|
calls = [
|
mock.call(bucket=self.BUCKET, object="fake_obj1"),
|
mock.call(bucket=self.BUCKET, object="fake_obj2")
|
]
|
resource_mock.delete.assert_has_calls(calls)
|
self.assertEqual(mock_api.execute.call_count, 2)
|
|
def testGetUrl(self):
|
"""Test GetUrl."""
|
fake_item = {"name": "fake-item-1", "selfLink": "link1"}
|
self.Patch(
|
gstorage_client.StorageClient, "Get", return_value=fake_item)
|
self.assertEqual(
|
self.client.GetUrl("fake_bucket", "fake-item-1"), "link1")
|
|
def testGetUrlNotFound(self):
|
"""Test GetUrl when object is not found."""
|
self.Patch(
|
gstorage_client.StorageClient,
|
"Get",
|
side_effect=errors.ResourceNotFoundError(404, "expected error"))
|
self.Patch(time, "sleep")
|
self.assertRaises(errors.ResourceNotFoundError, self.client.GetUrl,
|
"fake_bucket", "fake-item-1")
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
