221 lines
7.7 KiB
Python
221 lines
7.7 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2012 Pedro Navarro Perez
|
|
# 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 Windows Server 2012 OpenStack Manila volume driver
|
|
"""
|
|
import sys
|
|
|
|
import manila.flags
|
|
from manila.tests.windows import basetestcase
|
|
from manila.tests.windows import db_fakes
|
|
from manila.tests.windows import windowsutils
|
|
from manila.volume.drivers import windows
|
|
|
|
FLAGS = manila.flags.FLAGS
|
|
|
|
|
|
class TestWindowsDriver(basetestcase.BaseTestCase):
|
|
|
|
def __init__(self, method):
|
|
super(TestWindowsDriver, self).__init__(method)
|
|
|
|
def setUp(self):
|
|
super(TestWindowsDriver, self).setUp()
|
|
self.flags(
|
|
windows_iscsi_lun_path='C:\iSCSIVirtualDisks',
|
|
)
|
|
self._volume_data = None
|
|
self._volume_data_2 = None
|
|
self._snapshot_data = None
|
|
self._connector_data = None
|
|
self._volume_id = '10958016-e196-42e3-9e7f-5d8927ae3099'
|
|
self._volume_id_2 = '20958016-e196-42e3-9e7f-5d8927ae3098'
|
|
self._snapshot_id = '30958016-e196-42e3-9e7f-5d8927ae3097'
|
|
self._iqn = "iqn.1991-05.com.microsoft:dell1160dsy"
|
|
|
|
self._setup_stubs()
|
|
|
|
self._drv = windows.WindowsDriver()
|
|
self._drv.do_setup({})
|
|
self._wutils = windowsutils.WindowsUtils()
|
|
|
|
def _setup_stubs(self):
|
|
|
|
# Modules to mock
|
|
modules_to_mock = [
|
|
'wmi',
|
|
'os',
|
|
'subprocess',
|
|
'multiprocessing'
|
|
]
|
|
|
|
modules_to_test = [
|
|
windows,
|
|
windowsutils,
|
|
sys.modules[__name__]
|
|
]
|
|
|
|
self._inject_mocks_in_modules(modules_to_mock, modules_to_test)
|
|
|
|
def tearDown(self):
|
|
try:
|
|
if (self._volume_data_2 and
|
|
self._wutils.volume_exists(self._volume_data_2['name'])):
|
|
self._wutils.delete_volume(self._volume_data_2['name'])
|
|
|
|
if (self._volume_data and
|
|
self._wutils.volume_exists(
|
|
self._volume_data['name'])):
|
|
self._wutils.delete_volume(self._volume_data['name'])
|
|
if (self._snapshot_data and
|
|
self._wutils.snapshot_exists(
|
|
self._snapshot_data['name'])):
|
|
self._wutils.delete_snapshot(self._snapshot_data['name'])
|
|
if (self._connector_data and
|
|
self._wutils.initiator_id_exists(
|
|
"%s%s" % (FLAGS.iscsi_target_prefix,
|
|
self._volume_data['name']),
|
|
self._connector_data['initiator'])):
|
|
target_name = "%s%s" % (FLAGS.iscsi_target_prefix,
|
|
self._volume_data['name'])
|
|
initiator_name = self._connector_data['initiator']
|
|
self._wutils.delete_initiator_id(target_name, initiator_name)
|
|
if (self._volume_data and
|
|
self._wutils.export_exists("%s%s" %
|
|
(FLAGS.iscsi_target_prefix,
|
|
self._volume_data['name']))):
|
|
self._wutils.delete_export(
|
|
"%s%s" % (FLAGS.iscsi_target_prefix,
|
|
self._volume_data['name']))
|
|
|
|
finally:
|
|
super(TestWindowsDriver, self).tearDown()
|
|
|
|
def test_check_for_setup_errors(self):
|
|
self._drv.check_for_setup_error()
|
|
|
|
def test_create_volume(self):
|
|
self._create_volume()
|
|
|
|
wt_disks = self._wutils.find_vhd_by_name(self._volume_data['name'])
|
|
self.assertEquals(len(wt_disks), 1)
|
|
|
|
def _create_volume(self):
|
|
self._volume_data = db_fakes.get_fake_volume_info(self._volume_id)
|
|
self._drv.create_volume(self._volume_data)
|
|
|
|
def test_delete_volume(self):
|
|
self._create_volume()
|
|
|
|
self._drv.delete_volume(self._volume_data)
|
|
|
|
wt_disks = self._wutils.find_vhd_by_name(self._volume_data['name'])
|
|
self.assertEquals(len(wt_disks), 0)
|
|
|
|
def test_create_snapshot(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
|
|
wt_disks = self._wutils.find_vhd_by_name(self._volume_data['name'])
|
|
self.assertEquals(len(wt_disks), 1)
|
|
#Create a snapshot from the previous volume
|
|
self._create_snapshot()
|
|
|
|
snapshot_name = self._snapshot_data['name']
|
|
wt_snapshots = self._wutils.find_snapshot_by_name(snapshot_name)
|
|
self.assertEquals(len(wt_snapshots), 1)
|
|
|
|
def _create_snapshot(self):
|
|
volume_name = self._volume_data['name']
|
|
snapshot_id = self._snapshot_id
|
|
self._snapshot_data = db_fakes.get_fake_snapshot_info(volume_name,
|
|
snapshot_id)
|
|
self._drv.create_snapshot(self._snapshot_data)
|
|
|
|
def test_create_volume_from_snapshot(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
#Create a snapshot from the previous volume
|
|
self._create_snapshot()
|
|
|
|
self._volume_data_2 = db_fakes.get_fake_volume_info(self._volume_id_2)
|
|
|
|
self._drv.create_volume_from_snapshot(self._volume_data_2,
|
|
self._snapshot_data)
|
|
|
|
wt_disks = self._wutils.find_vhd_by_name(self._volume_data_2['name'])
|
|
self.assertEquals(len(wt_disks), 1)
|
|
|
|
def test_delete_snapshot(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
#Create a snapshot from the previous volume
|
|
self._create_snapshot()
|
|
|
|
self._drv.delete_snapshot(self._snapshot_data)
|
|
|
|
snapshot_name = self._snapshot_data['name']
|
|
wt_snapshots = self._wutils.find_snapshot_by_name(snapshot_name)
|
|
self.assertEquals(len(wt_snapshots), 0)
|
|
|
|
def test_create_export(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
|
|
retval = self._drv.create_export({}, self._volume_data)
|
|
|
|
volume_name = self._volume_data['name']
|
|
self.assertEquals(
|
|
retval,
|
|
{'provider_location': "%s%s" % (FLAGS.iscsi_target_prefix,
|
|
volume_name)})
|
|
|
|
def test_initialize_connection(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
|
|
self._drv.create_export({}, self._volume_data)
|
|
|
|
self._connector_data = db_fakes.get_fake_connector_info(self._iqn)
|
|
|
|
init_data = self._drv.initialize_connection(self._volume_data,
|
|
self._connector_data)
|
|
target_name = self._volume_data['provider_location']
|
|
initiator_name = self._connector_data['initiator']
|
|
|
|
wt_initiator_ids = self._wutils.find_initiator_ids(target_name,
|
|
initiator_name)
|
|
self.assertEquals(len(wt_initiator_ids), 1)
|
|
|
|
properties = init_data['data']
|
|
self.assertNotEqual(properties['target_iqn'], None)
|
|
|
|
def test_ensure_export(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
|
|
self._drv.ensure_export({}, self._volume_data)
|
|
|
|
def test_remove_export(self):
|
|
#Create a volume
|
|
self._create_volume()
|
|
|
|
self._drv.create_export({}, self._volume_data)
|
|
|
|
self._drv.remove_export({}, self._volume_data)
|