Introduce Blockdevice object

This object should simplify the usage of the block_device dictionary
provided by nova. Attributes can now easily be accesed via object
properties and method, instead of doing complex operations on the dict.

This patch does not yet exploit this object. This is done in a follow
up patch.

Change-Id: I4b88761c86dc8996fb490cf8680ceaf1480d445d
This commit is contained in:
Andreas Scheuring 2018-01-16 09:56:39 +01:00
parent bce9106265
commit 8aef196ad4
2 changed files with 122 additions and 0 deletions

View File

@ -0,0 +1,61 @@
# Copyright 2018 IBM Corp. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from nova.test import TestCase
from nova_dpm.virt.dpm.block_device import BlockDevice
from nova_dpm.virt.dpm import exceptions
PARTITION_WWPN = 'CCCCCCCCCCCCCCCC'
LUN = 0
BLOCK_DEVICE = {
'connection_info': {
'driver_volume_type': 'fibre_channel',
'connector': {
'wwpns': [PARTITION_WWPN],
'host': '3cfb165c-0df3-4d80-87b2-4c353e61318f'},
'data': {
'initiator_target_map': {
PARTITION_WWPN: [
'AAAAAAAAAAAAAAAA',
'BBBBBBBBBBBBBBBB']},
'target_discovered': False,
'target_lun': LUN}}}
class BlockDeviceTest(TestCase):
def test_volume_type_unsupported(self):
bd = {'connection_info': {'driver_volume_type': 'fake_vol_type'}}
try:
BlockDevice(bd)
except exceptions.UnsupportedVolumeTypeException:
pass
else:
self.fail("UnsupportedVolumeTypeException not raised.")
def test_volume_type_fc_supported(self):
bd = {'connection_info': {'driver_volume_type': 'fibre_channel'}}
bd_obj = BlockDevice(bd)
self.assertIsNotNone(bd_obj)
def test_get_target_wwpn(self):
bd = BlockDevice(BLOCK_DEVICE)
self.flags(group="dpm", target_wwpn_ignore_list=["AAAAAAAAAAAAAAAA"])
self.assertEqual("BBBBBBBBBBBBBBBB",
bd.get_target_wwpn(PARTITION_WWPN))
def test_lun(self):
bd = BlockDevice(BLOCK_DEVICE)
self.assertEqual(str(LUN), bd.lun)

View File

@ -0,0 +1,61 @@
# Copyright 2018 IBM Corp. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from nova_dpm import conf
from nova_dpm.virt.dpm import exceptions
CONF = conf.CONF
class BlockDevice(object):
def __init__(self, block_device):
self.bd = block_device
self._validate_volume_type()
def _validate_volume_type(self):
vol_type = self.bd['connection_info']['driver_volume_type']
if vol_type != 'fibre_channel':
raise exceptions.UnsupportedVolumeTypeException(
vol_type=vol_type)
@property
def host_wwpns(self):
return self.bd['connection_info']['connector']['wwpns']
def get_target_wwpn(self, partition_wwpn):
if partition_wwpn not in self.host_wwpns:
raise Exception('Partition WWPN not found from cinder')
bd_target_wwpns = (
self.bd['connection_info']['data']['initiator_target_map']
[partition_wwpn])
# Some storage systems return a complete list of target WWPNs on
# cinders request. Only some of them might be available in our FC
# network. Therefore we need to filter out the invalid ones. The user
# defines the invalid ones via a config setting.
valid_target_wwpns = [
wwpn for wwpn in bd_target_wwpns
if wwpn not in CONF.dpm.target_wwpn_ignore_list]
# target_wwpns is a list of wwpns which will be accessible
# from host wwpn. So we can use any of the target wwpn in the
# list.
target_wwpn = (valid_target_wwpns[0]
if len(valid_target_wwpns) > 0 else '')
return target_wwpn
@property
def lun(self):
return str(self.bd['connection_info']['data']['target_lun'])