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