openstack
/
cinder
Code Issues Proposed changes
cinder/cinder/tests/test_huawei_t_dorado.py

1716 lines
72 KiB
Python
Raw Blame History

# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2013 Huawei Technologies Co., Ltd.
# Copyright (c) 2012 OpenStack Foundation
# 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.
"""
Unit Tests for Huawei T and Dorado volume drivers.
"""
import mox
import os
import shutil
import socket
import tempfile
import time
from xml.dom.minidom import Document
from xml.etree import ElementTree as ET
from cinder import context
from cinder import exception
from cinder import test
from cinder import utils
from cinder.volume import configuration as conf
from cinder.volume.drivers.huawei import huawei_utils
from cinder.volume.drivers.huawei import HuaweiVolumeDriver
from cinder.volume.drivers.huawei import ssh_common
from cinder.volume import volume_types
LUN_INFO = {'ID': None,
'Name': None,
'Size': None,
'LUN WWN': None,
'Status': None,
'Visible Capacity': None,
'Disk Pool ID': None,
'Cache Prefetch Strategy': None,
'Lun Type': None,
'Consumed Capacity': None,
'Pool ID': None,
'SnapShot ID': None,
'LunCopy ID': None,
'Owner Controller': None,
'Worker Controller': None,
'RAID Group ID': None}
CLONED_LUN_INFO = {'ID': None,
'Name': None,
'Size': None,
'LUN WWN': None,
'Status': None,
'Visible Capacity': None,
'Disk Pool ID': None,
'Cache Prefetch Strategy': None,
'Lun Type': None,
'Consumed Capacity': None,
'Pool ID': None,
'SnapShot ID': None,
'LunCopy ID': None,
'Owner Controller': None,
'Worker Controller': None,
'RAID Group ID': None}
SNAPSHOT_INFO = {'Source LUN ID': None,
'Source LUN Name': None,
'ID': None,
'Name': None,
'Type': 'Public',
'Status': None}
MAP_INFO = {'Host Group ID': None,
'Host Group Name': None,
'Host ID': None,
'Host Name': None,
'Os Type': None,
'INI Port ID': None,
'INI Port Name': None,
'INI Port Info': None,
'INI Port WWN': None,
'INI Port Type': None,
'Link Status': None,
'LUN WWN': None,
'DEV LUN ID': None,
'Host LUN ID': None,
'CHAP status': False}
HOST_PORT_INFO = {'ID': None,
'Name': None,
'Info': None,
'WWN': None,
'Type': None}
LUNCOPY_INFO = {'Name': None,
'ID': None,
'Type': None,
'State': None,
'Status': None}
LUNCOPY_SETTING = {'ID': '1',
'Type': 'FULL',
'State': 'Created',
'Status': 'Normal'}
POOL_SETTING = {'ID': '2',
'Level': 'RAID6',
'Status': 'Normal',
'Free Capacity': '10240',
'Disk List': '0,1;0,2;0,3;0,4;0,5;0,6',
'Name': 'RAID_001',
'Type': 'Thick'}
INITIATOR_SETTING = {'TargetIQN': 'iqn.2006-08.com.huawei:oceanspace:2103037:',
'TargetIQN-form': 'iqn.2006-08.com.huawei:oceanspace:'
'2103037::1020001:192.168.100.2',
'Initiator Name': 'iqn.1993-08.debian:01:ec2bff7ac3a3',
'Initiator TargetIP': '192.168.100.2',
'WWN': ['2011666666666565']}
FAKE_VOLUME = {'name': 'Volume-lele34fe-223f-dd33-4423-asdfghjklqwe',
'id': 'lele34fe-223f-dd33-4423-asdfghjklqwe',
'size': '2',
'provider_auth': None,
'volume_type_id': None,
'provider_location': None}
FAKE_CLONED_VOLUME = {'name': 'Volume-jeje34fe-223f-dd33-4423-asdfghjklqwg',
'id': 'jeje34fe-223f-dd33-4423-asdfghjklqwg',
'size': '3',
'provider_auth': None,
'volume_type_id': None,
'provider_location': None}
FAKE_SNAPSHOT = {'name': 'keke34fe-223f-dd33-4423-asdfghjklqwf',
'id': '223f-dd33-4423-asdfghjklqwf',
'volume_name': 'Volume-lele34fe-223f-dd33-4423-asdfghjklqwe',
'provider_location': None}
FAKE_CONNECTOR = {'initiator': 'iqn.1993-08.debian:01:ec2bff7ac3a3',
'wwpns': ['1000000164s45126'],
'wwnns': ['2000666666666565'],
'host': 'fakehost',
'ip': '10.10.0.1'}
RESPOOL_A_SIM = {'Size': '10240', 'Valid Size': '5120'}
RESPOOL_B_SIM = {'Size': '10240', 'Valid Size': '10240'}
VOLUME_SNAP_ID = {'vol': '0', 'vol_copy': '1', 'snap': '2'}
cmd_error_list = [] # CLI cmds in this list will run failed
Curr_test = [''] # show current testing driver
class FakeChannel():
def __init__(self):
if Curr_test[0] == 'T':
self.simu = HuaweiTCLIResSimulator()
elif Curr_test[0] == 'Dorado5100':
self.simu = HuaweiDorado5100CLIResSimulator()
else:
self.simu = HuaweiDorado2100G2CLIResSimulator()
def resize_pty(self, width=80, height=24):
pass
def settimeout(self, time):
pass
def send(self, s):
self.command = s
def recv(self, nbytes):
command = self.command.split()
cmd = command[0]
params = command[1:]
if cmd in cmd_error_list:
reset_error_flg(cmd)
out = self.command[:-1] + 'ERROR' + '\nadmin:/>'
return out.replace('\n', '\r\n')
func_name = 'cli_' + cmd
cli_func = getattr(self.simu, func_name)
out = cli_func(params)
out = self.command[:-1] + out + '\nadmin:/>'
return out.replace('\n', '\r\n')
def close(self):
pass
class FakeSSHClient():
def invoke_shell(self):
return FakeChannel()
def get_transport(self):
class transport():
def __init__(self):
self.sock = sock()
class sock():
def settimeout(self, time):
pass
return transport()
def close(self):
pass
class FakeSSHPool():
def __init__(self, ip, port, conn_timeout, login, password=None,
*args, **kwargs):
self.ip = ip
self.port = port
self.login = login
self.password = password
def create(self):
return FakeSSHClient()
def get(self):
return FakeSSHClient()
def put(self, ssh):
pass
def remove(self, ssh):
pass
def Fake_sleep(time):
pass
def Fake_change_file_mode(obj, filepath):
pass
def create_fake_conf_file(filename):
doc = Document()
config = doc.createElement('config')
doc.appendChild(config)
storage = doc.createElement('Storage')
config.appendChild(storage)
product = doc.createElement('Product')
product_text = doc.createTextNode('T')
product.appendChild(product_text)
storage.appendChild(product)
config.appendChild(storage)
protocol = doc.createElement('Protocol')
protocol_text = doc.createTextNode('iSCSI')
protocol.appendChild(protocol_text)
storage.appendChild(protocol)
controllerip0 = doc.createElement('ControllerIP0')
controllerip0_text = doc.createTextNode('10.10.10.1')
controllerip0.appendChild(controllerip0_text)
storage.appendChild(controllerip0)
controllerip1 = doc.createElement('ControllerIP1')
controllerip1_text = doc.createTextNode('10.10.10.2')
controllerip1.appendChild(controllerip1_text)
storage.appendChild(controllerip1)
username = doc.createElement('UserName')
username_text = doc.createTextNode('admin')
username.appendChild(username_text)
storage.appendChild(username)
userpassword = doc.createElement('UserPassword')
userpassword_text = doc.createTextNode('123456')
userpassword.appendChild(userpassword_text)
storage.appendChild(userpassword)
lun = doc.createElement('LUN')
config.appendChild(lun)
storagepool = doc.createElement('StoragePool')
storagepool.setAttribute('Name', 'RAID_001')
lun.appendChild(storagepool)
luntype = doc.createElement('LUNType')
luntype_text = doc.createTextNode('Thick')
luntype.appendChild(luntype_text)
lun.appendChild(luntype)
iscsi = doc.createElement('iSCSI')
config.appendChild(iscsi)
defaulttargetip = doc.createElement('DefaultTargetIP')
defaulttargetip_text = doc.createTextNode('192.168.100.1')
defaulttargetip.appendChild(defaulttargetip_text)
iscsi.appendChild(defaulttargetip)
initiator = doc.createElement('Initiator')
initiator.setAttribute('Name', 'iqn.1993-08.debian:01:ec2bff7ac3a3')
initiator.setAttribute('TargetIP', '192.168.100.2')
iscsi.appendChild(initiator)
os_type = doc.createElement('Host')
os_type.setAttribute('OSType', 'Linux')
os_type.setAttribute('HostIP', '10.10.0.1')
config.appendChild(os_type)
tmp_file = open(filename, 'w')
tmp_file.write(doc.toprettyxml(indent=''))
tmp_file.close()
def modify_conf(conf, item, val, attrib=None):
tree = ET.parse(conf)
root = tree.getroot()
conf_item = root.find('%s' % item)
if not attrib:
conf_item.text = '%s' % val
else:
conf_item.attrib['%s' % attrib] = '%s' % val
tree.write(conf, 'UTF-8')
def set_error_flg(cmd):
cmd_error_list.append(cmd)
def reset_error_flg(cmd):
cmd_error_list.remove(cmd)
class HuaweiTCLIResSimulator():
def _name_translate(self, name):
return 'OpenStack_' + str(hash(name))
def cli_showsys(self, params):
pass
def cli_createlun(self, params):
lun_type = ('THIN' if '-pool' in params else 'THICK')
if LUN_INFO['ID'] is None:
LUN_INFO['Name'] = self._name_translate(FAKE_VOLUME['name'])
LUN_INFO['ID'] = VOLUME_SNAP_ID['vol']
LUN_INFO['Size'] = FAKE_VOLUME['size']
LUN_INFO['Lun Type'] = lun_type
LUN_INFO['Owner Controller'] = 'A'
LUN_INFO['Worker Controller'] = 'A'
LUN_INFO['RAID Group ID'] = POOL_SETTING['ID']
FAKE_VOLUME['provider_location'] = LUN_INFO['ID']
else:
CLONED_LUN_INFO['Name'] = \
self._name_translate(FAKE_CLONED_VOLUME['name'])
CLONED_LUN_INFO['ID'] = VOLUME_SNAP_ID['vol_copy']
CLONED_LUN_INFO['Size'] = FAKE_CLONED_VOLUME['size']
CLONED_LUN_INFO['Lun Type'] = lun_type
CLONED_LUN_INFO['Owner Controller'] = 'A'
CLONED_LUN_INFO['Worker Controller'] = 'A'
CLONED_LUN_INFO['RAID Group ID'] = POOL_SETTING['ID']
FAKE_CLONED_VOLUME['provider_location'] = CLONED_LUN_INFO['ID']
out = 'command operates successfully'
return out
def cli_showlun(self, params):
if '-lun' not in params:
if LUN_INFO['ID'] is None:
out = 'command operates successfully, but no information.'
elif CLONED_LUN_INFO['ID'] is None:
out = """/>showlun
===========================================================================
LUN Information
---------------------------------------------------------------------------
ID RAID Group ID Disk Pool ID Status Controller Visible Capacity(MB) \
LUN Name Stripe Unit Size(KB) Lun Type
---------------------------------------------------------------------------
%s %s -- Normal %s %s %s 64 THICK
===========================================================================
""" % (LUN_INFO['ID'], LUN_INFO['RAID Group ID'], LUN_INFO['Owner Controller'],
str(int(LUN_INFO['Size']) * 1024), LUN_INFO['Name'])
else:
out = """/>showlun
============================================================================
LUN Information
----------------------------------------------------------------------------
ID RAID Group ID Disk Pool ID Status Controller Visible Capacity(MB)\
LUN Name Stripe Unit Size(KB) Lun Type
----------------------------------------------------------------------------
%s %s -- Normal %s %s %s 64 THICK
%s %s -- Normal %s %s %s 64 THICK
============================================================================
""" % (LUN_INFO['ID'], LUN_INFO['RAID Group ID'], LUN_INFO['Owner Controller'],
str(int(LUN_INFO['Size']) * 1024), LUN_INFO['Name'],
CLONED_LUN_INFO['ID'], CLONED_LUN_INFO['RAID Group ID'],
CLONED_LUN_INFO['Owner Controller'],
str(int(CLONED_LUN_INFO['Size']) * 1024),
CLONED_LUN_INFO['Name'])
elif params[params.index('-lun') + 1] in VOLUME_SNAP_ID.values():
out = """/>showlun
================================================
LUN Information
------------------------------------------------
ID | %s
Name | %s
LUN WWN | --
Visible Capacity | %s
RAID GROUP ID | %s
Owning Controller | %s
Workong Controller | %s
Lun Type | %s
SnapShot ID | %s
LunCopy ID | %s
================================================
""" % ((LUN_INFO['ID'], LUN_INFO['Name'], LUN_INFO['Visible Capacity'],
LUN_INFO['RAID Group ID'], LUN_INFO['Owner Controller'],
LUN_INFO['Worker Controller'], LUN_INFO['Lun Type'],
LUN_INFO['SnapShot ID'], LUN_INFO['LunCopy ID'])
if params[params.index('-lun') + 1] == VOLUME_SNAP_ID['vol'] else
(CLONED_LUN_INFO['ID'], CLONED_LUN_INFO['Name'],
CLONED_LUN_INFO['Visible Capacity'], CLONED_LUN_INFO['RAID Group ID'],
CLONED_LUN_INFO['Owner Controller'],
CLONED_LUN_INFO['Worker Controller'],
CLONED_LUN_INFO['Lun Type'], CLONED_LUN_INFO['SnapShot ID'],
CLONED_LUN_INFO['LunCopy ID']))
else:
out = 'ERROR: The object does not exist.'
return out
def cli_dellun(self, params):
if params[params.index('-lun') + 1] == VOLUME_SNAP_ID['vol']:
LUN_INFO['Name'] = None
LUN_INFO['ID'] = None
LUN_INFO['Size'] = None
LUN_INFO['Lun Type'] = None
LUN_INFO['LUN WWN'] = None
LUN_INFO['Owner Controller'] = None
LUN_INFO['Worker Controller'] = None
LUN_INFO['RAID Group ID'] = None
FAKE_VOLUME['provider_location'] = None
else:
CLONED_LUN_INFO['Name'] = None
CLONED_LUN_INFO['ID'] = None
CLONED_LUN_INFO['Size'] = None
CLONED_LUN_INFO['Lun Type'] = None
CLONED_LUN_INFO['LUN WWN'] = None
CLONED_LUN_INFO['Owner Controller'] = None
CLONED_LUN_INFO['Worker Controller'] = None
CLONED_LUN_INFO['RAID Group ID'] = None
CLONED_LUN_INFO['provider_location'] = None
FAKE_CLONED_VOLUME['provider_location'] = None
out = 'command operates successfully'
return out
def cli_showrg(self, params):
out = """/>showrg
=====================================================================
RAID Group Information
---------------------------------------------------------------------
ID Level Status Free Capacity(MB) Disk List Name
---------------------------------------------------------------------
0 RAID6 Normal 1024 0,0;0,2; RAID003
%s %s %s %s %s %s
=====================================================================
-""" % (POOL_SETTING['ID'], POOL_SETTING['Level'],
POOL_SETTING['Status'], POOL_SETTING['Free Capacity'],
POOL_SETTING['Disk List'], POOL_SETTING['Name'])
return out
def cli_showpool(self, params):
out = """/>showpool
=====================================================================
Pool Information
---------------------------------------------------------------------
Level Status Available Capacity(MB) Disk List
---------------------------------------------------------------------
RAID6 Normal %s 0,0;0,2;0,4;0,5;
=====================================================================
-""" % POOL_SETTING['Free Capacity']
return out
def cli_createluncopy(self, params):
src_id = params[params.index('-slun') + 1]
tgt_id = params[params.index('-tlun') + 1]
LUNCOPY_INFO['Name'] = 'OpenStack_%s_%s' % (src_id, tgt_id)
LUNCOPY_INFO['ID'] = LUNCOPY_SETTING['ID']
LUNCOPY_INFO['Type'] = LUNCOPY_SETTING['Type']
LUNCOPY_INFO['State'] = LUNCOPY_SETTING['State']
LUNCOPY_INFO['Status'] = LUNCOPY_SETTING['Status']
out = 'command operates successfully'
return out
def cli_chgluncopystatus(self, params):
LUNCOPY_INFO['State'] = 'Start'
out = 'command operates successfully'
return out
def cli_showluncopy(self, params):
if LUNCOPY_INFO['State'] == 'Start':
LUNCOPY_INFO['State'] = 'Copying'
elif LUNCOPY_INFO['State'] == 'Copying':
LUNCOPY_INFO['State'] = 'Complete'
out = """/>showluncopy
============================================================================
LUN Copy Information
----------------------------------------------------------------------------
LUN Copy Name LUN Copy ID Type LUN Copy State LUN Copy Status
----------------------------------------------------------------------------
%s %s %s %s %s
============================================================================
""" % (LUNCOPY_INFO['Name'], LUNCOPY_INFO['ID'], LUNCOPY_INFO['Type'],
LUNCOPY_INFO['State'], LUNCOPY_INFO['Status'])
return out
def cli_delluncopy(self, params):
LUNCOPY_INFO['Name'] = None
LUNCOPY_INFO['ID'] = None
LUNCOPY_INFO['Type'] = None
LUNCOPY_INFO['State'] = None
LUNCOPY_INFO['Status'] = None
out = 'command operates successfully'
return out
def cli_createsnapshot(self, params):
SNAPSHOT_INFO['Source LUN ID'] = LUN_INFO['ID']
SNAPSHOT_INFO['Source LUN Name'] = LUN_INFO['Name']
SNAPSHOT_INFO['ID'] = VOLUME_SNAP_ID['snap']
SNAPSHOT_INFO['Name'] =\
self._name_translate(FAKE_SNAPSHOT['name'])
SNAPSHOT_INFO['Status'] = 'Disable'
out = 'command operates successfully'
return out
def cli_showsnapshot(self, params):
if SNAPSHOT_INFO['ID'] is None:
out = 'command operates successfully, but no information.'
else:
out = """/>showsnapshot
==========================================================================
Snapshot Information
--------------------------------------------------------------------------
Name ID Type Status Time Stamp
--------------------------------------------------------------------------
%s %s Public %s 2013-01-15 14:21:13
==========================================================================
""" % (SNAPSHOT_INFO['Name'], SNAPSHOT_INFO['ID'], SNAPSHOT_INFO['Status'])
return out
def cli_actvsnapshot(self, params):
SNAPSHOT_INFO['Status'] = 'Active'
FAKE_SNAPSHOT['provider_location'] = SNAPSHOT_INFO['ID']
out = 'command operates successfully'
return out
def cli_disablesnapshot(self, params):
SNAPSHOT_INFO['Status'] = 'Disable'
out = 'command operates successfully'
return out
def cli_delsnapshot(self, params):
SNAPSHOT_INFO['Source LUN ID'] = None
SNAPSHOT_INFO['Source LUN Name'] = None
SNAPSHOT_INFO['ID'] = None
SNAPSHOT_INFO['Name'] = None
SNAPSHOT_INFO['Status'] = None
FAKE_SNAPSHOT['provider_location'] = None
out = 'command operates successfully'
return out
def cli_showrespool(self, params):
out = """/>showrespool
===========================================================================
Resource Pool Information
---------------------------------------------------------------------------
Pool ID Size(MB) Usage(MB) Valid Size(MB) Alarm Threshold
---------------------------------------------------------------------------
A %s 0.0 %s 80
B %s 0.0 %s 80
===========================================================================
-""" % (RESPOOL_A_SIM['Size'], RESPOOL_A_SIM['Valid Size'],
RESPOOL_B_SIM['Size'], RESPOOL_B_SIM['Valid Size'])
return out
def cli_showiscsitgtname(self, params):
iqn = INITIATOR_SETTING['TargetIQN']
out = """/>showiscsitgtname
===================================================================
ISCSI Name
-------------------------------------------------------------------
Iscsi Name | %s
===================================================================
""" % iqn
return out
def cli_showiscsiip(self, params):
out = """/>showiscsiip
============================================================================
iSCSI IP Information
----------------------------------------------------------------------------
Controller ID Interface Module ID Port ID IP Address Mask
----------------------------------------------------------------------------
B 0 P1 %s 255.255.255.0
============================================================================
-""" % INITIATOR_SETTING['Initiator TargetIP']
return out
def cli_showhostgroup(self, params):
if MAP_INFO['Host Group ID'] is None:
out = """/>showhostgroup
============================================================
Host Group Information
------------------------------------------------------------
Host Group ID Name File Engine Cluster
------------------------------------------------------------
0 Default Group NO
============================================================
"""
else:
out = """/>showhostgroup
============================================================
Host Group Information
------------------------------------------------------------
Host Group ID Name File Engine Cluster
------------------------------------------------------------
0 Default Group NO
%s %s NO
============================================================
""" % (MAP_INFO['Host Group ID'], MAP_INFO['Host Group Name'])
return out
def cli_createhostgroup(self, params):
MAP_INFO['Host Group ID'] = '1'
MAP_INFO['Host Group Name'] = 'HostGroup_OpenStack'
out = 'command operates successfully'
return out
def cli_showhost(self, params):
if MAP_INFO['Host ID'] is None:
out = 'command operates successfully, but no information.'
else:
out = """/>showhost
=======================================================
Host Information
-------------------------------------------------------
Host ID Host Name Host Group ID Os Type
-------------------------------------------------------
%s %s %s Linux
=======================================================
""" % (MAP_INFO['Host ID'], MAP_INFO['Host Name'], MAP_INFO['Host Group ID'])
return out
def cli_addhost(self, params):
MAP_INFO['Host ID'] = '1'
MAP_INFO['Host Name'] = 'Host_' + FAKE_CONNECTOR['host']
MAP_INFO['Os Type'] = 'Linux'
out = 'command operates successfully'
return out
def cli_delhost(self, params):
MAP_INFO['Host ID'] = None
MAP_INFO['Host Name'] = None
MAP_INFO['Os Type'] = None
out = 'command operates successfully'
return out
def cli_showiscsiini(self, params):
if HOST_PORT_INFO['ID'] is None:
out = 'Error: The parameter is wrong.'
else:
out = """/>showiscsiini
========================================================
Initiator Information
--------------------------------------------------------
Initiator Name Chap Status
--------------------------------------------------------
%s Disable
========================================================
""" % HOST_PORT_INFO['Info']
return out
def cli_addiscsiini(self, params):
HOST_PORT_INFO['ID'] = '1'
HOST_PORT_INFO['Name'] = 'iSCSIInitiator001'
HOST_PORT_INFO['Info'] = INITIATOR_SETTING['Initiator Name']
HOST_PORT_INFO['Type'] = 'ISCSITGT'
out = 'command operates successfully'
return out
def cli_deliscsiini(self, params):
HOST_PORT_INFO['ID'] = None
HOST_PORT_INFO['Name'] = None
HOST_PORT_INFO['Info'] = None
HOST_PORT_INFO['Type'] = None
out = 'command operates successfully'
return out
def cli_showhostport(self, params):
if MAP_INFO['INI Port ID'] is None:
out = 'command operates successfully, but no information.'
else:
out = """/>showhostport
============================================================================
Host Port Information
----------------------------------------------------------------------------
Port ID Port Name Port Information Port Type Host ID Link Status \
Multipath Type
----------------------------------------------------------------------------
%s %s %s %s %s Unconnected Default
============================================================================
""" % (MAP_INFO['INI Port ID'], MAP_INFO['INI Port Name'],
MAP_INFO['INI Port Info'], MAP_INFO['INI Port Type'],
MAP_INFO['Host ID'])
return out
def cli_addhostport(self, params):
MAP_INFO['INI Port ID'] = HOST_PORT_INFO['ID']
MAP_INFO['INI Port Name'] = HOST_PORT_INFO['Name']
MAP_INFO['INI Port Info'] = HOST_PORT_INFO['Info']
MAP_INFO['INI Port Type'] = HOST_PORT_INFO['Type']
out = 'command operates successfully'
return out
def cli_delhostport(self, params):
MAP_INFO['INI Port ID'] = None
MAP_INFO['INI Port Name'] = None
MAP_INFO['INI Port Info'] = None
MAP_INFO['INI Port Type'] = None
HOST_PORT_INFO['ID'] = None
HOST_PORT_INFO['Name'] = None
HOST_PORT_INFO['Info'] = None
HOST_PORT_INFO['Type'] = None
out = 'command operates successfully'
return out
def cli_showhostmap(self, params):
if MAP_INFO['DEV LUN ID'] is None:
out = 'command operates successfully, but no information.'
else:
out = """/>showhostmap
===========================================================================
Map Information
---------------------------------------------------------------------------
Map ID Working Controller Dev LUN ID LUN WWN Host LUN ID Mapped to\
RAID ID Dev LUN Cap(MB) Map Type Whether Command LUN Pool ID
----------------------------------------------------------------------------
2147483649 %s %s %s %s Host: %s %s %s HOST No --
============================================================================
""" % (LUN_INFO['Worker Controller'], LUN_INFO['ID'], LUN_INFO['LUN WWN'],
MAP_INFO['Host LUN ID'], MAP_INFO['Host ID'], LUN_INFO['RAID Group ID'],
str(int(LUN_INFO['Size']) * 1024))
return out
def cli_addhostmap(self, params):
MAP_INFO['DEV LUN ID'] = LUN_INFO['ID']
MAP_INFO['LUN WWN'] = LUN_INFO['LUN WWN']
MAP_INFO['Host LUN ID'] = '2'
MAP_INFO['Link Status'] = 'Linked'
out = 'command operates successfully'
return out
def cli_delhostmap(self, params):
if MAP_INFO['Link Status'] == 'Linked':
MAP_INFO['Link Status'] = 'Deleting'
out = 'there are IOs accessing the system, please try later'
else:
MAP_INFO['Link Status'] = None
MAP_INFO['DEV LUN ID'] = None
MAP_INFO['LUN WWN'] = None
MAP_INFO['Host LUN ID'] = None
out = 'command operates successfully'
return out
def cli_showfreeport(self, params):
out = """/>showfreeport
=======================================================================
Host Free Port Information
-----------------------------------------------------------------------
WWN Or MAC Type Location Connection Status
-----------------------------------------------------------------------
1000000164s45126 FC Primary Controller Connected
=======================================================================
"""
HOST_PORT_INFO['ID'] = '2'
HOST_PORT_INFO['Name'] = 'FCInitiator001'
HOST_PORT_INFO['Info'] = '1000000164s45126'
HOST_PORT_INFO['Type'] = 'FC'
return out
def cli_showhostpath(self, params):
host = params[params.index('-host') + 1]
out = """/>showhostpath -host 1
=======================================
Multi Path Information
---------------------------------------
Host ID | %s
Controller ID | B
Port Type | FC
Initiator WWN | 1000000164s45126
Target WWN | %s
Host Port ID | 0
Link Status | Normal
=======================================
""" % (host, INITIATOR_SETTING['WWN'][0])
return out
def cli_showfcmode(self, params):
out = """/>showfcport
=========================================================================
FC Port Topology Mode
-------------------------------------------------------------------------
Controller ID Interface Module ID Port ID WWN Current Mode
-------------------------------------------------------------------------
B 1 P0 %s --
=========================================================================
-""" % INITIATOR_SETTING['WWN'][0]
return out
def cli_chglun(self, params):
if params[params.index('-lun') + 1] == VOLUME_SNAP_ID['vol']:
LUN_INFO['Owner Controller'] = 'B'
else:
CLONED_LUN_INFO['Owner Controller'] = 'B'
out = 'command operates successfully'
return out
class HuaweiDorado5100CLIResSimulator(HuaweiTCLIResSimulator):
def cli_showsys(self, params):
out = """/>showsys
=============================================================
System Information
-------------------------------------------------------------
System Name | SN_Dorado5100
Device Type | Oceanstor Dorado5100
Current System Mode | Double Controllers Normal
Mirroring Link Status | Link Up
Location |
Time | 2013-01-01 01:01:01
Product Version | V100R001C00
=============================================================
"""
return out
def cli_showlun(self, params):
if '-lun' not in params:
if LUN_INFO['ID'] is None:
out = 'command operates successfully, but no information.'
elif CLONED_LUN_INFO['ID'] is None:
out = """/>showlun
===========================================================================
LUN Information
---------------------------------------------------------------------------
ID RAIDgroup ID Status Controller Visible Capacity(MB) LUN Name..\
Strip Unit Size(KB) Lun Type
---------------------------------------------------------------------------
%s %s Normal %s %s %s 64 THICK
===========================================================================
""" % (LUN_INFO['ID'], LUN_INFO['RAID Group ID'],
LUN_INFO['Owner Controller'], str(int(LUN_INFO['Size']) * 1024),
LUN_INFO['Name'])
else:
out = """/>showlun
===========================================================================
LUN Information
---------------------------------------------------------------------------
ID RAIDgroup ID Status Controller Visible Capacity(MB) LUN Name \
Strip Unit Size(KB) Lun Type
---------------------------------------------------------------------------
%s %s Normal %s %s %s 64 THICK
%s %s Norma %s %s %s 64 THICK
===========================================================================
""" % (LUN_INFO['ID'], LUN_INFO['RAID Group ID'], LUN_INFO['Owner Controller'],
str(int(LUN_INFO['Size']) * 1024), LUN_INFO['Name'],
CLONED_LUN_INFO['ID'], CLONED_LUN_INFO['RAID Group ID'],
CLONED_LUN_INFO['Owner Controller'],
str(int(CLONED_LUN_INFO['Size']) * 1024),
CLONED_LUN_INFO['Name'])
elif params[params.index('-lun') + 1] in VOLUME_SNAP_ID.values():
out = """/>showlun
================================================
LUN Information
------------------------------------------------
ID | %s
Name | %s
LUN WWN | --
Visible Capacity | %s
RAID GROUP ID | %s
Owning Controller | %s
Workong Controller | %s
Lun Type | %s
SnapShot ID | %s
LunCopy ID | %s
================================================
""" % ((LUN_INFO['ID'], LUN_INFO['Name'], LUN_INFO['Visible Capacity'],
LUN_INFO['RAID Group ID'], LUN_INFO['Owner Controller'],
LUN_INFO['Worker Controller'], LUN_INFO['Lun Type'],
LUN_INFO['SnapShot ID'], LUN_INFO['LunCopy ID'])
if params[params.index('-lun') + 1] == VOLUME_SNAP_ID['vol'] else
(CLONED_LUN_INFO['ID'], CLONED_LUN_INFO['Name'],
CLONED_LUN_INFO['Visible Capacity'], CLONED_LUN_INFO['RAID Group ID'],
CLONED_LUN_INFO['Owner Controller'],
CLONED_LUN_INFO['Worker Controller'],
CLONED_LUN_INFO['Lun Type'], CLONED_LUN_INFO['SnapShot ID'],
CLONED_LUN_INFO['LunCopy ID']))
else:
out = 'ERROR: The object does not exist.'
return out
class HuaweiDorado2100G2CLIResSimulator(HuaweiTCLIResSimulator):
def cli_showsys(self, params):
out = """/>showsys
==========================================================================
System Information
--------------------------------------------------------------------------
System Name | SN_Dorado2100_G2
Device Type | Oceanstor Dorado2100 G2
Current System Mode | Double Controllers Normal
Mirroring Link Status | Link Up
Location |
Time | 2013-01-01 01:01:01
Product Version | V100R001C00
===========================================================================
"""
return out
def cli_createlun(self, params):
lun_type = ('THIN' if params[params.index('-type') + 1] == '2' else
'THICK')
if LUN_INFO['ID'] is None:
LUN_INFO['Name'] = self._name_translate(FAKE_VOLUME['name'])
LUN_INFO['ID'] = VOLUME_SNAP_ID['vol']
LUN_INFO['Size'] = FAKE_VOLUME['size']
LUN_INFO['Lun Type'] = lun_type
LUN_INFO['Owner Controller'] = 'A'
LUN_INFO['Worker Controller'] = 'A'
LUN_INFO['RAID Group ID'] = POOL_SETTING['ID']
FAKE_VOLUME['provider_location'] = LUN_INFO['ID']
else:
CLONED_LUN_INFO['Name'] = \
self._name_translate(FAKE_CLONED_VOLUME['name'])
CLONED_LUN_INFO['ID'] = VOLUME_SNAP_ID['vol_copy']
CLONED_LUN_INFO['Size'] = FAKE_CLONED_VOLUME['size']
CLONED_LUN_INFO['Lun Type'] = lun_type
CLONED_LUN_INFO['Owner Controller'] = 'A'
CLONED_LUN_INFO['Worker Controller'] = 'A'
CLONED_LUN_INFO['RAID Group ID'] = POOL_SETTING['ID']
CLONED_LUN_INFO['provider_location'] = CLONED_LUN_INFO['ID']
FAKE_CLONED_VOLUME['provider_location'] = CLONED_LUN_INFO['ID']
out = 'command operates successfully'
return out
def cli_showlun(self, params):
if '-lun' not in params:
if LUN_INFO['ID'] is None:
out = 'command operates successfully, but no information.'
elif CLONED_LUN_INFO['ID'] is None:
out = """/>showlun
===========================================================================
LUN Information
---------------------------------------------------------------------------
ID Status Controller Visible Capacity(MB) LUN Name Lun Type
---------------------------------------------------------------------------
%s Normal %s %s %s THICK
===========================================================================
""" % (LUN_INFO['ID'], LUN_INFO['Owner Controller'],
str(int(LUN_INFO['Size']) * 1024), LUN_INFO['Name'])
else:
out = """/>showlun
===========================================================================
LUN Information
---------------------------------------------------------------------------
ID Status Controller Visible Capacity(MB) LUN Name Lun Type
---------------------------------------------------------------------------
%s Normal %s %s %s THICK
%s Normal %s %s %s THICK
===========================================================================
""" % (LUN_INFO['ID'], LUN_INFO['Owner Controller'],
str(int(LUN_INFO['Size']) * 1024), LUN_INFO['Name'],
CLONED_LUN_INFO['ID'], CLONED_LUN_INFO['Owner Controller'],
str(int(CLONED_LUN_INFO['Size']) * 1024), CLONED_LUN_INFO['Name'])
elif params[params.index('-lun') + 1] in VOLUME_SNAP_ID.values():
out = """/>showlun
================================================
LUN Information
------------------------------------------------
ID | %s
Name | %s
LUN WWN | --
Visible Capacity | %s
RAID GROUP ID | %s
Owning Controller | %s
Workong Controller | %s
Lun Type | %s
SnapShot ID | %s
LunCopy ID | %s
================================================
""" % ((LUN_INFO['ID'], LUN_INFO['Name'], LUN_INFO['Visible Capacity'],
LUN_INFO['RAID Group ID'], LUN_INFO['Owner Controller'],
LUN_INFO['Worker Controller'], LUN_INFO['Lun Type'],
LUN_INFO['SnapShot ID'], LUN_INFO['LunCopy ID'])
if params[params.index('-lun')] == VOLUME_SNAP_ID['vol'] else
(CLONED_LUN_INFO['ID'], CLONED_LUN_INFO['Name'],
CLONED_LUN_INFO['Visible Capacity'], CLONED_LUN_INFO['RAID Group ID'],
CLONED_LUN_INFO['Owner Controller'],
CLONED_LUN_INFO['Worker Controller'],
CLONED_LUN_INFO['Lun Type'], CLONED_LUN_INFO['SnapShot ID'],
CLONED_LUN_INFO['LunCopy ID']))
else:
out = 'ERROR: The object does not exist.'
return out
class HuaweiTISCSIDriverTestCase(test.TestCase):
def __init__(self, *args, **kwargs):
super(HuaweiTISCSIDriverTestCase, self).__init__(*args, **kwargs)
def setUp(self):
super(HuaweiTISCSIDriverTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
create_fake_conf_file(self.fake_conf_file)
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.configuration.append_config_values(mox.IgnoreArg())
self.stubs.Set(time, 'sleep', Fake_sleep)
self.stubs.Set(utils, 'SSHPool', FakeSSHPool)
self.stubs.Set(ssh_common.TseriesCommon, '_change_file_mode',
Fake_change_file_mode)
self._init_driver()
def _init_driver(self):
Curr_test[0] = 'T'
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def tearDown(self):
if os.path.exists(self.fake_conf_file):
os.remove(self.fake_conf_file)
shutil.rmtree(self.tmp_dir)
super(HuaweiTISCSIDriverTestCase, self).tearDown()
def test_conf_invalid(self):
# Test config file not found
tmp_fonf_file = '/xxx/cinder_huawei_conf.xml'
tmp_configuration = mox.MockObject(conf.Configuration)
tmp_configuration.cinder_huawei_conf_file = tmp_fonf_file
tmp_configuration.append_config_values(mox.IgnoreArg())
self.assertRaises(IOError,
HuaweiVolumeDriver,
configuration=tmp_configuration)
# Test Product and Protocol invalid
tmp_dict = {'Storage/Product': 'T', 'Storage/Protocol': 'iSCSI'}
for k, v in tmp_dict.items():
modify_conf(self.fake_conf_file, k, 'xx')
self.assertRaises(exception.InvalidInput,
HuaweiVolumeDriver,
configuration=self.configuration)
modify_conf(self.fake_conf_file, k, v)
# Test ctr ip, UserName and password unspecified
tmp_dict = {'Storage/ControllerIP0': '10.10.10.1',
'Storage/ControllerIP1': '10.10.10.2',
'Storage/UserName': 'admin',
'Storage/UserPassword': '123456'}
for k, v in tmp_dict.items():
modify_conf(self.fake_conf_file, k, '')
tmp_driver = HuaweiVolumeDriver(configuration=self.configuration)
self.assertRaises(exception.InvalidInput,
tmp_driver.do_setup, None)
modify_conf(self.fake_conf_file, k, v)
# Test StoragePool unspecified
modify_conf(self.fake_conf_file, 'LUN/StoragePool', '', attrib='Name')
tmp_driver = HuaweiVolumeDriver(configuration=self.configuration)
self.assertRaises(exception.InvalidInput,
tmp_driver.do_setup, None)
modify_conf(self.fake_conf_file, 'LUN/StoragePool', 'RAID_001',
attrib='Name')
# Test LUN type invalid
modify_conf(self.fake_conf_file, 'LUN/LUNType', 'thick')
tmp_driver = HuaweiVolumeDriver(configuration=self.configuration)
tmp_driver.do_setup(None)
self.assertRaises(exception.InvalidInput,
tmp_driver.create_volume, FAKE_VOLUME)
modify_conf(self.fake_conf_file, 'LUN/LUNType', 'Thick')
# Test OSType invalid
modify_conf(self.fake_conf_file, 'Host', 'invalid_type',
attrib='OSType')
tmp_driver = HuaweiVolumeDriver(configuration=self.configuration)
self.assertRaises(exception.InvalidInput,
tmp_driver.do_setup, None)
modify_conf(self.fake_conf_file, 'Host', 'Linux', attrib='OSType')
# Test TargetIP not found
modify_conf(self.fake_conf_file, 'iSCSI/DefaultTargetIP', '')
modify_conf(self.fake_conf_file, 'iSCSI/Initiator', '', attrib='Name')
tmp_driver = HuaweiVolumeDriver(configuration=self.configuration)
tmp_driver.do_setup(None)
tmp_driver.create_volume(FAKE_VOLUME)
self.assertRaises(exception.InvalidInput,
tmp_driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
tmp_driver.delete_volume(FAKE_VOLUME)
modify_conf(self.fake_conf_file, 'iSCSI/DefaultTargetIP',
'192.168.100.1')
modify_conf(self.fake_conf_file, 'iSCSI/Initiator',
'iqn.1993-08.debian:01:ec2bff7ac3a3', attrib='Name')
def test_volume_type(self):
ctxt = context.get_admin_context()
extra_specs = {'drivers:LUNType': 'Thin'}
type_ref = volume_types.create(ctxt, 'THIN', extra_specs)
FAKE_VOLUME['volume_type_id'] = type_ref['id']
self.driver.create_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO["ID"], VOLUME_SNAP_ID['vol'])
self.assertEqual(LUN_INFO['Lun Type'], 'THIN')
self.driver.delete_volume(FAKE_VOLUME)
FAKE_VOLUME['volume_type_id'] = None
# Test volume type invalid
extra_specs = {'drivers:InvalidLUNType': 'Thin'}
type_ref = volume_types.create(ctxt, 'Invalid_THIN', extra_specs)
FAKE_VOLUME['volume_type_id'] = type_ref['id']
self.driver.create_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO["ID"], VOLUME_SNAP_ID['vol'])
self.assertNotEqual(LUN_INFO['Lun Type'], 'THIN')
self.driver.delete_volume(FAKE_VOLUME)
FAKE_VOLUME['volume_type_id'] = None
def test_create_delete_volume(self):
# Test create lun cli exception
set_error_flg('createlun')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_volume, FAKE_VOLUME)
ret = self.driver.create_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
self.assertEqual(ret['provider_location'], LUN_INFO['ID'])
# Test delete lun cli exception
set_error_flg('dellun')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.delete_volume, FAKE_VOLUME)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertEqual(FAKE_VOLUME['provider_location'], None)
def test_create_delete_cloned_volume(self):
# Test no source volume
self.assertRaises(exception.VolumeNotFound,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.driver.create_volume(FAKE_VOLUME)
# Test create luncopy failed
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
set_error_flg('createluncopy')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
# Test start luncopy failed
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
set_error_flg('chgluncopystatus')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
# Test luncopy status abnormal
LUNCOPY_SETTING['Status'] = 'Disable'
self.assertEqual(LUN_INFO['ID'], '0')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
LUNCOPY_SETTING['Status'] = 'Normal'
# Test delete luncopy failed
set_error_flg('delluncopy')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
# need to clean up LUNCopy
LUNCOPY_INFO['Name'] = None
LUNCOPY_INFO['ID'] = None
LUNCOPY_INFO['Type'] = None
LUNCOPY_INFO['State'] = None
LUNCOPY_INFO['Status'] = None
# Test normal create and delete cloned volume
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
ret = self.driver.create_cloned_volume(FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(FAKE_CLONED_VOLUME['provider_location'], None)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
def test_create_delete_snapshot(self):
# Test no resource pool
RESPOOL_A_SIM['Valid Size'] = '0'
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_snapshot, FAKE_SNAPSHOT)
RESPOOL_A_SIM['Valid Size'] = '5120'
# Test no source volume
self.assertRaises(exception.VolumeNotFound,
self.driver.create_snapshot, FAKE_SNAPSHOT)
# Test create snapshot cli exception
self.driver.create_volume(FAKE_VOLUME)
set_error_flg('createsnapshot')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_snapshot,
FAKE_SNAPSHOT)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
# Test active snapshot failed
set_error_flg('actvsnapshot')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_snapshot,
FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.assertEqual(SNAPSHOT_INFO['Status'], None)
# Test disable snapshot failed
set_error_flg('disablesnapshot')
self.driver.create_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], VOLUME_SNAP_ID['snap'])
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.delete_snapshot,
FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['Status'], 'Active')
# Test delsnapshot failed
set_error_flg('delsnapshot')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.delete_snapshot,
FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['Status'], 'Disable')
self.driver.delete_snapshot(FAKE_SNAPSHOT)
# Test normal create and delete snapshot
self.driver.create_volume(FAKE_VOLUME)
ret = self.driver.create_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], VOLUME_SNAP_ID['snap'])
self.assertEqual(SNAPSHOT_INFO['Status'], 'Active')
self.assertEqual(ret['provider_location'], SNAPSHOT_INFO['ID'])
self.driver.delete_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.assertEqual(SNAPSHOT_INFO['Status'], None)
def test_create_delete_snapshot_volume(self):
# Test no source snapshot
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_volume_from_snapshot,
FAKE_CLONED_VOLUME, FAKE_SNAPSHOT)
# Test normal create and delete snapshot volume
self.driver.create_volume(FAKE_VOLUME)
self.driver.create_snapshot(FAKE_SNAPSHOT)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
self.assertEqual(SNAPSHOT_INFO['ID'], VOLUME_SNAP_ID['snap'])
ret = self.driver.create_volume_from_snapshot(FAKE_CLONED_VOLUME,
FAKE_SNAPSHOT)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_snapshot(FAKE_SNAPSHOT)
self.driver.delete_volume(FAKE_VOLUME)
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
def test_initialize_connection(self):
# Test can not get iscsi iqn
set_error_flg('showiscsitgtname')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test failed to get iSCSI port info
set_error_flg('showiscsiip')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test create hostgroup failed
set_error_flg('createhostgroup')
MAP_INFO['Host Group ID'] = None
MAP_INFO['Host Group Name'] = None
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test create host failed
set_error_flg('addhost')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test add iSCSI initiator failed
set_error_flg('addiscsiini')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test add hostport failed
set_error_flg('addhostport')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test no volume
FAKE_VOLUME['provider_location'] = '100'
self.assertRaises(exception.VolumeNotFound,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
FAKE_VOLUME['provider_location'] = None
# Test map volume failed
self.driver.create_volume(FAKE_VOLUME)
set_error_flg('addhostmap')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.initialize_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test normal initialize connection
self.assertEqual(FAKE_VOLUME['provider_location'],
VOLUME_SNAP_ID['vol'])
self.assertEqual(LUN_INFO['Owner Controller'], 'A')
ret = self.driver.initialize_connection(FAKE_VOLUME, FAKE_CONNECTOR)
iscsi_propers = ret['data']
self.assertEqual(iscsi_propers['target_iqn'],
INITIATOR_SETTING['TargetIQN-form'])
self.assertEqual(iscsi_propers['target_portal'],
INITIATOR_SETTING['Initiator TargetIP'] + ':3260')
self.assertEqual(MAP_INFO["DEV LUN ID"], LUN_INFO['ID'])
self.assertEqual(MAP_INFO["INI Port Info"],
FAKE_CONNECTOR['initiator'])
self.assertEqual(LUN_INFO['Owner Controller'], 'B')
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
def test_terminate_connection(self):
# Test no host was found
self.assertRaises(exception.HostNotFound,
self.driver.terminate_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test no volume was found
self.driver .create_volume(FAKE_VOLUME)
self.driver.initialize_connection(FAKE_VOLUME, FAKE_CONNECTOR)
FAKE_VOLUME['provider_location'] = None
self.assertRaises(exception.VolumeNotFound,
self.driver.terminate_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
FAKE_VOLUME['provider_location'] = LUN_INFO['ID']
# Test delete map failed
set_error_flg('delhostmap')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.terminate_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Delete hostport failed
set_error_flg('delhostport')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.terminate_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test delete initiator failed
set_error_flg('deliscsiini')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.terminate_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test delete host failed
set_error_flg('delhost')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.terminate_connection,
FAKE_VOLUME, FAKE_CONNECTOR)
# Test normal terminate connection
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
self.driver.initialize_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.assertEqual(MAP_INFO["DEV LUN ID"], None)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
def test_get_volume_stats(self):
stats = self.driver.get_volume_stats(True)
free_capacity = float(POOL_SETTING['Free Capacity']) / 1024
self.assertEqual(stats['free_capacity_gb'], free_capacity)
self.assertEqual(stats['storage_protocol'], 'iSCSI')
class HuaweiTFCDriverTestCase(test.TestCase):
def __init__(self, *args, **kwargs):
super(HuaweiTFCDriverTestCase, self).__init__(*args, **kwargs)
def setUp(self):
super(HuaweiTFCDriverTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
create_fake_conf_file(self.fake_conf_file)
modify_conf(self.fake_conf_file, 'Storage/Protocol', 'FC')
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.configuration.append_config_values(mox.IgnoreArg())
self.stubs.Set(time, 'sleep', Fake_sleep)
self.stubs.Set(utils, 'SSHPool', FakeSSHPool)
self.stubs.Set(ssh_common.TseriesCommon, '_change_file_mode',
Fake_change_file_mode)
self._init_driver()
def _init_driver(self):
Curr_test[0] = 'T'
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def tearDown(self):
if os.path.exists(self.fake_conf_file):
os.remove(self.fake_conf_file)
shutil.rmtree(self.tmp_dir)
super(HuaweiTFCDriverTestCase, self).tearDown()
def test_validate_connector_failed(self):
invalid_connector = {'host': 'testhost'}
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.validate_connector,
invalid_connector)
def test_create_delete_volume(self):
self.driver.create_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
def test_create_delete_snapshot(self):
self.driver.create_volume(FAKE_VOLUME)
self.driver.create_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], VOLUME_SNAP_ID['snap'])
self.driver.delete_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
def test_create_cloned_volume(self):
self.driver.create_volume(FAKE_VOLUME)
ret = self.driver.create_cloned_volume(FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(LUN_INFO['ID'], None)
def test_create_snapshot_volume(self):
self.driver.create_volume(FAKE_VOLUME)
self.driver.create_snapshot(FAKE_SNAPSHOT)
ret = self.driver.create_volume_from_snapshot(FAKE_CLONED_VOLUME,
FAKE_SNAPSHOT)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(LUN_INFO['ID'], None)
def test_initialize_terminitat_connection(self):
self.driver.create_volume(FAKE_VOLUME)
ret = self.driver.initialize_connection(FAKE_VOLUME, FAKE_CONNECTOR)
fc_properties = ret['data']
self.assertEqual(fc_properties['target_wwn'],
INITIATOR_SETTING['WWN'])
self.assertEqual(MAP_INFO["DEV LUN ID"], LUN_INFO['ID'])
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.assertEqual(MAP_INFO["DEV LUN ID"], None)
self.assertEqual(MAP_INFO["Host LUN ID"], None)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
def _test_get_volume_stats(self):
stats = self.driver.get_volume_stats(True)
fakecapacity = float(POOL_SETTING['Free Capacity']) / 1024
self.assertEqual(stats['free_capacity_gb'], fakecapacity)
self.assertEqual(stats['storage_protocol'], 'FC')
class HuaweiDorado5100FCDriverTestCase(HuaweiTFCDriverTestCase):
def __init__(self, *args, **kwargs):
super(HuaweiDorado5100FCDriverTestCase, self).__init__(*args, **kwargs)
def setUp(self):
super(HuaweiDorado5100FCDriverTestCase, self).setUp()
def _init_driver(self):
Curr_test[0] = 'Dorado5100'
modify_conf(self.fake_conf_file, 'Storage/Product', 'Dorado')
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def test_create_cloned_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
def test_create_snapshot_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_volume_from_snapshot,
FAKE_CLONED_VOLUME, FAKE_SNAPSHOT)
class HuaweiDorado2100G2FCDriverTestCase(HuaweiTFCDriverTestCase):
def __init__(self, *args, **kwargs):
super(HuaweiDorado2100G2FCDriverTestCase, self).__init__(*args,
**kwargs)
def setUp(self):
super(HuaweiDorado2100G2FCDriverTestCase, self).setUp()
def _init_driver(self):
Curr_test[0] = 'Dorado2100G2'
modify_conf(self.fake_conf_file, 'Storage/Product', 'Dorado')
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def test_create_cloned_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
def test_create_delete_snapshot(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_snapshot, FAKE_SNAPSHOT)
def test_create_snapshot_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_volume_from_snapshot,
FAKE_CLONED_VOLUME, FAKE_SNAPSHOT)
class HuaweiDorado5100ISCSIDriverTestCase(HuaweiTISCSIDriverTestCase):
def __init__(self, *args, **kwargs):
super(HuaweiDorado5100ISCSIDriverTestCase, self).__init__(*args,
**kwargs)
def setUp(self):
super(HuaweiDorado5100ISCSIDriverTestCase, self).setUp()
def _init_driver(self):
Curr_test[0] = 'Dorado5100'
modify_conf(self.fake_conf_file, 'Storage/Product', 'Dorado')
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def test_create_delete_cloned_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
def test_create_delete_snapshot_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_volume_from_snapshot,
FAKE_CLONED_VOLUME, FAKE_SNAPSHOT)
def test_volume_type(self):
pass
class HuaweiDorado2100G2ISCSIDriverTestCase(HuaweiTISCSIDriverTestCase):
def __init__(self, *args, **kwargs):
super(HuaweiDorado2100G2ISCSIDriverTestCase, self).__init__(*args,
**kwargs)
def setUp(self):
super(HuaweiDorado2100G2ISCSIDriverTestCase, self).setUp()
def _init_driver(self):
Curr_test[0] = 'Dorado2100G2'
modify_conf(self.fake_conf_file, 'Storage/Product', 'Dorado')
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def test_conf_invalid(self):
pass
def test_create_delete_cloned_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
def test_create_delete_snapshot(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_snapshot, FAKE_SNAPSHOT)
def test_create_delete_snapshot_volume(self):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_volume_from_snapshot,
FAKE_CLONED_VOLUME, FAKE_SNAPSHOT)
def test_initialize_connection(self):
self.driver.create_volume(FAKE_VOLUME)
ret = self.driver.initialize_connection(FAKE_VOLUME, FAKE_CONNECTOR)
iscsi_propers = ret['data']
self.assertEqual(iscsi_propers['target_iqn'],
INITIATOR_SETTING['TargetIQN-form'])
self.assertEqual(iscsi_propers['target_portal'],
INITIATOR_SETTING['Initiator TargetIP'] + ':3260')
self.assertEqual(MAP_INFO["DEV LUN ID"], LUN_INFO['ID'])
self.assertEqual(MAP_INFO["INI Port Info"],
FAKE_CONNECTOR['initiator'])
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
class SSHMethodTestCase(test.TestCase):
def __init__(self, *args, **kwargs):
super(SSHMethodTestCase, self).__init__(*args, **kwargs)
def setUp(self):
super(SSHMethodTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
create_fake_conf_file(self.fake_conf_file)
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.configuration.append_config_values(mox.IgnoreArg())
self.stubs.Set(time, 'sleep', Fake_sleep)
self.stubs.Set(utils, 'SSHPool', FakeSSHPool)
self.stubs.Set(ssh_common.TseriesCommon, '_change_file_mode',
Fake_change_file_mode)
Curr_test[0] = 'T'
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
def tearDown(self):
if os.path.exists(self.fake_conf_file):
os.remove(self.fake_conf_file)
shutil.rmtree(self.tmp_dir)
super(SSHMethodTestCase, self).tearDown()
def test_reach_max_connection_limit(self):
self.stubs.Set(FakeChannel, 'recv', self._fake_recv1)
self.assertRaises(exception.CinderException,
self.driver.create_volume, FAKE_VOLUME)
def test_socket_timeout(self):
self.stubs.Set(FakeChannel, 'recv', self._fake_recv2)
self.assertRaises(socket.timeout,
self.driver.create_volume, FAKE_VOLUME)
def _fake_recv1(self, nbytes):
return "No response message"
def _fake_recv2(self, nBytes):
raise socket.timeout()
class HuaweiUtilsTestCase(test.TestCase):
def __init__(self, *args, **kwargs):
super(HuaweiUtilsTestCase, self).__init__(*args, **kwargs)
def setUp(self):
super(HuaweiUtilsTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
create_fake_conf_file(self.fake_conf_file)
def tearDown(self):
if os.path.exists(self.fake_conf_file):
os.remove(self.fake_conf_file)
shutil.rmtree(self.tmp_dir)
super(HuaweiUtilsTestCase, self).tearDown()
def test_parse_xml_file_ioerror(self):
tmp_fonf_file = '/xxx/cinder_huawei_conf.xml'
self.assertRaises(IOError, huawei_utils.parse_xml_file, tmp_fonf_file)
def test_is_xml_item_exist(self):
root = huawei_utils.parse_xml_file(self.fake_conf_file)
res = huawei_utils.is_xml_item_exist(root, 'Storage/UserName')
self.assertTrue(res)
res = huawei_utils.is_xml_item_exist(root, 'xxx')
self.assertFalse(res)
res = huawei_utils.is_xml_item_exist(root, 'LUN/StoragePool', 'Name')
self.assertTrue(res)
res = huawei_utils.is_xml_item_exist(root, 'LUN/StoragePool', 'xxx')
self.assertFalse(res)
def test_is_xml_item_valid(self):
root = huawei_utils.parse_xml_file(self.fake_conf_file)
res = huawei_utils.is_xml_item_valid(root, 'LUN/LUNType',
['Thin', 'Thick'])
self.assertTrue(res)
res = huawei_utils.is_xml_item_valid(root, 'LUN/LUNType', ['test'])
self.assertFalse(res)
res = huawei_utils.is_xml_item_valid(root, 'Host',
['Linux', 'Windows'], 'OSType')
self.assertTrue(res)
res = huawei_utils.is_xml_item_valid(root, 'Host', ['test'], 'OSType')
self.assertFalse(res)
def test_get_conf_host_os_type(self):
# Default os is Linux
res = huawei_utils.get_conf_host_os_type('10.10.10.1',
self.fake_conf_file)
self.assertEqual(res, '0')
modify_conf(self.fake_conf_file, 'Host', 'Windows', 'OSType')
res = huawei_utils.get_conf_host_os_type(FAKE_CONNECTOR['ip'],
self.fake_conf_file)
self.assertEqual(res, '1')
View Git Blame Copy Permalink