• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Tests for acloud.internal.lib.gstorage_client."""
2
3import io
4import time
5
6import unittest
7
8from unittest import mock
9
10import apiclient
11
12from acloud import errors
13from acloud.internal.lib import driver_test_lib
14from acloud.internal.lib import gstorage_client
15
16
17# pylint: disable=protected-access, no-member
18class StorageClientTest(driver_test_lib.BaseDriverTest):
19    """Test StorageClient."""
20
21    LOCAL_SRC = "/fake/local/path"
22    BUCKET = "fake_bucket"
23    OBJECT = "fake_obj"
24    MIME_TYPE = "fake_mimetype"
25
26    def setUp(self):
27        """Set up test."""
28        super().setUp()
29        self.Patch(gstorage_client.StorageClient, "InitResourceHandle")
30        self.client = gstorage_client.StorageClient(mock.MagicMock())
31        self.client._service = mock.MagicMock()
32
33    def testGet(self):
34        """Test Get."""
35        mock_api = mock.MagicMock()
36        resource_mock = mock.MagicMock()
37        self.client._service.objects = mock.MagicMock(
38            return_value=resource_mock)
39        resource_mock.get = mock.MagicMock(return_value=mock_api)
40        self.client.Get(self.BUCKET, self.OBJECT)
41        resource_mock.get.assert_called_with(
42            bucket=self.BUCKET, object=self.OBJECT)
43        self.assertTrue(mock_api.execute.called)
44
45    def testList(self):
46        """Test List."""
47        mock_items = ["fake/return"]
48        self.Patch(
49            gstorage_client.StorageClient,
50            "ListWithMultiPages",
51            return_value=mock_items)
52        resource_mock = mock.MagicMock()
53        self.client._service.objects = mock.MagicMock(
54            return_value=resource_mock)
55        items = self.client.List(self.BUCKET, self.OBJECT)
56        self.client.ListWithMultiPages.assert_called_once_with(
57            api_resource=resource_mock.list,
58            bucket=self.BUCKET,
59            prefix=self.OBJECT)
60        self.assertEqual(mock_items, items)
61
62    def testUpload(self):
63        """Test Upload."""
64        # Create mocks
65        mock_file = mock.MagicMock()
66        mock_file_io = mock.MagicMock()
67        mock_file_io.__enter__.return_value = mock_file
68        mock_media = mock.MagicMock()
69        mock_api = mock.MagicMock()
70        mock_response = mock.MagicMock()
71
72        self.Patch(io, "FileIO", return_value=mock_file_io)
73        self.Patch(
74            apiclient.http, "MediaIoBaseUpload", return_value=mock_media)
75        resource_mock = mock.MagicMock()
76        self.client._service.objects = mock.MagicMock(
77            return_value=resource_mock)
78        resource_mock.insert = mock.MagicMock(return_value=mock_api)
79        mock_api.execute = mock.MagicMock(return_value=mock_response)
80
81        # Make the call to the api
82        response = self.client.Upload(self.LOCAL_SRC, self.BUCKET, self.OBJECT,
83                                      self.MIME_TYPE)
84
85        # Verify
86        self.assertEqual(response, mock_response)
87        io.FileIO.assert_called_with(self.LOCAL_SRC, mode="rb")
88        apiclient.http.MediaIoBaseUpload.assert_called_with(
89            mock_file, self.MIME_TYPE)
90        resource_mock.insert.assert_called_with(
91            bucket=self.BUCKET, name=self.OBJECT, media_body=mock_media)
92
93    def testUploadOSError(self):
94        """Test Upload when OSError is raised."""
95        self.Patch(io, "FileIO", side_effect=OSError("fake OSError"))
96        self.assertRaises(errors.DriverError, self.client.Upload,
97                          self.LOCAL_SRC, self.BUCKET, self.OBJECT,
98                          self.MIME_TYPE)
99
100    def testDelete(self):
101        """Test Delete."""
102        mock_api = mock.MagicMock()
103        resource_mock = mock.MagicMock()
104        self.client._service.objects = mock.MagicMock(
105            return_value=resource_mock)
106        resource_mock.delete = mock.MagicMock(return_value=mock_api)
107        self.client.Delete(self.BUCKET, self.OBJECT)
108        resource_mock.delete.assert_called_with(
109            bucket=self.BUCKET, object=self.OBJECT)
110        self.assertTrue(mock_api.execute.called)
111
112    def testDeleteMultipleFiles(self):
113        """Test Delete multiple files."""
114        fake_objs = ["fake_obj1", "fake_obj2"]
115        mock_api = mock.MagicMock()
116        resource_mock = mock.MagicMock()
117        self.client._service.objects = mock.MagicMock(
118            return_value=resource_mock)
119        resource_mock.delete = mock.MagicMock(return_value=mock_api)
120        deleted, failed, error_msgs = self.client.DeleteFiles(
121            self.BUCKET, fake_objs)
122        self.assertEqual(deleted, fake_objs)
123        self.assertEqual(failed, [])
124        self.assertEqual(error_msgs, [])
125        calls = [
126            mock.call(bucket=self.BUCKET, object="fake_obj1"),
127            mock.call(bucket=self.BUCKET, object="fake_obj2")
128        ]
129        resource_mock.delete.assert_has_calls(calls)
130        self.assertEqual(mock_api.execute.call_count, 2)
131
132    def testGetUrl(self):
133        """Test GetUrl."""
134        fake_item = {"name": "fake-item-1", "selfLink": "link1"}
135        self.Patch(
136            gstorage_client.StorageClient, "Get", return_value=fake_item)
137        self.assertEqual(
138            self.client.GetUrl("fake_bucket", "fake-item-1"), "link1")
139
140    def testGetUrlNotFound(self):
141        """Test GetUrl when object is not found."""
142        self.Patch(
143            gstorage_client.StorageClient,
144            "Get",
145            side_effect=errors.ResourceNotFoundError(404, "expected error"))
146        self.Patch(time, "sleep")
147        self.assertRaises(errors.ResourceNotFoundError, self.client.GetUrl,
148                          "fake_bucket", "fake-item-1")
149
150
151if __name__ == "__main__":
152    unittest.main()
153