1# Copyright (c) 2015 Tintri.  All rights reserved.
2#
3#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4#    not use this file except in compliance with the License. You may obtain
5#    a copy of the License at
6#
7#         http://www.apache.org/licenses/LICENSE-2.0
8#
9#    Unless required by applicable law or agreed to in writing, software
10#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12#    License for the specific language governing permissions and limitations
13#    under the License.
14"""
15Volume driver test for Tintri storage.
16"""
17
18import ddt
19import mock
20
21from oslo_utils import units
22
23from cinder import context
24from cinder import exception
25from cinder import test
26from cinder.tests.unit import fake_constants as fake
27from cinder.tests.unit import fake_snapshot
28from cinder.tests.unit import fake_volume
29from cinder.tests.unit import utils as cinder_utils
30from cinder.volume.drivers.tintri import TClient
31from cinder.volume.drivers.tintri import TintriDriver
32
33
34class FakeImage(object):
35    def __init__(self):
36        self.id = 'image-id'
37        self.name = 'image-name'
38        self.properties = {'provider_location': 'nfs://share'}
39
40    def __getitem__(self, key):
41        return self.__dict__[key]
42
43
44@ddt.ddt
45class TintriDriverTestCase(test.TestCase):
46    def setUp(self):
47        super(TintriDriverTestCase, self).setUp()
48        self.context = context.get_admin_context()
49        kwargs = {'configuration': self.create_configuration()}
50        self._driver = TintriDriver(**kwargs)
51        self._driver._hostname = 'host'
52        self._driver._username = 'user'
53        self._driver._password = 'password'
54        self._driver._api_version = 'v310'
55        self._driver._image_cache_expiry = 30
56        self._provider_location = 'localhost:/share'
57        self._driver._mounted_shares = [self._provider_location]
58        self.fake_stubs()
59
60    def create_configuration(self):
61        configuration = mock.Mock()
62        configuration.nfs_mount_point_base = '/mnt/test'
63        configuration.nfs_mount_options = None
64        configuration.nas_mount_options = None
65        return configuration
66
67    def fake_stubs(self):
68        self.mock_object(TClient, 'login', self.fake_login)
69        self.mock_object(TClient, 'logout', self.fake_logout)
70        self.mock_object(TClient, 'get_snapshot', self.fake_get_snapshot)
71        self.mock_object(TClient, 'get_image_snapshots_to_date',
72                         self.fake_get_image_snapshots_to_date)
73        self.mock_object(TintriDriver, '_move_cloned_volume',
74                         self.fake_move_cloned_volume)
75        self.mock_object(TintriDriver, '_get_provider_location',
76                         self.fake_get_provider_location)
77        self.mock_object(TintriDriver, '_set_rw_permissions',
78                         self.fake_set_rw_permissions)
79        self.mock_object(TintriDriver, '_is_volume_present',
80                         self.fake_is_volume_present)
81        self.mock_object(TintriDriver, '_is_share_vol_compatible',
82                         self.fake_is_share_vol_compatible)
83        self.mock_object(TintriDriver, '_is_file_size_equal',
84                         self.fake_is_file_size_equal)
85
86    def fake_login(self, user_name, password):
87        return 'session-id'
88
89    def fake_logout(self):
90        pass
91
92    def fake_get_snapshot(self, volume_id):
93        return fake.SNAPSHOT_ID
94
95    def fake_get_image_snapshots_to_date(self, date):
96        return [{'uuid': {'uuid': 'image_snapshot-id'}}]
97
98    def fake_move_cloned_volume(self, clone_name, volume_id, share=None):
99        pass
100
101    def fake_get_provider_location(self, volume_path):
102        return self._provider_location
103
104    def fake_set_rw_permissions(self, path):
105        pass
106
107    def fake_is_volume_present(self, volume_path):
108        return True
109
110    def fake_is_share_vol_compatible(self, volume, share):
111        return True
112
113    def fake_is_file_size_equal(self, path, size):
114        return True
115
116    @mock.patch.object(TClient, 'create_snapshot',
117                       mock.Mock(return_value=fake.PROVIDER_ID))
118    def test_create_snapshot(self):
119        snapshot = fake_snapshot.fake_snapshot_obj(self.context)
120        volume = fake_volume.fake_volume_obj(self.context)
121        provider_id = fake.PROVIDER_ID
122        snapshot.volume = volume
123        with mock.patch('cinder.objects.snapshot.Snapshot.save'):
124            self.assertEqual({'provider_id': fake.PROVIDER_ID},
125                             self._driver.create_snapshot(snapshot))
126            self.assertEqual(provider_id, snapshot.provider_id)
127
128    @mock.patch.object(TClient, 'create_snapshot', mock.Mock(
129                       side_effect=exception.VolumeDriverException))
130    def test_create_snapshot_failure(self):
131        snapshot = fake_snapshot.fake_snapshot_obj(self.context)
132        volume = fake_volume.fake_volume_obj(self.context)
133        snapshot.volume = volume
134        self.assertRaises(exception.VolumeDriverException,
135                          self._driver.create_snapshot, snapshot)
136
137    @mock.patch.object(TClient, 'delete_snapshot', mock.Mock())
138    @mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall', new=
139                cinder_utils.ZeroIntervalLoopingCall)
140    def test_cleanup_cache(self):
141        self.assertFalse(self._driver.cache_cleanup)
142        timer = self._driver._initiate_image_cache_cleanup()
143        # wait for cache cleanup to complete
144        timer.wait()
145        self.assertFalse(self._driver.cache_cleanup)
146
147    @mock.patch.object(TClient, 'delete_snapshot', mock.Mock(
148                       side_effect=exception.VolumeDriverException))
149    @mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall', new=
150                cinder_utils.ZeroIntervalLoopingCall)
151    def test_cleanup_cache_delete_fail(self):
152        self.assertFalse(self._driver.cache_cleanup)
153        timer = self._driver._initiate_image_cache_cleanup()
154        # wait for cache cleanup to complete
155        timer.wait()
156        self.assertFalse(self._driver.cache_cleanup)
157
158    @mock.patch.object(TClient, 'delete_snapshot', mock.Mock())
159    def test_delete_snapshot(self):
160        snapshot = fake_snapshot.fake_snapshot_obj(self.context)
161        snapshot.provider_id = fake.PROVIDER_ID
162        self.assertIsNone(self._driver.delete_snapshot(snapshot))
163
164    @mock.patch.object(TClient, 'delete_snapshot', mock.Mock(
165                       side_effect=exception.VolumeDriverException))
166    def test_delete_snapshot_failure(self):
167        snapshot = fake_snapshot.fake_snapshot_obj(self.context)
168        snapshot.provider_id = fake.PROVIDER_ID
169        self.assertRaises(exception.VolumeDriverException,
170                          self._driver.delete_snapshot, snapshot)
171
172    @mock.patch.object(TClient, 'clone_volume', mock.Mock())
173    def test_create_volume_from_snapshot(self):
174        snapshot = fake_snapshot.fake_snapshot_obj(self.context)
175        volume = fake_volume.fake_volume_obj(self.context)
176        self.assertEqual({'provider_location': self._provider_location},
177                         self._driver.create_volume_from_snapshot(
178                         volume, snapshot))
179
180    @mock.patch.object(TClient, 'clone_volume', mock.Mock(
181                       side_effect=exception.VolumeDriverException))
182    def test_create_volume_from_snapshot_failure(self):
183        snapshot = fake_snapshot.fake_snapshot_obj(self.context)
184        volume = fake_volume.fake_volume_obj(self.context)
185        self.assertRaises(exception.VolumeDriverException,
186                          self._driver.create_volume_from_snapshot,
187                          volume, snapshot)
188
189    @mock.patch.object(TClient, 'clone_volume', mock.Mock())
190    @mock.patch.object(TClient, 'create_snapshot', mock.Mock())
191    def test_create_cloned_volume(self):
192        volume = fake_volume.fake_volume_obj(self.context)
193        self.assertEqual({'provider_location': self._provider_location},
194                         self._driver.create_cloned_volume(volume, volume))
195
196    @mock.patch.object(TClient, 'clone_volume', mock.Mock(
197                       side_effect=exception.VolumeDriverException))
198    @mock.patch.object(TClient, 'create_snapshot', mock.Mock())
199    def test_create_cloned_volume_failure(self):
200        volume = fake_volume.fake_volume_obj(self.context)
201        self.assertRaises(exception.VolumeDriverException,
202                          self._driver.create_cloned_volume, volume, volume)
203
204    @mock.patch.object(TClient, 'clone_volume', mock.Mock())
205    def test_clone_image(self):
206        volume = fake_volume.fake_volume_obj(self.context)
207        self.assertEqual(({'provider_location': self._provider_location,
208                           'bootable': True}, True),
209                         self._driver.clone_image(
210                         None, volume, 'image-name', FakeImage().__dict__,
211                         None))
212
213    @mock.patch.object(TClient, 'clone_volume', mock.Mock(
214                       side_effect=exception.VolumeDriverException))
215    def test_clone_image_failure(self):
216        volume = fake_volume.fake_volume_obj(self.context)
217        self.assertEqual(({'provider_location': None,
218                           'bootable': False}, False),
219                         self._driver.clone_image(
220                         None, volume, 'image-name', FakeImage().__dict__,
221                         None))
222
223    def test_manage_existing(self):
224        volume = fake_volume.fake_volume_obj(self.context)
225        existing = {'source-name': self._provider_location + '/' +
226                    volume.name}
227        with mock.patch('os.path.isfile', return_value=True):
228            self.assertEqual({'provider_location': self._provider_location},
229                             self._driver.manage_existing(volume, existing))
230
231    def test_manage_existing_invalid_ref(self):
232        existing = fake_volume.fake_volume_obj(self.context)
233        volume = fake_volume.fake_volume_obj(self.context)
234        self.assertRaises(exception.ManageExistingInvalidReference,
235                          self._driver.manage_existing, volume, existing)
236
237    def test_manage_existing_not_found(self):
238        volume = fake_volume.fake_volume_obj(self.context)
239        existing = {'source-name': self._provider_location + '/' +
240                    volume.name}
241        with mock.patch('os.path.isfile', return_value=False):
242            self.assertRaises(exception.ManageExistingInvalidReference,
243                              self._driver.manage_existing, volume, existing)
244
245    @mock.patch.object(TintriDriver, '_move_file', mock.Mock(
246        return_value=False))
247    def test_manage_existing_move_failure(self):
248        volume = fake_volume.fake_volume_obj(self.context)
249        existing = {'source-name': self._provider_location + '/source-volume'}
250        with mock.patch('os.path.isfile', return_value=True):
251            self.assertRaises(exception.VolumeDriverException,
252                              self._driver.manage_existing,
253                              volume, existing)
254
255    @ddt.data((123, 123), (123.5, 124))
256    @ddt.unpack
257    def test_manage_existing_get_size(self, st_size, exp_size):
258        volume = fake_volume.fake_volume_obj(self.context)
259        existing = {'source-name': self._provider_location + '/' +
260                    volume.name}
261        file = mock.Mock(st_size=int(st_size * units.Gi))
262        with mock.patch('os.path.isfile', return_value=True):
263            with mock.patch('os.stat', return_value=file):
264                self.assertEqual(exp_size,
265                                 self._driver.manage_existing_get_size(
266                                     volume, existing))
267
268    def test_manage_existing_get_size_failure(self):
269        volume = fake_volume.fake_volume_obj(self.context)
270        existing = {'source-name': self._provider_location + '/' +
271                    volume.name}
272        with mock.patch('os.path.isfile', return_value=True):
273            with mock.patch('os.stat', side_effect=OSError):
274                self.assertRaises(exception.VolumeDriverException,
275                                  self._driver.manage_existing_get_size,
276                                  volume, existing)
277
278    def test_unmanage(self):
279        volume = fake_volume.fake_volume_obj(self.context)
280        volume.provider_location = self._provider_location
281        self._driver.unmanage(volume)
282
283    def test_retype(self):
284        volume = fake_volume.fake_volume_obj(self.context)
285        retype, update = self._driver.retype(None, volume, None, None, None)
286        self.assertTrue(retype)
287        self.assertIsNone(update)
288