"Fossies" - the Fresh Open Source Software Archive

Member "cinder-13.0.7/cinder/tests/unit/test_volume_utils.py" (4 Oct 2019, 50744 Bytes) of package /linux/misc/openstack/cinder-13.0.7.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_volume_utils.py": 14.0.2_vs_15.0.0.

    1 # Copyright 2011 OpenStack Foundation
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 """Tests For miscellaneous util methods used with volume."""
   17 
   18 
   19 import datetime
   20 import io
   21 import mock
   22 import six
   23 
   24 from castellan import key_manager
   25 import ddt
   26 from oslo_concurrency import processutils
   27 from oslo_config import cfg
   28 from oslo_utils import units
   29 
   30 from cinder import context
   31 from cinder import db
   32 from cinder.db.sqlalchemy import models
   33 from cinder import exception
   34 from cinder.objects import fields
   35 from cinder import test
   36 from cinder.tests.unit.backup import fake_backup
   37 from cinder.tests.unit import fake_constants as fake
   38 from cinder.tests.unit import fake_group
   39 from cinder.tests.unit import fake_snapshot
   40 from cinder.tests.unit import fake_volume
   41 from cinder import utils
   42 from cinder.volume import throttling
   43 from cinder.volume import utils as volume_utils
   44 from cinder.volume import volume_types
   45 
   46 
   47 CONF = cfg.CONF
   48 
   49 
   50 class NotifyUsageTestCase(test.TestCase):
   51     @mock.patch('cinder.volume.utils._usage_from_volume')
   52     @mock.patch('cinder.volume.utils.CONF')
   53     @mock.patch('cinder.volume.utils.rpc')
   54     def test_notify_about_volume_usage(self, mock_rpc, mock_conf, mock_usage):
   55         mock_conf.host = 'host1'
   56         output = volume_utils.notify_about_volume_usage(mock.sentinel.context,
   57                                                         mock.sentinel.volume,
   58                                                         'test_suffix')
   59         self.assertIsNone(output)
   60         mock_usage.assert_called_once_with(mock.sentinel.context,
   61                                            mock.sentinel.volume)
   62         mock_rpc.get_notifier.assert_called_once_with('volume', 'host1')
   63         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
   64             mock.sentinel.context,
   65             'volume.test_suffix',
   66             mock_usage.return_value)
   67 
   68     @mock.patch('cinder.volume.utils._usage_from_volume')
   69     @mock.patch('cinder.volume.utils.CONF')
   70     @mock.patch('cinder.volume.utils.rpc')
   71     def test_notify_about_volume_usage_with_kwargs(self, mock_rpc, mock_conf,
   72                                                    mock_usage):
   73         mock_conf.host = 'host1'
   74         output = volume_utils.notify_about_volume_usage(
   75             mock.sentinel.context,
   76             mock.sentinel.volume,
   77             'test_suffix',
   78             extra_usage_info={'a': 'b', 'c': 'd'},
   79             host='host2')
   80         self.assertIsNone(output)
   81         mock_usage.assert_called_once_with(mock.sentinel.context,
   82                                            mock.sentinel.volume, a='b', c='d')
   83         mock_rpc.get_notifier.assert_called_once_with('volume', 'host2')
   84         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
   85             mock.sentinel.context,
   86             'volume.test_suffix',
   87             mock_usage.return_value)
   88 
   89     @mock.patch('cinder.volume.utils._usage_from_snapshot')
   90     @mock.patch('cinder.volume.utils.CONF')
   91     @mock.patch('cinder.volume.utils.rpc')
   92     def test_notify_about_snapshot_usage(self, mock_rpc,
   93                                          mock_conf, mock_usage):
   94         mock_conf.host = 'host1'
   95         output = volume_utils.notify_about_snapshot_usage(
   96             mock.sentinel.context,
   97             mock.sentinel.snapshot,
   98             'test_suffix')
   99         self.assertIsNone(output)
  100         mock_usage.assert_called_once_with(mock.sentinel.snapshot,
  101                                            mock.sentinel.context)
  102         mock_rpc.get_notifier.assert_called_once_with('snapshot', 'host1')
  103         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  104             mock.sentinel.context,
  105             'snapshot.test_suffix',
  106             mock_usage.return_value)
  107 
  108     @mock.patch('cinder.volume.utils._usage_from_snapshot')
  109     @mock.patch('cinder.volume.utils.CONF')
  110     @mock.patch('cinder.volume.utils.rpc')
  111     def test_notify_about_snapshot_usage_with_kwargs(self, mock_rpc, mock_conf,
  112                                                      mock_usage):
  113         mock_conf.host = 'host1'
  114         output = volume_utils.notify_about_snapshot_usage(
  115             mock.sentinel.context,
  116             mock.sentinel.snapshot,
  117             'test_suffix',
  118             extra_usage_info={'a': 'b', 'c': 'd'},
  119             host='host2')
  120         self.assertIsNone(output)
  121         mock_usage.assert_called_once_with(mock.sentinel.snapshot,
  122                                            mock.sentinel.context,
  123                                            a='b', c='d')
  124         mock_rpc.get_notifier.assert_called_once_with('snapshot', 'host2')
  125         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  126             mock.sentinel.context,
  127             'snapshot.test_suffix',
  128             mock_usage.return_value)
  129 
  130     @mock.patch('cinder.db.volume_get')
  131     def test_usage_from_snapshot(self, volume_get):
  132         raw_volume = {
  133             'id': fake.VOLUME_ID,
  134             'availability_zone': 'nova'
  135         }
  136         ctxt = context.get_admin_context()
  137         volume_obj = fake_volume.fake_volume_obj(ctxt, **raw_volume)
  138         volume_get.return_value = volume_obj
  139         raw_snapshot = {
  140             'project_id': fake.PROJECT_ID,
  141             'user_id': fake.USER_ID,
  142             'volume': volume_obj,
  143             'volume_id': fake.VOLUME_ID,
  144             'volume_size': 1,
  145             'id': fake.SNAPSHOT_ID,
  146             'display_name': '11',
  147             'created_at': '2014-12-11T10:10:00',
  148             'status': fields.SnapshotStatus.ERROR,
  149             'deleted': '',
  150             'snapshot_metadata': [{'key': 'fake_snap_meta_key',
  151                                    'value': 'fake_snap_meta_value'}],
  152             'expected_attrs': ['metadata'],
  153         }
  154 
  155         snapshot_obj = fake_snapshot.fake_snapshot_obj(ctxt, **raw_snapshot)
  156         usage_info = volume_utils._usage_from_snapshot(snapshot_obj, ctxt)
  157         expected_snapshot = {
  158             'tenant_id': fake.PROJECT_ID,
  159             'user_id': fake.USER_ID,
  160             'availability_zone': 'nova',
  161             'volume_id': fake.VOLUME_ID,
  162             'volume_size': 1,
  163             'snapshot_id': fake.SNAPSHOT_ID,
  164             'display_name': '11',
  165             'created_at': '2014-12-11T10:10:00+00:00',
  166             'status': fields.SnapshotStatus.ERROR,
  167             'deleted': '',
  168             'metadata': six.text_type({'fake_snap_meta_key':
  169                                       u'fake_snap_meta_value'}),
  170         }
  171         self.assertDictEqual(expected_snapshot, usage_info)
  172 
  173     @mock.patch('cinder.db.volume_get')
  174     def test_usage_from_deleted_snapshot(self, volume_get):
  175         raw_volume = {
  176             'id': fake.VOLUME_ID,
  177             'availability_zone': 'nova',
  178             'deleted': 1
  179         }
  180         ctxt = context.get_admin_context()
  181         volume_obj = fake_volume.fake_volume_obj(ctxt, **raw_volume)
  182         volume_get.return_value = volume_obj
  183 
  184         raw_snapshot = {
  185             'project_id': fake.PROJECT_ID,
  186             'user_id': fake.USER_ID,
  187             'volume': volume_obj,
  188             'volume_id': fake.VOLUME_ID,
  189             'volume_size': 1,
  190             'id': fake.SNAPSHOT_ID,
  191             'display_name': '11',
  192             'created_at': '2014-12-11T10:10:00',
  193             'status': fields.SnapshotStatus.ERROR,
  194             'deleted': '',
  195             'snapshot_metadata': [{'key': 'fake_snap_meta_key',
  196                                    'value': 'fake_snap_meta_value'}],
  197             'expected_attrs': ['metadata'],
  198         }
  199 
  200         snapshot_obj = fake_snapshot.fake_snapshot_obj(ctxt, **raw_snapshot)
  201         usage_info = volume_utils._usage_from_snapshot(snapshot_obj, ctxt)
  202         expected_snapshot = {
  203             'tenant_id': fake.PROJECT_ID,
  204             'user_id': fake.USER_ID,
  205             'availability_zone': 'nova',
  206             'volume_id': fake.VOLUME_ID,
  207             'volume_size': 1,
  208             'snapshot_id': fake.SNAPSHOT_ID,
  209             'display_name': '11',
  210             'created_at': mock.ANY,
  211             'status': fields.SnapshotStatus.ERROR,
  212             'deleted': '',
  213             'metadata': six.text_type({'fake_snap_meta_key':
  214                                       u'fake_snap_meta_value'}),
  215         }
  216         self.assertDictEqual(expected_snapshot, usage_info)
  217 
  218     @mock.patch('cinder.db.volume_glance_metadata_get')
  219     @mock.patch('cinder.db.volume_attachment_get_all_by_volume_id')
  220     def test_usage_from_volume(self, mock_attachment, mock_image_metadata):
  221         mock_image_metadata.return_value = {'image_id': 'fake_image_id'}
  222         mock_attachment.return_value = [{'instance_uuid': 'fake_instance_id'}]
  223         raw_volume = {
  224             'project_id': '12b0330ec2584a',
  225             'user_id': '158cba1b8c2bb6008e',
  226             'host': 'fake_host',
  227             'availability_zone': 'nova',
  228             'volume_type_id': 'fake_volume_type_id',
  229             'id': 'fake_volume_id',
  230             'size': 1,
  231             'display_name': 'test_volume',
  232             'created_at': datetime.datetime(2015, 1, 1, 1, 1, 1),
  233             'launched_at': datetime.datetime(2015, 1, 1, 1, 1, 1),
  234             'snapshot_id': None,
  235             'replication_status': None,
  236             'replication_extended_status': None,
  237             'replication_driver_data': None,
  238             'status': 'available',
  239             'volume_metadata': {'fake_metadata_key': 'fake_metadata_value'},
  240         }
  241         usage_info = volume_utils._usage_from_volume(
  242             mock.sentinel.context,
  243             raw_volume)
  244         expected_volume = {
  245             'tenant_id': '12b0330ec2584a',
  246             'user_id': '158cba1b8c2bb6008e',
  247             'host': 'fake_host',
  248             'availability_zone': 'nova',
  249             'volume_type': 'fake_volume_type_id',
  250             'volume_id': 'fake_volume_id',
  251             'size': 1,
  252             'display_name': 'test_volume',
  253             'created_at': '2015-01-01T01:01:01',
  254             'launched_at': '2015-01-01T01:01:01',
  255             'snapshot_id': None,
  256             'replication_status': None,
  257             'replication_extended_status': None,
  258             'replication_driver_data': None,
  259             'status': 'available',
  260             'metadata': {'fake_metadata_key': 'fake_metadata_value'},
  261             'glance_metadata': {'image_id': 'fake_image_id'},
  262             'volume_attachment': [{'instance_uuid': 'fake_instance_id'}],
  263         }
  264         self.assertEqual(expected_volume, usage_info)
  265 
  266     @mock.patch('cinder.volume.utils._usage_from_consistencygroup')
  267     @mock.patch('cinder.volume.utils.CONF')
  268     @mock.patch('cinder.volume.utils.rpc')
  269     def test_notify_about_consistencygroup_usage(self, mock_rpc,
  270                                                  mock_conf, mock_usage):
  271         mock_conf.host = 'host1'
  272         output = volume_utils.notify_about_consistencygroup_usage(
  273             mock.sentinel.context,
  274             mock.sentinel.consistencygroup,
  275             'test_suffix')
  276         self.assertIsNone(output)
  277         mock_usage.assert_called_once_with(mock.sentinel.consistencygroup)
  278         mock_rpc.get_notifier.assert_called_once_with('consistencygroup',
  279                                                       'host1')
  280         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  281             mock.sentinel.context,
  282             'consistencygroup.test_suffix',
  283             mock_usage.return_value)
  284 
  285     @mock.patch('cinder.volume.utils._usage_from_consistencygroup')
  286     @mock.patch('cinder.volume.utils.CONF')
  287     @mock.patch('cinder.volume.utils.rpc')
  288     def test_notify_about_consistencygroup_usage_with_kwargs(self, mock_rpc,
  289                                                              mock_conf,
  290                                                              mock_usage):
  291         mock_conf.host = 'host1'
  292         output = volume_utils.notify_about_consistencygroup_usage(
  293             mock.sentinel.context,
  294             mock.sentinel.consistencygroup,
  295             'test_suffix',
  296             extra_usage_info={'a': 'b', 'c': 'd'},
  297             host='host2')
  298         self.assertIsNone(output)
  299         mock_usage.assert_called_once_with(mock.sentinel.consistencygroup,
  300                                            a='b', c='d')
  301         mock_rpc.get_notifier.assert_called_once_with('consistencygroup',
  302                                                       'host2')
  303         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  304             mock.sentinel.context,
  305             'consistencygroup.test_suffix',
  306             mock_usage.return_value)
  307 
  308     @mock.patch('cinder.volume.utils._usage_from_cgsnapshot')
  309     @mock.patch('cinder.volume.utils.CONF')
  310     @mock.patch('cinder.volume.utils.rpc')
  311     def test_notify_about_cgsnapshot_usage(self, mock_rpc,
  312                                            mock_conf, mock_usage):
  313         mock_conf.host = 'host1'
  314         output = volume_utils.notify_about_cgsnapshot_usage(
  315             mock.sentinel.context,
  316             mock.sentinel.cgsnapshot,
  317             'test_suffix')
  318         self.assertIsNone(output)
  319         mock_usage.assert_called_once_with(mock.sentinel.cgsnapshot)
  320         mock_rpc.get_notifier.assert_called_once_with('cgsnapshot', 'host1')
  321         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  322             mock.sentinel.context,
  323             'cgsnapshot.test_suffix',
  324             mock_usage.return_value)
  325 
  326     @mock.patch('cinder.volume.utils._usage_from_cgsnapshot')
  327     @mock.patch('cinder.volume.utils.CONF')
  328     @mock.patch('cinder.volume.utils.rpc')
  329     def test_notify_about_cgsnapshot_usage_with_kwargs(self, mock_rpc,
  330                                                        mock_conf, mock_usage):
  331         mock_conf.host = 'host1'
  332         output = volume_utils.notify_about_cgsnapshot_usage(
  333             mock.sentinel.context,
  334             mock.sentinel.cgsnapshot,
  335             'test_suffix',
  336             extra_usage_info={'a': 'b', 'c': 'd'},
  337             host='host2')
  338         self.assertIsNone(output)
  339         mock_usage.assert_called_once_with(mock.sentinel.cgsnapshot,
  340                                            a='b', c='d')
  341         mock_rpc.get_notifier.assert_called_once_with('cgsnapshot', 'host2')
  342         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  343             mock.sentinel.context,
  344             'cgsnapshot.test_suffix',
  345             mock_usage.return_value)
  346 
  347     def test_usage_from_backup(self):
  348         raw_backup = {
  349             'project_id': fake.PROJECT_ID,
  350             'user_id': fake.USER_ID,
  351             'availability_zone': 'nova',
  352             'id': fake.BACKUP_ID,
  353             'host': 'fake_host',
  354             'display_name': 'test_backup',
  355             'created_at': datetime.datetime(2015, 1, 1, 1, 1, 1),
  356             'status': 'available',
  357             'volume_id': fake.VOLUME_ID,
  358             'size': 1,
  359             'service_metadata': None,
  360             'service': 'cinder.backup.drivers.swift',
  361             'fail_reason': None,
  362             'parent_id': fake.BACKUP2_ID,
  363             'num_dependent_backups': 0,
  364             'snapshot_id': None,
  365         }
  366 
  367         ctxt = context.get_admin_context()
  368         backup_obj = fake_backup.fake_backup_obj(ctxt, **raw_backup)
  369 
  370         # Make it easier to find out differences between raw and expected.
  371         expected_backup = raw_backup.copy()
  372         expected_backup['tenant_id'] = expected_backup.pop('project_id')
  373         expected_backup['backup_id'] = expected_backup.pop('id')
  374         expected_backup['created_at'] = '2015-01-01T01:01:01+00:00'
  375 
  376         usage_info = volume_utils._usage_from_backup(backup_obj)
  377         self.assertDictEqual(expected_backup, usage_info)
  378 
  379 
  380 class LVMVolumeDriverTestCase(test.TestCase):
  381     def test_convert_blocksize_option(self):
  382         # Test valid volume_dd_blocksize
  383         bs = volume_utils._check_blocksize('10M')
  384         self.assertEqual('10M', bs)
  385 
  386         bs = volume_utils._check_blocksize('1xBBB')
  387         self.assertEqual('1M', bs)
  388 
  389         # Test 'volume_dd_blocksize' with fraction
  390         bs = volume_utils._check_blocksize('1.3M')
  391         self.assertEqual('1M', bs)
  392 
  393         # Test zero-size 'volume_dd_blocksize'
  394         bs = volume_utils._check_blocksize('0M')
  395         self.assertEqual('1M', bs)
  396 
  397         # Test negative 'volume_dd_blocksize'
  398         bs = volume_utils._check_blocksize('-1M')
  399         self.assertEqual('1M', bs)
  400 
  401         # Test non-digital 'volume_dd_blocksize'
  402         bs = volume_utils._check_blocksize('ABM')
  403         self.assertEqual('1M', bs)
  404 
  405     @mock.patch('cinder.volume.utils._usage_from_capacity')
  406     @mock.patch('cinder.volume.utils.CONF')
  407     @mock.patch('cinder.volume.utils.rpc')
  408     def test_notify_about_capacity_usage(self, mock_rpc,
  409                                          mock_conf, mock_usage):
  410         mock_conf.host = 'host1'
  411         output = volume_utils.notify_about_capacity_usage(
  412             mock.sentinel.context,
  413             mock.sentinel.capacity,
  414             'test_suffix')
  415         self.assertIsNone(output)
  416         mock_usage.assert_called_once_with(mock.sentinel.capacity)
  417         mock_rpc.get_notifier.assert_called_once_with('capacity', 'host1')
  418         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  419             mock.sentinel.context,
  420             'capacity.test_suffix',
  421             mock_usage.return_value)
  422 
  423     @mock.patch('cinder.volume.utils._usage_from_capacity')
  424     @mock.patch('cinder.volume.utils.CONF')
  425     @mock.patch('cinder.volume.utils.rpc')
  426     def test_notify_about_capacity_usage_with_kwargs(self, mock_rpc, mock_conf,
  427                                                      mock_usage):
  428         mock_conf.host = 'host1'
  429         output = volume_utils.notify_about_capacity_usage(
  430             mock.sentinel.context,
  431             mock.sentinel.capacity,
  432             'test_suffix',
  433             extra_usage_info={'a': 'b', 'c': 'd'},
  434             host='host2')
  435         self.assertIsNone(output)
  436         mock_usage.assert_called_once_with(mock.sentinel.capacity,
  437                                            a='b', c='d')
  438         mock_rpc.get_notifier.assert_called_once_with('capacity', 'host2')
  439         mock_rpc.get_notifier.return_value.info.assert_called_once_with(
  440             mock.sentinel.context,
  441             'capacity.test_suffix',
  442             mock_usage.return_value)
  443 
  444     def test_usage_from_capacity(self):
  445         test_capacity = {
  446             'name_to_id': 'host1@backend1#pool1',
  447             'type': 'pool',
  448             'total': '10.01',
  449             'free': '8.01',
  450             'allocated': '2',
  451             'provisioned': '2',
  452             'virtual_free': '8.01',
  453             'reported_at': '2014-12-11T10:10:00',
  454         }
  455 
  456         usage_info = volume_utils._usage_from_capacity(
  457             test_capacity)
  458         expected_capacity = {
  459             'name_to_id': 'host1@backend1#pool1',
  460             'total': '10.01',
  461             'free': '8.01',
  462             'allocated': '2',
  463             'provisioned': '2',
  464             'virtual_free': '8.01',
  465             'reported_at': '2014-12-11T10:10:00',
  466         }
  467         self.assertEqual(expected_capacity, usage_info)
  468 
  469 
  470 class OdirectSupportTestCase(test.TestCase):
  471     @mock.patch('cinder.utils.execute')
  472     def test_check_for_odirect_support(self, mock_exec):
  473         output = volume_utils.check_for_odirect_support('/dev/abc', '/dev/def')
  474         self.assertTrue(output)
  475         mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/abc',
  476                                           'of=/dev/def', 'oflag=direct',
  477                                           run_as_root=True)
  478         mock_exec.reset_mock()
  479 
  480         output = volume_utils.check_for_odirect_support('/dev/abc', '/dev/def',
  481                                                         'iflag=direct')
  482         self.assertTrue(output)
  483         mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/abc',
  484                                           'of=/dev/def', 'iflag=direct',
  485                                           run_as_root=True)
  486         mock_exec.reset_mock()
  487 
  488         output = volume_utils.check_for_odirect_support('/dev/zero',
  489                                                         '/dev/def',
  490                                                         'iflag=direct')
  491         self.assertFalse(output)
  492         mock_exec.reset_mock()
  493 
  494         output = volume_utils.check_for_odirect_support('/dev/zero',
  495                                                         '/dev/def')
  496         self.assertTrue(output)
  497         mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/zero',
  498                                           'of=/dev/def', 'oflag=direct',
  499                                           run_as_root=True)
  500 
  501     @mock.patch('cinder.utils.execute',
  502                 side_effect=processutils.ProcessExecutionError)
  503     def test_check_for_odirect_support_error(self, mock_exec):
  504         output = volume_utils.check_for_odirect_support('/dev/abc', '/dev/def')
  505         self.assertFalse(output)
  506         mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/abc',
  507                                           'of=/dev/def', 'oflag=direct',
  508                                           run_as_root=True)
  509         mock_exec.reset_mock()
  510         output = volume_utils.check_for_odirect_support('/dev/zero',
  511                                                         '/dev/def')
  512         self.assertFalse(output)
  513         mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/zero',
  514                                           'of=/dev/def', 'oflag=direct',
  515                                           run_as_root=True)
  516 
  517 
  518 class ClearVolumeTestCase(test.TestCase):
  519     @mock.patch('cinder.volume.utils.copy_volume', return_value=None)
  520     @mock.patch('cinder.volume.utils.CONF')
  521     def test_clear_volume_conf(self, mock_conf, mock_copy):
  522         mock_conf.volume_clear = 'zero'
  523         mock_conf.volume_clear_size = 0
  524         mock_conf.volume_dd_blocksize = '1M'
  525         mock_conf.volume_clear_ionice = '-c3'
  526         output = volume_utils.clear_volume(1024, 'volume_path')
  527         self.assertIsNone(output)
  528         mock_copy.assert_called_once_with('/dev/zero', 'volume_path', 1024,
  529                                           '1M', sync=True,
  530                                           execute=utils.execute, ionice='-c3',
  531                                           throttle=None, sparse=False)
  532 
  533     @mock.patch('cinder.volume.utils.copy_volume', return_value=None)
  534     @mock.patch('cinder.volume.utils.CONF')
  535     def test_clear_volume_args(self, mock_conf, mock_copy):
  536         mock_conf.volume_clear = 'should_override_with_arg'
  537         mock_conf.volume_clear_size = 0
  538         mock_conf.volume_dd_blocksize = '1M'
  539         mock_conf.volume_clear_ionice = '-c3'
  540         output = volume_utils.clear_volume(1024, 'volume_path', 'zero', 1,
  541                                            '-c0')
  542         self.assertIsNone(output)
  543         mock_copy.assert_called_once_with('/dev/zero', 'volume_path', 1,
  544                                           '1M', sync=True,
  545                                           execute=utils.execute, ionice='-c0',
  546                                           throttle=None, sparse=False)
  547 
  548     @mock.patch('cinder.volume.utils.CONF')
  549     def test_clear_volume_invalid_opt(self, mock_conf):
  550         mock_conf.volume_clear = 'non_existent_volume_clearer'
  551         mock_conf.volume_clear_size = 0
  552         mock_conf.volume_clear_ionice = None
  553         self.assertRaises(exception.InvalidConfigurationValue,
  554                           volume_utils.clear_volume,
  555                           1024, "volume_path")
  556 
  557 
  558 class CopyVolumeTestCase(test.TestCase):
  559     @mock.patch('cinder.volume.utils.check_for_odirect_support',
  560                 return_value=True)
  561     @mock.patch('cinder.utils.execute')
  562     @mock.patch('cinder.volume.utils.CONF')
  563     def test_copy_volume_dd_iflag_and_oflag(self, mock_conf, mock_exec,
  564                                             mock_support):
  565         fake_throttle = throttling.Throttle(['fake_throttle'])
  566         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  567                                           sync=True, execute=utils.execute,
  568                                           ionice=None, throttle=fake_throttle)
  569         self.assertIsNone(output)
  570         mock_exec.assert_called_once_with('fake_throttle', 'dd',
  571                                           'if=/dev/zero',
  572                                           'of=/dev/null',
  573                                           'count=%s' % units.Gi,
  574                                           'bs=3M', 'iflag=count_bytes,direct',
  575                                           'oflag=direct', run_as_root=True)
  576 
  577         mock_exec.reset_mock()
  578 
  579         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  580                                           sync=False, execute=utils.execute,
  581                                           ionice=None, throttle=fake_throttle)
  582         self.assertIsNone(output)
  583         mock_exec.assert_called_once_with('fake_throttle', 'dd',
  584                                           'if=/dev/zero',
  585                                           'of=/dev/null',
  586                                           'count=%s' % units.Gi,
  587                                           'bs=3M', 'iflag=count_bytes,direct',
  588                                           'oflag=direct', run_as_root=True)
  589 
  590     @mock.patch('cinder.volume.utils.check_for_odirect_support',
  591                 return_value=False)
  592     @mock.patch('cinder.utils.execute')
  593     def test_copy_volume_dd_no_iflag_or_oflag(self, mock_exec, mock_support):
  594         fake_throttle = throttling.Throttle(['fake_throttle'])
  595         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  596                                           sync=True, execute=utils.execute,
  597                                           ionice=None, throttle=fake_throttle)
  598         self.assertIsNone(output)
  599         mock_exec.assert_called_once_with('fake_throttle', 'dd',
  600                                           'if=/dev/zero',
  601                                           'of=/dev/null',
  602                                           'count=%s' % units.Gi,
  603                                           'bs=3M', 'iflag=count_bytes',
  604                                           'conv=fdatasync', run_as_root=True)
  605 
  606         mock_exec.reset_mock()
  607 
  608         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  609                                           sync=False, execute=utils.execute,
  610                                           ionice=None, throttle=fake_throttle)
  611         self.assertIsNone(output)
  612         mock_exec.assert_called_once_with('fake_throttle', 'dd',
  613                                           'if=/dev/zero',
  614                                           'of=/dev/null',
  615                                           'count=%s' % units.Gi,
  616                                           'bs=3M', 'iflag=count_bytes',
  617                                           run_as_root=True)
  618 
  619     @mock.patch('cinder.volume.utils.check_for_odirect_support',
  620                 return_value=False)
  621     @mock.patch('cinder.utils.execute')
  622     def test_copy_volume_dd_no_throttle(self, mock_exec, mock_support):
  623         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  624                                           sync=True, execute=utils.execute,
  625                                           ionice=None)
  626         self.assertIsNone(output)
  627         mock_exec.assert_called_once_with('dd', 'if=/dev/zero', 'of=/dev/null',
  628                                           'count=%s' % units.Gi, 'bs=3M',
  629                                           'iflag=count_bytes',
  630                                           'conv=fdatasync', run_as_root=True)
  631 
  632     @mock.patch('cinder.volume.utils.check_for_odirect_support',
  633                 return_value=False)
  634     @mock.patch('cinder.utils.execute')
  635     def test_copy_volume_dd_with_ionice(self, mock_exec, mock_support):
  636         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  637                                           sync=True, execute=utils.execute,
  638                                           ionice='-c3')
  639         self.assertIsNone(output)
  640         mock_exec.assert_called_once_with('ionice', '-c3', 'dd',
  641                                           'if=/dev/zero', 'of=/dev/null',
  642                                           'count=%s' % units.Gi, 'bs=3M',
  643                                           'iflag=count_bytes',
  644                                           'conv=fdatasync', run_as_root=True)
  645 
  646     @mock.patch('cinder.volume.utils.check_for_odirect_support',
  647                 return_value=False)
  648     @mock.patch('cinder.utils.execute')
  649     def test_copy_volume_dd_with_sparse(self, mock_exec, mock_support):
  650         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  651                                           sync=True, execute=utils.execute,
  652                                           sparse=True)
  653         self.assertIsNone(output)
  654         mock_exec.assert_called_once_with('dd', 'if=/dev/zero', 'of=/dev/null',
  655                                           'count=%s' % units.Gi, 'bs=3M',
  656                                           'iflag=count_bytes',
  657                                           'conv=fdatasync,sparse',
  658                                           run_as_root=True)
  659 
  660     @mock.patch('cinder.volume.utils.check_for_odirect_support',
  661                 return_value=True)
  662     @mock.patch('cinder.utils.execute')
  663     def test_copy_volume_dd_with_sparse_iflag_and_oflag(self, mock_exec,
  664                                                         mock_support):
  665         output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, '3M',
  666                                           sync=True, execute=utils.execute,
  667                                           sparse=True)
  668         self.assertIsNone(output)
  669         mock_exec.assert_called_once_with('dd', 'if=/dev/zero', 'of=/dev/null',
  670                                           'count=%s' % units.Gi, 'bs=3M',
  671                                           'iflag=count_bytes,direct',
  672                                           'oflag=direct', 'conv=sparse',
  673                                           run_as_root=True)
  674 
  675     @mock.patch('cinder.volume.utils._copy_volume_with_file')
  676     def test_copy_volume_handles(self, mock_copy):
  677         handle1 = io.RawIOBase()
  678         handle2 = io.RawIOBase()
  679         output = volume_utils.copy_volume(handle1, handle2, 1024, 1)
  680         self.assertIsNone(output)
  681         mock_copy.assert_called_once_with(handle1, handle2, 1024)
  682 
  683     @mock.patch('cinder.volume.utils._transfer_data')
  684     @mock.patch('cinder.volume.utils._open_volume_with_path')
  685     def test_copy_volume_handle_transfer(self, mock_open, mock_transfer):
  686         handle = io.RawIOBase()
  687         output = volume_utils.copy_volume('/foo/bar', handle, 1024, 1)
  688         self.assertIsNone(output)
  689         mock_transfer.assert_called_once_with(mock.ANY, mock.ANY,
  690                                               1073741824, mock.ANY)
  691 
  692 
  693 @ddt.ddt
  694 class VolumeUtilsTestCase(test.TestCase):
  695     def test_null_safe_str(self):
  696         self.assertEqual('', volume_utils.null_safe_str(None))
  697         self.assertEqual('', volume_utils.null_safe_str(False))
  698         self.assertEqual('', volume_utils.null_safe_str(0))
  699         self.assertEqual('', volume_utils.null_safe_str([]))
  700         self.assertEqual('', volume_utils.null_safe_str(()))
  701         self.assertEqual('', volume_utils.null_safe_str({}))
  702         self.assertEqual('', volume_utils.null_safe_str(set()))
  703         self.assertEqual('a', volume_utils.null_safe_str('a'))
  704         self.assertEqual('1', volume_utils.null_safe_str(1))
  705         self.assertEqual('True', volume_utils.null_safe_str(True))
  706 
  707     @mock.patch('cinder.utils.get_root_helper')
  708     @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning')
  709     def test_supports_thin_provisioning(self, mock_supports_thin, mock_helper):
  710         self.assertEqual(mock_supports_thin.return_value,
  711                          volume_utils.supports_thin_provisioning())
  712         mock_helper.assert_called_once_with()
  713 
  714     @mock.patch('cinder.utils.get_root_helper')
  715     @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
  716     def test_get_all_physical_volumes(self, mock_get_vols, mock_helper):
  717         self.assertEqual(mock_get_vols.return_value,
  718                          volume_utils.get_all_physical_volumes())
  719         mock_helper.assert_called_once_with()
  720 
  721     @mock.patch('cinder.utils.get_root_helper')
  722     @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_volume_groups')
  723     def test_get_all_volume_groups(self, mock_get_groups, mock_helper):
  724         self.assertEqual(mock_get_groups.return_value,
  725                          volume_utils.get_all_volume_groups())
  726         mock_helper.assert_called_once_with()
  727 
  728     def test_generate_password(self):
  729         password = volume_utils.generate_password()
  730         self.assertTrue(any(c for c in password if c in '23456789'))
  731         self.assertTrue(any(c for c in password
  732                             if c in 'abcdefghijkmnopqrstuvwxyz'))
  733         self.assertTrue(any(c for c in password
  734                             if c in 'ABCDEFGHJKLMNPQRSTUVWXYZ'))
  735         self.assertEqual(16, len(password))
  736         self.assertEqual(10, len(volume_utils.generate_password(10)))
  737 
  738     @mock.patch('cinder.volume.utils.generate_password')
  739     def test_generate_username(self, mock_gen_pass):
  740         output = volume_utils.generate_username()
  741         self.assertEqual(mock_gen_pass.return_value, output)
  742 
  743     def test_extract_host(self):
  744         host = 'Host'
  745         # default level is 'backend'
  746         self.assertEqual(host,
  747                          volume_utils.extract_host(host))
  748         self.assertEqual(host,
  749                          volume_utils.extract_host(host, 'host'))
  750         self.assertEqual(host,
  751                          volume_utils.extract_host(host, 'backend'))
  752         # default_pool_name doesn't work for level other than 'pool'
  753         self.assertEqual(host,
  754                          volume_utils.extract_host(host, 'host', True))
  755         self.assertEqual(host,
  756                          volume_utils.extract_host(host, 'host', False))
  757         self.assertEqual(host,
  758                          volume_utils.extract_host(host, 'backend', True))
  759         self.assertEqual(host,
  760                          volume_utils.extract_host(host, 'backend', False))
  761         self.assertIsNone(volume_utils.extract_host(host, 'pool'))
  762         self.assertEqual('_pool0',
  763                          volume_utils.extract_host(host, 'pool', True))
  764 
  765         host = 'Host@Backend'
  766         self.assertEqual('Host@Backend',
  767                          volume_utils.extract_host(host))
  768         self.assertEqual('Host',
  769                          volume_utils.extract_host(host, 'host'))
  770         self.assertEqual(host,
  771                          volume_utils.extract_host(host, 'backend'))
  772         self.assertIsNone(volume_utils.extract_host(host, 'pool'))
  773         self.assertEqual('_pool0',
  774                          volume_utils.extract_host(host, 'pool', True))
  775 
  776         host = 'Host@Backend#Pool'
  777         pool = 'Pool'
  778         self.assertEqual('Host@Backend',
  779                          volume_utils.extract_host(host))
  780         self.assertEqual('Host',
  781                          volume_utils.extract_host(host, 'host'))
  782         self.assertEqual('Host@Backend',
  783                          volume_utils.extract_host(host, 'backend'))
  784         self.assertEqual(pool,
  785                          volume_utils.extract_host(host, 'pool'))
  786         self.assertEqual(pool,
  787                          volume_utils.extract_host(host, 'pool', True))
  788 
  789         host = 'Host#Pool'
  790         self.assertEqual('Host',
  791                          volume_utils.extract_host(host))
  792         self.assertEqual('Host',
  793                          volume_utils.extract_host(host, 'host'))
  794         self.assertEqual('Host',
  795                          volume_utils.extract_host(host, 'backend'))
  796         self.assertEqual(pool,
  797                          volume_utils.extract_host(host, 'pool'))
  798         self.assertEqual(pool,
  799                          volume_utils.extract_host(host, 'pool', True))
  800 
  801     def test_extract_host_none_string(self):
  802         self.assertRaises(exception.InvalidVolume,
  803                           volume_utils.extract_host,
  804                           None)
  805 
  806     def test_append_host(self):
  807         host = 'Host'
  808         pool = 'Pool'
  809         expected = 'Host#Pool'
  810         self.assertEqual(expected,
  811                          volume_utils.append_host(host, pool))
  812 
  813         pool = None
  814         expected = 'Host'
  815         self.assertEqual(expected,
  816                          volume_utils.append_host(host, pool))
  817 
  818         host = None
  819         pool = 'pool'
  820         expected = None
  821         self.assertEqual(expected,
  822                          volume_utils.append_host(host, pool))
  823 
  824         host = None
  825         pool = None
  826         expected = None
  827         self.assertEqual(expected,
  828                          volume_utils.append_host(host, pool))
  829 
  830     def test_compare_hosts(self):
  831         host_1 = 'fake_host@backend1'
  832         host_2 = 'fake_host@backend1#pool1'
  833         self.assertTrue(volume_utils.hosts_are_equivalent(host_1, host_2))
  834 
  835         host_2 = 'fake_host@backend1'
  836         self.assertTrue(volume_utils.hosts_are_equivalent(host_1, host_2))
  837 
  838         host_2 = 'fake_host2@backend1'
  839         self.assertFalse(volume_utils.hosts_are_equivalent(host_1, host_2))
  840 
  841     @mock.patch('cinder.volume.utils.CONF')
  842     def test_extract_id_from_volume_name_vol_id_pattern(self, conf_mock):
  843         conf_mock.volume_name_template = 'volume-%s'
  844         vol_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  845         vol_name = conf_mock.volume_name_template % vol_id
  846         result = volume_utils.extract_id_from_volume_name(vol_name)
  847         self.assertEqual(vol_id, result)
  848 
  849     @mock.patch('cinder.volume.utils.CONF')
  850     def test_extract_id_from_volume_name_vol_id_vol_pattern(self, conf_mock):
  851         conf_mock.volume_name_template = 'volume-%s-volume'
  852         vol_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  853         vol_name = conf_mock.volume_name_template % vol_id
  854         result = volume_utils.extract_id_from_volume_name(vol_name)
  855         self.assertEqual(vol_id, result)
  856 
  857     @mock.patch('cinder.volume.utils.CONF')
  858     def test_extract_id_from_volume_name_id_vol_pattern(self, conf_mock):
  859         conf_mock.volume_name_template = '%s-volume'
  860         vol_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  861         vol_name = conf_mock.volume_name_template % vol_id
  862         result = volume_utils.extract_id_from_volume_name(vol_name)
  863         self.assertEqual(vol_id, result)
  864 
  865     @mock.patch('cinder.volume.utils.CONF')
  866     def test_extract_id_from_volume_name_no_match(self, conf_mock):
  867         conf_mock.volume_name_template = '%s-volume'
  868         vol_name = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  869         result = volume_utils.extract_id_from_volume_name(vol_name)
  870         self.assertIsNone(result)
  871         vol_name = 'blahblahblah'
  872         result = volume_utils.extract_id_from_volume_name(vol_name)
  873         self.assertIsNone(result)
  874 
  875     @mock.patch('cinder.db.sqlalchemy.api.resource_exists', return_value=True)
  876     def test_check_managed_volume_already_managed(self, exists_mock):
  877         id_ = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  878         result = volume_utils.check_already_managed_volume(id_)
  879         self.assertTrue(result)
  880         exists_mock.assert_called_once_with(mock.ANY, models.Volume, id_)
  881 
  882     @mock.patch('cinder.db.sqlalchemy.api.resource_exists', return_value=False)
  883     def test_check_managed_volume_not_managed_proper_uuid(self, exists_mock):
  884         id_ = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  885         result = volume_utils.check_already_managed_volume(id_)
  886         self.assertFalse(result)
  887         exists_mock.assert_called_once_with(mock.ANY, models.Volume, id_)
  888 
  889     def test_check_managed_volume_not_managed_invalid_id(self):
  890         result = volume_utils.check_already_managed_volume(1)
  891         self.assertFalse(result)
  892         result = volume_utils.check_already_managed_volume('not-a-uuid')
  893         self.assertFalse(result)
  894 
  895     @mock.patch('cinder.volume.utils.CONF')
  896     def test_extract_id_from_snapshot_name(self, conf_mock):
  897         conf_mock.snapshot_name_template = '%s-snapshot'
  898         snap_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  899         snap_name = conf_mock.snapshot_name_template % snap_id
  900         result = volume_utils.extract_id_from_snapshot_name(snap_name)
  901         self.assertEqual(snap_id, result)
  902 
  903     @mock.patch('cinder.volume.utils.CONF')
  904     def test_extract_id_from_snapshot_name_no_match(self, conf_mock):
  905         conf_mock.snapshot_name_template = '%s-snapshot'
  906         snap_name = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
  907         result = volume_utils.extract_id_from_snapshot_name(snap_name)
  908         self.assertIsNone(result)
  909         snap_name = 'blahblahblah'
  910         result = volume_utils.extract_id_from_snapshot_name(snap_name)
  911         self.assertIsNone(result)
  912 
  913     @ddt.data({"name": "vol02"}, '{"name": "vol02"}')
  914     def test_paginate_entries_list_with_marker(self, marker):
  915         entries = [{'reference': {'name': 'vol03'}, 'size': 1},
  916                    {'reference': {'name': 'vol01'}, 'size': 3},
  917                    {'reference': {'name': 'vol02'}, 'size': 3},
  918                    {'reference': {'name': 'vol04'}, 'size': 2},
  919                    {'reference': {'name': 'vol06'}, 'size': 3},
  920                    {'reference': {'name': 'vol07'}, 'size': 1},
  921                    {'reference': {'name': 'vol05'}, 'size': 1}]
  922         expected = [{'reference': {'name': 'vol04'}, 'size': 2},
  923                     {'reference': {'name': 'vol03'}, 'size': 1},
  924                     {'reference': {'name': 'vol05'}, 'size': 1}]
  925         res = volume_utils.paginate_entries_list(entries, marker, 3,
  926                                                  1, ['size', 'reference'],
  927                                                  ['desc', 'asc'])
  928         self.assertEqual(expected, res)
  929 
  930     def test_paginate_entries_list_without_marker(self):
  931         entries = [{'reference': {'name': 'vol03'}, 'size': 1},
  932                    {'reference': {'name': 'vol01'}, 'size': 3},
  933                    {'reference': {'name': 'vol02'}, 'size': 3},
  934                    {'reference': {'name': 'vol04'}, 'size': 2},
  935                    {'reference': {'name': 'vol06'}, 'size': 3},
  936                    {'reference': {'name': 'vol07'}, 'size': 1},
  937                    {'reference': {'name': 'vol05'}, 'size': 1}]
  938         expected = [{'reference': {'name': 'vol07'}, 'size': 1},
  939                     {'reference': {'name': 'vol06'}, 'size': 3},
  940                     {'reference': {'name': 'vol05'}, 'size': 1}]
  941         res = volume_utils.paginate_entries_list(entries, None, 3, None,
  942                                                  ['reference'], ['desc'])
  943         self.assertEqual(expected, res)
  944 
  945     def test_paginate_entries_list_marker_invalid_format(self):
  946         entries = [{'reference': {'name': 'vol03'}, 'size': 1},
  947                    {'reference': {'name': 'vol01'}, 'size': 3}]
  948         self.assertRaises(exception.InvalidInput,
  949                           volume_utils.paginate_entries_list,
  950                           entries, "invalid_format", 3, None,
  951                           ['size', 'reference'], ['desc', 'asc'])
  952 
  953     def test_paginate_entries_list_marker_not_found(self):
  954         entries = [{'reference': {'name': 'vol03'}, 'size': 1},
  955                    {'reference': {'name': 'vol01'}, 'size': 3}]
  956         self.assertRaises(exception.InvalidInput,
  957                           volume_utils.paginate_entries_list,
  958                           entries, {'name': 'vol02'}, 3, None,
  959                           ['size', 'reference'], ['desc', 'asc'])
  960 
  961     def test_convert_config_string_to_dict(self):
  962         test_string = "{'key-1'='val-1' 'key-2'='val-2' 'key-3'='val-3'}"
  963         expected_dict = {'key-1': 'val-1', 'key-2': 'val-2', 'key-3': 'val-3'}
  964 
  965         self.assertEqual(
  966             expected_dict,
  967             volume_utils.convert_config_string_to_dict(test_string))
  968 
  969     @mock.patch('cinder.volume.volume_types.is_encrypted', return_value=False)
  970     def test_create_encryption_key_unencrypted(self, is_encrypted):
  971         result = volume_utils.create_encryption_key(mock.ANY,
  972                                                     mock.ANY,
  973                                                     fake.VOLUME_TYPE_ID)
  974         self.assertIsNone(result)
  975 
  976     @mock.patch('cinder.volume.volume_types.is_encrypted', return_value=True)
  977     @mock.patch('cinder.volume.volume_types.get_volume_type_encryption')
  978     @mock.patch('cinder.keymgr.conf_key_mgr.ConfKeyManager.create_key')
  979     def test_create_encryption_key_encrypted(self, create_key,
  980                                              get_volume_type_encryption,
  981                                              is_encryption):
  982         enc_key = {'cipher': 'aes-xts-plain64',
  983                    'key_size': 256,
  984                    'provider': 'p1',
  985                    'control_location': 'front-end',
  986                    'encryption_id': 'uuid1'}
  987         ctxt = context.get_admin_context()
  988         type_ref1 = volume_types.create(ctxt, "type1")
  989         encryption = db.volume_type_encryption_create(
  990             ctxt, type_ref1['id'], enc_key)
  991         get_volume_type_encryption.return_value = encryption
  992         CONF.set_override(
  993             'backend',
  994             'cinder.keymgr.conf_key_mgr.ConfKeyManager',
  995             group='key_manager')
  996         km = key_manager.API()
  997         volume_utils.create_encryption_key(ctxt,
  998                                            km,
  999                                            fake.VOLUME_TYPE_ID)
 1000         is_encryption.assert_called_once_with(ctxt,
 1001                                               fake.VOLUME_TYPE_ID)
 1002         get_volume_type_encryption.assert_called_once_with(
 1003             ctxt,
 1004             fake.VOLUME_TYPE_ID)
 1005         create_key.assert_called_once_with(ctxt,
 1006                                            algorithm='aes',
 1007                                            length=256)
 1008 
 1009     @ddt.data('<is> True', '<is> true', '<is> yes')
 1010     def test_is_replicated_spec_true(self, enabled):
 1011         res = volume_utils.is_replicated_spec({'replication_enabled': enabled})
 1012         self.assertTrue(res)
 1013 
 1014     @ddt.data({}, None, {'key': 'value'})
 1015     def test_is_replicated_no_specs(self, extra_specs):
 1016         res = volume_utils.is_replicated_spec(extra_specs)
 1017         self.assertFalse(bool(res))
 1018 
 1019     @ddt.data('<is> False', '<is> false', '<is> f', 'baddata', 'bad data')
 1020     def test_is_replicated_spec_false(self, enabled):
 1021         res = volume_utils.is_replicated_spec({'replication_enabled': enabled})
 1022         self.assertFalse(res)
 1023 
 1024     @mock.patch('cinder.db.group_get')
 1025     def test_group_get_by_id(self, mock_db_group_get):
 1026         expected = mock.Mock()
 1027         mock_db_group_get.return_value = expected
 1028         group_id = fake.GROUP_ID
 1029         actual = volume_utils.group_get_by_id(group_id)
 1030         self.assertEqual(expected, actual)
 1031 
 1032     @mock.patch('cinder.db.group_get')
 1033     def test_group_get_by_id_group_not_found(self, mock_db_group_get):
 1034         group_id = fake.GROUP_ID
 1035         mock_db_group_get.side_effect = exception.GroupNotFound(
 1036             group_id=group_id)
 1037         self.assertRaises(
 1038             exception.GroupNotFound,
 1039             volume_utils.group_get_by_id,
 1040             group_id
 1041         )
 1042 
 1043     @ddt.data('<is> False', None, 'notASpecValueWeCareAbout')
 1044     def test_is_group_a_cg_snapshot_type_is_false(self, spec_value):
 1045         with mock.patch('cinder.volume.group_types'
 1046                         '.get_group_type_specs') as mock_get_specs:
 1047             mock_get_specs.return_value = spec_value
 1048             group = fake_group.fake_group_obj(
 1049                 None, group_type_id=fake.GROUP_TYPE_ID)
 1050             self.assertFalse(volume_utils.is_group_a_cg_snapshot_type(group))
 1051 
 1052     @mock.patch('cinder.volume.group_types.get_group_type_specs')
 1053     def test_is_group_a_cg_snapshot_type_is_true(self, mock_get_specs):
 1054         mock_get_specs.return_value = '<is> True'
 1055         group = fake_group.fake_group_obj(
 1056             None, group_type_id=fake.GROUP_TYPE_ID)
 1057         self.assertTrue(volume_utils.is_group_a_cg_snapshot_type(group))
 1058 
 1059     @ddt.data({'max_over_subscription_ratio': '10', 'supports_auto': True},
 1060               {'max_over_subscription_ratio': 'auto', 'supports_auto': True},
 1061               {'max_over_subscription_ratio': 'auto', 'supports_auto': False},
 1062               {'max_over_subscription_ratio': '1.2', 'supports_auto': False},)
 1063     @ddt.unpack
 1064     def test_get_max_over_subscription_ratio(self,
 1065                                              max_over_subscription_ratio,
 1066                                              supports_auto):
 1067 
 1068         if not supports_auto and max_over_subscription_ratio == 'auto':
 1069             self.assertRaises(exception.VolumeDriverException,
 1070                               volume_utils.get_max_over_subscription_ratio,
 1071                               max_over_subscription_ratio, supports_auto)
 1072         elif not supports_auto:
 1073             mosr = volume_utils.get_max_over_subscription_ratio(
 1074                 max_over_subscription_ratio, supports_auto)
 1075             self.assertEqual(float(max_over_subscription_ratio), mosr)
 1076         else:  # supports_auto
 1077             mosr = volume_utils.get_max_over_subscription_ratio(
 1078                 max_over_subscription_ratio, supports_auto)
 1079             if max_over_subscription_ratio == 'auto':
 1080                 self.assertEqual(max_over_subscription_ratio, mosr)
 1081             else:
 1082                 self.assertEqual(float(max_over_subscription_ratio), mosr)
 1083 
 1084     def test_make_initiator_target_all2all_map(self):
 1085         initiator_wwpns = ['ff00000000000000', 'ff00000000000001']
 1086         target_wwpns = ['bb00000000000000', 'bb00000000000001']
 1087 
 1088         expected = {
 1089             'ff00000000000000': ['bb00000000000000', 'bb00000000000001'],
 1090             'ff00000000000001': ['bb00000000000000', 'bb00000000000001']
 1091         }
 1092 
 1093         ret = volume_utils.make_initiator_target_all2all_map(initiator_wwpns,
 1094                                                              target_wwpns)
 1095         self.assertEqual(ret, expected)