backup_obj = objects.Backup.get_by_id(self.context, backup.id)
self.assertEqual(temp_vol.id, backup_obj.temp_volume_id)
- @mock.patch.object(utils, 'brick_get_connector_properties')
- @mock.patch.object(cinder.volume.driver.VolumeDriver, '_attach_volume')
- @mock.patch.object(cinder.volume.driver.VolumeDriver, '_detach_volume')
- @mock.patch.object(volutils, 'copy_volume')
- @mock.patch.object(volume_rpcapi.VolumeAPI, 'get_capabilities')
- def test_copy_volume_data(self,
- mock_get_capabilities,
- mock_copy,
- mock_detach,
- mock_attach,
- mock_get_connector):
-
- src_vol = tests_utils.create_volume(self.context, size=1,
- host=CONF.host)
- dest_vol = tests_utils.create_volume(self.context, size=1,
- host=CONF.host)
- mock_get_connector.return_value = {}
- self.volume.driver._throttle = mock.MagicMock()
-
- attach_expected = [
- mock.call(self.context, dest_vol, {}, remote=False),
- mock.call(self.context, src_vol, {}, remote=False)]
-
- detach_expected = [
- mock.call(self.context, {'device': {'path': 'bar'}},
- dest_vol, {}, force=False, remote=False),
- mock.call(self.context, {'device': {'path': 'foo'}},
- src_vol, {}, force=False, remote=False)]
-
- attach_volume_returns = [
- ({'device': {'path': 'bar'}}, dest_vol),
- ({'device': {'path': 'foo'}}, src_vol),
- ]
-
- # Test case for sparse_copy_volume = False
- mock_attach.side_effect = attach_volume_returns
- mock_get_capabilities.return_value = {}
- self.volume.driver.copy_volume_data(self.context,
- src_vol,
- dest_vol)
-
- self.assertEqual(attach_expected, mock_attach.mock_calls)
- mock_copy.assert_called_with(
- 'foo', 'bar', 1024, '1M',
- throttle=self.volume.driver._throttle,
- sparse=False)
- self.assertEqual(detach_expected, mock_detach.mock_calls)
-
- # Test case for sparse_copy_volume = True
- mock_attach.reset_mock()
- mock_detach.reset_mock()
- mock_attach.side_effect = attach_volume_returns
- mock_get_capabilities.return_value = {'sparse_copy_volume': True}
- self.volume.driver.copy_volume_data(self.context,
- src_vol,
- dest_vol)
-
- self.assertEqual(attach_expected, mock_attach.mock_calls)
- mock_copy.assert_called_with(
- 'foo', 'bar', 1024, '1M',
- throttle=self.volume.driver._throttle,
- sparse=True)
- self.assertEqual(detach_expected, mock_detach.mock_calls)
-
- # cleanup resource
- db.volume_destroy(self.context, src_vol['id'])
- db.volume_destroy(self.context, dest_vol['id'])
-
+ def test__create_temp_volume_from_snapshot(self):
+ volume_dict = {'id': 'fake',
+ 'host': 'fakehost',
+ 'availability_zone': 'fakezone',
+ 'size': 1}
+ vol = fake_volume.fake_volume_obj(self.context, **volume_dict)
+ snapshot = fake_snapshot.fake_snapshot_obj(self.context)
+
+ with mock.patch.object(
+ self.volume.driver,
+ 'create_volume_from_snapshot'):
+ temp_vol = self.volume.driver._create_temp_volume_from_snapshot(
+ self.context,
+ vol, snapshot)
+ self.assertEqual('detached', temp_vol['attach_status'])
+ self.assertEqual('fakezone', temp_vol['availability_zone'])
+
@mock.patch.object(utils, 'brick_get_connector_properties')
@mock.patch.object(cinder.volume.manager.VolumeManager, '_attach_volume')
@mock.patch.object(cinder.volume.manager.VolumeManager, '_detach_volume')