Shared filesystem management project for OpenStack.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

6792 lines
248 KiB

# Copyright (c) 2014 Alex Meade. All rights reserved.
# Copyright (c) 2015 Clinton Knight. All rights reserved.
# Copyright (c) 2015 Tom Barron. 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.
import copy
import hashlib
import time
from unittest import mock
import ddt
from oslo_log import log
import six
from manila import exception
from manila.share.drivers.netapp.dataontap.client import api as netapp_api
from manila.share.drivers.netapp.dataontap.client import client_base
from manila.share.drivers.netapp.dataontap.client import client_cmode
from manila import test
from manila.tests.share.drivers.netapp.dataontap.client import fakes as fake
@ddt.ddt
class NetAppClientCmodeTestCase(test.TestCase):
def setUp(self):
super(NetAppClientCmodeTestCase, self).setUp()
# Mock loggers as themselves to allow logger arg validation
mock_logger = log.getLogger('mock_logger')
self.mock_object(client_cmode.LOG,
'error',
mock.Mock(side_effect=mock_logger.error))
self.mock_object(client_cmode.LOG,
'warning',
mock.Mock(side_effect=mock_logger.warning))
self.mock_object(client_cmode.LOG,
'debug',
mock.Mock(side_effect=mock_logger.debug))
self.mock_object(client_base.NetAppBaseClient,
'get_ontapi_version',
mock.Mock(return_value=(1, 20)))
self.mock_object(client_base.NetAppBaseClient,
'get_system_version',
mock.Mock(return_value={
'version-tuple': (8, 3, 0),
'version': fake.VERSION,
}))
self.client = client_cmode.NetAppCmodeClient(**fake.CONNECTION_INFO)
self.client.connection = mock.MagicMock()
self.vserver_client = client_cmode.NetAppCmodeClient(
**fake.CONNECTION_INFO)
self.vserver_client.set_vserver(fake.VSERVER_NAME)
self.vserver_client.connection = mock.MagicMock()
def _mock_api_error(self, code='fake', message='fake'):
return mock.Mock(side_effect=netapp_api.NaApiError(code=code,
message=message))
def test_init_features_ontapi_1_21(self):
self.mock_object(client_base.NetAppBaseClient,
'get_ontapi_version',
mock.Mock(return_value=(1, 21)))
self.client._init_features()
self.assertFalse(self.client.features.BROADCAST_DOMAINS)
self.assertFalse(self.client.features.IPSPACES)
self.assertFalse(self.client.features.SUBNETS)
self.assertFalse(self.client.features.FLEXVOL_ENCRYPTION)
@ddt.data((1, 30), (1, 40), (2, 0))
def test_init_features_ontapi_1_30(self, ontapi_version):
self.mock_object(client_base.NetAppBaseClient,
'get_ontapi_version',
mock.Mock(return_value=ontapi_version))
self.client._init_features()
self.assertTrue(self.client.features.BROADCAST_DOMAINS)
self.assertTrue(self.client.features.IPSPACES)
self.assertTrue(self.client.features.SUBNETS)
@ddt.data((1, 110), (2, 0))
def test_init_features_ontap_1_110(self, ontapi_version):
self.mock_object(client_base.NetAppBaseClient,
'get_ontapi_version',
mock.Mock(return_value=ontapi_version))
self.client._init_features()
self.assertTrue(self.client.features.BROADCAST_DOMAINS)
self.assertTrue(self.client.features.IPSPACES)
self.assertTrue(self.client.features.SUBNETS)
self.assertTrue(self.client.features.FLEXVOL_ENCRYPTION)
@ddt.data(((9, 1, 0), fake.VERSION_NO_DARE), ((8, 3, 2), fake.VERSION))
@ddt.unpack
def test_is_nve_supported_unsupported_release_or_platform(self, gen, ver):
system_version = {'version-tuple': gen, 'version': ver}
self.mock_object(client_base.NetAppBaseClient,
'get_system_version',
mock.Mock(return_value=system_version))
self.mock_object(self.client,
'get_security_key_manager_nve_support',
mock.Mock(return_value=True))
self.mock_object(self.client,
'list_cluster_nodes',
mock.Mock(return_value=fake.NODE_NAMES))
result = self.client.is_nve_supported()
self.assertFalse(result)
def test_is_nve_supported_valid_platform_and_supported_release(self):
system_version = {
'version-tuple': (9, 1, 0),
'version': fake.VERSION,
}
self.mock_object(client_base.NetAppBaseClient,
'get_system_version',
mock.Mock(return_value=system_version))
self.mock_object(self.client,
'get_security_key_manager_nve_support',
mock.Mock(return_value=True))
self.mock_object(self.client,
'list_cluster_nodes',
mock.Mock(return_value=fake.NODE_NAMES))
result = self.client.is_nve_supported()
self.assertTrue(result)
def test_is_nve_supported_key_manager_not_enabled(self):
system_version = {
'version-tuple': (9, 1, 0),
'version': fake.VERSION,
}
self.mock_object(client_base.NetAppBaseClient,
'get_system_version',
mock.Mock(return_value=system_version))
self.mock_object(self.client,
'get_security_key_manager_nve_support',
mock.Mock(return_value=False))
self.mock_object(self.client,
'list_cluster_nodes',
mock.Mock(return_value=fake.NODE_NAMES))
result = self.client.is_nve_supported()
self.assertFalse(result)
def test_get_security_key_manager_nve_support_enabled(self):
api_response = netapp_api.NaElement(
fake.SECUTITY_KEY_MANAGER_NVE_SUPPORT_RESPONSE_TRUE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.get_security_key_manager_nve_support(
fake.NODE_NAME)
self.assertTrue(result)
api_args = {'node': fake.NODE_NAME}
self.client.send_request.assert_has_calls([
mock.call('security-key-manager-volume-encryption-supported',
api_args)])
def test_get_security_key_manager_nve_support_disabled(self):
api_response = netapp_api.NaElement(
fake.SECUTITY_KEY_MANAGER_NVE_SUPPORT_RESPONSE_FALSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.get_security_key_manager_nve_support(
fake.NODE_NAME)
self.assertFalse(result)
api_args = {'node': fake.NODE_NAME}
self.client.send_request.assert_has_calls([
mock.call('security-key-manager-volume-encryption-supported',
api_args)])
def test_get_security_key_manager_nve_support_disabled_no_license(self):
self.mock_object(self.client,
'send_request',
self._mock_api_error())
result = self.client.get_security_key_manager_nve_support(
fake.NODE_NAME)
self.assertFalse(result)
api_args = {'node': fake.NODE_NAME}
self.client.send_request.assert_has_calls([
mock.call('security-key-manager-volume-encryption-supported',
api_args)])
@ddt.data((True, True, True), (False, None, False))
@ddt.unpack
def test_send_volume_move_request_success(self, validation_only,
encrypt_dst, fv_encryption):
self.mock_object(self.client, 'features',
mock.Mock(FLEXVOL_ENCRYPTION=fv_encryption))
self.client._send_volume_move_request(fake.ROOT_VOLUME_NAME,
fake.NODE_VSERVER_NAME,
fake.SHARE_AGGREGATE_NAME,
validation_only=validation_only,
encrypt_destination=encrypt_dst)
@ddt.data((True, True, False))
@ddt.unpack
def test_send_volume_move_request_failure(self, validation_only,
encrypt_dst, fv_encrypt):
self.mock_object(self.client, 'features',
mock.Mock(FLEXVOL_ENCRYPTION=fv_encrypt))
self.assertRaises(exception.NetAppException,
self.client._send_volume_move_request,
fake.ROOT_VOLUME_NAME,
fake.NODE_VSERVER_NAME,
fake.SHARE_AGGREGATE_NAME,
validation_only=validation_only,
encrypt_destination=encrypt_dst)
def test_invoke_vserver_api(self):
self.client._invoke_vserver_api('fake-api', 'fake_vserver')
self.client.connection.set_vserver.assert_has_calls(
[mock.call('fake_vserver')])
self.client.connection.invoke_successfully.assert_has_calls(
[mock.call('fake-api', True)])
def test_has_records(self):
self.assertTrue(self.client._has_records(
netapp_api.NaElement(fake.VSERVER_GET_ITER_RESPONSE)))
def test_has_records_not_found(self):
self.assertFalse(self.client._has_records(
netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)))
@ddt.data((fake.VSERVER_GET_ITER_RESPONSE, 1),
(fake.NO_RECORDS_RESPONSE, 0))
@ddt.unpack
def test_get_record_count(self, response, expected):
api_response = netapp_api.NaElement(response)
result = self.client._get_record_count(api_response)
self.assertEqual(expected, result)
def test_get_records_count_invalid(self):
api_response = netapp_api.NaElement(
fake.INVALID_GET_ITER_RESPONSE_NO_RECORDS)
self.assertRaises(exception.NetAppException,
self.client._get_record_count,
api_response)
def test_send_iter_request(self):
api_responses = [
netapp_api.NaElement(fake.STORAGE_DISK_GET_ITER_RESPONSE_PAGE_1),
netapp_api.NaElement(fake.STORAGE_DISK_GET_ITER_RESPONSE_PAGE_2),
netapp_api.NaElement(fake.STORAGE_DISK_GET_ITER_RESPONSE_PAGE_3),
]
mock_send_request = self.mock_object(
self.client, 'send_request',
mock.Mock(side_effect=api_responses))
storage_disk_get_iter_args = {
'desired-attributes': {
'storage-disk-info': {
'disk-name': None,
}
}
}
result = self.client.send_iter_request(
'storage-disk-get-iter', api_args=storage_disk_get_iter_args,
max_page_length=10)
num_records = result.get_child_content('num-records')
self.assertEqual('28', num_records)
next_tag = result.get_child_content('next-tag')
self.assertEqual('', next_tag)
args1 = copy.deepcopy(storage_disk_get_iter_args)
args1['max-records'] = 10
args2 = copy.deepcopy(storage_disk_get_iter_args)
args2['max-records'] = 10
args2['tag'] = 'next_tag_1'
args3 = copy.deepcopy(storage_disk_get_iter_args)
args3['max-records'] = 10
args3['tag'] = 'next_tag_2'
mock_send_request.assert_has_calls([
mock.call('storage-disk-get-iter', args1),
mock.call('storage-disk-get-iter', args2),
mock.call('storage-disk-get-iter', args3),
])
def test_send_iter_request_single_page(self):
api_response = netapp_api.NaElement(
fake.STORAGE_DISK_GET_ITER_RESPONSE)
mock_send_request = self.mock_object(
self.client, 'send_request',
mock.Mock(return_value=api_response))
storage_disk_get_iter_args = {
'desired-attributes': {
'storage-disk-info': {
'disk-name': None,
}
}
}
result = self.client.send_iter_request(
'storage-disk-get-iter', api_args=storage_disk_get_iter_args,
max_page_length=10)
num_records = result.get_child_content('num-records')
self.assertEqual('4', num_records)
args = copy.deepcopy(storage_disk_get_iter_args)
args['max-records'] = 10
mock_send_request.assert_has_calls([
mock.call('storage-disk-get-iter', args),
])
def test_send_iter_request_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
mock_send_request = self.mock_object(
self.client, 'send_request',
mock.Mock(return_value=api_response))
result = self.client.send_iter_request('storage-disk-get-iter')
num_records = result.get_child_content('num-records')
self.assertEqual('0', num_records)
args = {'max-records': client_cmode.DEFAULT_MAX_PAGE_LENGTH}
mock_send_request.assert_has_calls([
mock.call('storage-disk-get-iter', args),
])
@ddt.data(fake.INVALID_GET_ITER_RESPONSE_NO_ATTRIBUTES,
fake.INVALID_GET_ITER_RESPONSE_NO_RECORDS)
def test_send_iter_request_invalid(self, fake_response):
api_response = netapp_api.NaElement(fake_response)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NetAppException,
self.client.send_iter_request,
'storage-disk-get-iter')
def test_set_vserver(self):
self.client.set_vserver(fake.VSERVER_NAME)
self.client.connection.set_vserver.assert_has_calls(
[mock.call('fake_vserver')])
def test_vserver_exists(self):
api_response = netapp_api.NaElement(fake.VSERVER_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
vserver_get_args = {
'query': {'vserver-info': {'vserver-name': fake.VSERVER_NAME}},
'desired-attributes': {'vserver-info': {'vserver-name': None}}
}
result = self.client.vserver_exists(fake.VSERVER_NAME)
self.client.send_iter_request.assert_has_calls([
mock.call('vserver-get-iter', vserver_get_args)])
self.assertTrue(result)
def test_vserver_exists_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.vserver_exists(fake.VSERVER_NAME)
self.assertFalse(result)
def test_create_vserver_no_ipspace(self):
self.mock_object(self.client, 'send_request')
vserver_create_args = {
'vserver-name': fake.VSERVER_NAME,
'root-volume-security-style': 'unix',
'root-volume-aggregate': fake.ROOT_VOLUME_AGGREGATE_NAME,
'root-volume': fake.ROOT_VOLUME_NAME,
'name-server-switch': {'nsswitch': 'file'}
}
vserver_modify_args = {
'aggr-list': [{'aggr-name': aggr_name} for aggr_name
in fake.SHARE_AGGREGATE_NAMES],
'vserver-name': fake.VSERVER_NAME
}
self.client.create_vserver(fake.VSERVER_NAME,
fake.ROOT_VOLUME_AGGREGATE_NAME,
fake.ROOT_VOLUME_NAME,
fake.SHARE_AGGREGATE_NAMES,
None)
self.client.send_request.assert_has_calls([
mock.call('vserver-create', vserver_create_args),
mock.call('vserver-modify', vserver_modify_args)])
def test_create_vserver_with_ipspace(self):
self.client.features.add_feature('IPSPACES')
self.mock_object(self.client, 'send_request')
vserver_create_args = {
'vserver-name': fake.VSERVER_NAME,
'root-volume-security-style': 'unix',
'root-volume-aggregate': fake.ROOT_VOLUME_AGGREGATE_NAME,
'root-volume': fake.ROOT_VOLUME_NAME,
'name-server-switch': {'nsswitch': 'file'},
'ipspace': fake.IPSPACE_NAME,
}
vserver_modify_args = {
'aggr-list': [{'aggr-name': aggr_name} for aggr_name
in fake.SHARE_AGGREGATE_NAMES],
'vserver-name': fake.VSERVER_NAME
}
self.client.create_vserver(fake.VSERVER_NAME,
fake.ROOT_VOLUME_AGGREGATE_NAME,
fake.ROOT_VOLUME_NAME,
fake.SHARE_AGGREGATE_NAMES,
fake.IPSPACE_NAME)
self.client.send_request.assert_has_calls([
mock.call('vserver-create', vserver_create_args),
mock.call('vserver-modify', vserver_modify_args)])
def test_create_vserver_ipspaces_not_supported(self):
self.assertRaises(exception.NetAppException,
self.client.create_vserver,
fake.VSERVER_NAME,
fake.ROOT_VOLUME_AGGREGATE_NAME,
fake.ROOT_VOLUME_NAME,
fake.SHARE_AGGREGATE_NAMES,
fake.IPSPACE_NAME)
def test_get_vserver_root_volume_name(self):
api_response = netapp_api.NaElement(
fake.VSERVER_GET_ROOT_VOLUME_NAME_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
vserver_get_args = {
'query': {'vserver-info': {'vserver-name': fake.VSERVER_NAME}},
'desired-attributes': {'vserver-info': {'root-volume': None}}
}
result = self.client.get_vserver_root_volume_name(fake.VSERVER_NAME)
self.client.send_iter_request.assert_has_calls([
mock.call('vserver-get-iter', vserver_get_args)])
self.assertEqual(fake.ROOT_VOLUME_NAME, result)
def test_get_vserver_root_volume_name_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NetAppException,
self.client.get_vserver_root_volume_name,
fake.VSERVER_NAME)
def test_get_vserver_ipspace(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(
fake.VSERVER_GET_IPSPACE_NAME_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_vserver_ipspace(fake.VSERVER_NAME)
vserver_get_iter_args = {
'query': {
'vserver-info': {
'vserver-name': fake.VSERVER_NAME,
},
},
'desired-attributes': {
'vserver-info': {
'ipspace': None,
},
},
}
self.client.send_iter_request.assert_has_calls([
mock.call('vserver-get-iter', vserver_get_iter_args)])
self.assertEqual(fake.IPSPACE_NAME, result)
def test_get_vserver_ipspace_not_supported(self):
result = self.client.get_vserver_ipspace(fake.IPSPACE_NAME)
self.assertIsNone(result)
def test_get_vserver_ipspace_not_found(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NetAppException,
self.client.get_vserver_ipspace,
fake.IPSPACE_NAME)
def test_ipspace_has_data_vservers(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(fake.VSERVER_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.ipspace_has_data_vservers(fake.IPSPACE_NAME)
vserver_get_iter_args = {
'query': {
'vserver-info': {
'ipspace': fake.IPSPACE_NAME,
'vserver-type': 'data'
},
},
'desired-attributes': {
'vserver-info': {
'vserver-name': None,
},
},
}
self.client.send_iter_request.assert_has_calls([
mock.call('vserver-get-iter', vserver_get_iter_args)])
self.assertTrue(result)
def test_ipspace_has_data_vservers_not_supported(self):
result = self.client.ipspace_has_data_vservers(fake.IPSPACE_NAME)
self.assertFalse(result)
def test_ipspace_has_data_vservers_not_found(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.ipspace_has_data_vservers(fake.IPSPACE_NAME)
self.assertFalse(result)
def test_list_vservers(self):
api_response = netapp_api.NaElement(
fake.VSERVER_DATA_LIST_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.list_vservers()
vserver_get_iter_args = {
'query': {
'vserver-info': {
'vserver-type': 'data'
}
},
'desired-attributes': {
'vserver-info': {
'vserver-name': None
}
}
}
self.client.send_iter_request.assert_has_calls([
mock.call('vserver-get-iter', vserver_get_iter_args)])
self.assertListEqual([fake.VSERVER_NAME], result)
def test_list_vservers_node_type(self):
api_response = netapp_api.NaElement(
fake.VSERVER_DATA_LIST_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.list_vservers(vserver_type='node')
vserver_get_iter_args = {
'query': {
'vserver-info': {
'vserver-type': 'node'
}
},
'desired-attributes': {
'vserver-info': {
'vserver-name': None
}
}
}
self.client.send_iter_request.assert_has_calls([
mock.call('vserver-get-iter', vserver_get_iter_args)])
self.assertListEqual([fake.VSERVER_NAME], result)
def test_list_vservers_not_found(self):
api_response = netapp_api.NaElement(
fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.list_vservers(vserver_type='data')
self.assertListEqual([], result)
def test_get_vserver_volume_count(self):
api_response = netapp_api.NaElement(fake.VOLUME_COUNT_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_vserver_volume_count()
self.assertEqual(2, result)
def test_delete_vserver_no_volumes(self):
self.mock_object(self.client,
'vserver_exists',
mock.Mock(return_value=True))
self.mock_object(self.client,
'get_vserver_root_volume_name',
mock.Mock(return_value=fake.ROOT_VOLUME_NAME))
self.mock_object(self.vserver_client,
'get_vserver_volume_count',
mock.Mock(return_value=0))
self.mock_object(self.client, '_terminate_vserver_services')
self.mock_object(self.client, 'send_request')
self.client.delete_vserver(
fake.VSERVER_NAME,
self.vserver_client,
security_services=[fake.CIFS_SECURITY_SERVICE])
self.client._terminate_vserver_services.assert_called_with(
fake.VSERVER_NAME, self.vserver_client,
[fake.CIFS_SECURITY_SERVICE])
vserver_destroy_args = {'vserver-name': fake.VSERVER_NAME}
self.client.send_request.assert_has_calls([
mock.call('vserver-destroy', vserver_destroy_args)])
def test_delete_vserver_one_volume(self):
self.mock_object(self.client,
'vserver_exists',
mock.Mock(return_value=True))
self.mock_object(self.client,
'get_vserver_root_volume_name',
mock.Mock(return_value=fake.ROOT_VOLUME_NAME))
self.mock_object(self.vserver_client,
'get_vserver_volume_count',
mock.Mock(return_value=1))
self.mock_object(self.client, 'send_request')
self.mock_object(self.vserver_client, 'offline_volume')
self.mock_object(self.vserver_client, 'delete_volume')
self.client.delete_vserver(fake.VSERVER_NAME,
self.vserver_client)
self.vserver_client.offline_volume.assert_called_with(
fake.ROOT_VOLUME_NAME)
self.vserver_client.delete_volume.assert_called_with(
fake.ROOT_VOLUME_NAME)
vserver_destroy_args = {'vserver-name': fake.VSERVER_NAME}
self.client.send_request.assert_has_calls([
mock.call('vserver-destroy', vserver_destroy_args)])
def test_delete_vserver_one_volume_already_offline(self):
self.mock_object(self.client,
'vserver_exists',
mock.Mock(return_value=True))
self.mock_object(self.client,
'get_vserver_root_volume_name',
mock.Mock(return_value=fake.ROOT_VOLUME_NAME))
self.mock_object(self.vserver_client,
'get_vserver_volume_count',
mock.Mock(return_value=1))
self.mock_object(self.client, 'send_request')
self.mock_object(self.vserver_client,
'offline_volume',
self._mock_api_error(code=netapp_api.EVOLUMEOFFLINE))
self.mock_object(self.vserver_client, 'delete_volume')
self.client.delete_vserver(fake.VSERVER_NAME,
self.vserver_client)
self.vserver_client.offline_volume.assert_called_with(
fake.ROOT_VOLUME_NAME)
self.vserver_client.delete_volume.assert_called_with(
fake.ROOT_VOLUME_NAME)
vserver_destroy_args = {'vserver-name': fake.VSERVER_NAME}
self.client.send_request.assert_has_calls([
mock.call('vserver-destroy', vserver_destroy_args)])
self.assertEqual(1, client_cmode.LOG.error.call_count)
def test_delete_vserver_one_volume_api_error(self):
self.mock_object(self.client,
'vserver_exists',
mock.Mock(return_value=True))
self.mock_object(self.client,
'get_vserver_root_volume_name',
mock.Mock(return_value=fake.ROOT_VOLUME_NAME))
self.mock_object(self.vserver_client,
'get_vserver_volume_count',
mock.Mock(return_value=1))
self.mock_object(self.client, 'send_request')
self.mock_object(self.vserver_client,
'offline_volume',
self._mock_api_error())
self.mock_object(self.vserver_client, 'delete_volume')
self.assertRaises(netapp_api.NaApiError,
self.client.delete_vserver,
fake.VSERVER_NAME,
self.vserver_client)
def test_delete_vserver_multiple_volumes(self):
self.mock_object(self.client,
'vserver_exists',
mock.Mock(return_value=True))
self.mock_object(self.client,
'get_vserver_root_volume_name',
mock.Mock(return_value=fake.ROOT_VOLUME_NAME))
self.mock_object(self.vserver_client,
'get_vserver_volume_count',
mock.Mock(return_value=2))
self.assertRaises(exception.NetAppException,
self.client.delete_vserver,
fake.VSERVER_NAME,
self.vserver_client)
def test_delete_vserver_not_found(self):
self.mock_object(self.client,
'vserver_exists',
mock.Mock(return_value=False))
self.client.delete_vserver(fake.VSERVER_NAME,
self.vserver_client)
self.assertEqual(1, client_cmode.LOG.error.call_count)
def test_terminate_vserver_services(self):
self.mock_object(self.vserver_client, 'send_request')
self.client._terminate_vserver_services(fake.VSERVER_NAME,
self.vserver_client,
[fake.CIFS_SECURITY_SERVICE])
cifs_server_delete_args = {
'admin-password': fake.CIFS_SECURITY_SERVICE['password'],
'admin-username': fake.CIFS_SECURITY_SERVICE['user'],
}
self.vserver_client.send_request.assert_has_calls([
mock.call('cifs-server-delete', cifs_server_delete_args)])
def test_terminate_vserver_services_cifs_not_found(self):
self.mock_object(self.vserver_client,
'send_request',
self._mock_api_error(
code=netapp_api.EOBJECTNOTFOUND))
self.client._terminate_vserver_services(fake.VSERVER_NAME,
self.vserver_client,
[fake.CIFS_SECURITY_SERVICE])
cifs_server_delete_args = {
'admin-password': fake.CIFS_SECURITY_SERVICE['password'],
'admin-username': fake.CIFS_SECURITY_SERVICE['user'],
}
self.vserver_client.send_request.assert_has_calls([
mock.call('cifs-server-delete', cifs_server_delete_args)])
self.assertEqual(1, client_cmode.LOG.error.call_count)
def test_terminate_vserver_services_api_error(self):
side_effects = [netapp_api.NaApiError(code='fake'), None]
self.mock_object(self.vserver_client,
'send_request',
mock.Mock(side_effect=side_effects))
self.client._terminate_vserver_services(fake.VSERVER_NAME,
self.vserver_client,
[fake.CIFS_SECURITY_SERVICE])
cifs_server_delete_args = {
'admin-password': fake.CIFS_SECURITY_SERVICE['password'],
'admin-username': fake.CIFS_SECURITY_SERVICE['user'],
}
self.vserver_client.send_request.assert_has_calls([
mock.call('cifs-server-delete', cifs_server_delete_args),
mock.call('cifs-server-delete')])
self.assertEqual(0, client_cmode.LOG.error.call_count)
def test_list_cluster_nodes(self):
api_response = netapp_api.NaElement(
fake.SYSTEM_NODE_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.list_cluster_nodes()
self.assertListEqual([fake.NODE_NAME], result)
def test_list_cluster_nodes_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.list_cluster_nodes()
self.assertListEqual([], result)
def test_list_node_data_ports(self):
self.mock_object(self.client,
'get_node_data_ports',
mock.Mock(return_value=fake.SPEED_SORTED_PORTS))
result = self.client.list_node_data_ports(fake.NODE_NAME)
self.assertSequenceEqual(fake.SPEED_SORTED_PORT_NAMES, result)
def test_get_node_data_ports(self):
api_response = netapp_api.NaElement(fake.NET_PORT_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_node_data_ports(fake.NODE_NAME)
net_port_get_iter_args = {
'query': {
'net-port-info': {
'node': fake.NODE_NAME,
'link-status': 'up',
'port-type': 'physical|if_group',
'role': 'data',
},
},
'desired-attributes': {
'net-port-info': {
'port': None,
'node': None,
'operational-speed': None,
'ifgrp-port': None,
},
},
}
self.assertSequenceEqual(fake.SPEED_SORTED_PORTS, result)
self.client.send_iter_request.assert_has_calls([
mock.call('net-port-get-iter', net_port_get_iter_args)])
def test_get_node_data_ports_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_node_data_ports(fake.NODE_NAME)
self.assertSequenceEqual([], result)
def test_sort_data_ports_by_speed(self):
result = self.client._sort_data_ports_by_speed(
fake.UNSORTED_PORTS_ALL_SPEEDS)
self.assertSequenceEqual(fake.SORTED_PORTS_ALL_SPEEDS, result)
def test_list_root_aggregates(self):
api_response = netapp_api.NaElement(
fake.AGGR_GET_ITER_ROOT_AGGR_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.list_root_aggregates()
aggr_get_iter_args = {
'desired-attributes': {
'aggr-attributes': {
'aggregate-name': None,
'aggr-raid-attributes': {
'has-local-root': None,
'has-partner-root': None,
},
},
}
}
self.assertSequenceEqual(fake.ROOT_AGGREGATE_NAMES, result)
self.client.send_iter_request.assert_has_calls([
mock.call('aggr-get-iter', aggr_get_iter_args)])
def test_list_non_root_aggregates(self):
api_response = netapp_api.NaElement(
fake.AGGR_GET_ITER_NON_ROOT_AGGR_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.list_non_root_aggregates()
aggr_get_iter_args = {
'query': {
'aggr-attributes': {
'aggr-raid-attributes': {
'has-local-root': 'false',
'has-partner-root': 'false',
}
},
},
'desired-attributes': {
'aggr-attributes': {
'aggregate-name': None,
},
},
}
self.assertSequenceEqual(fake.SHARE_AGGREGATE_NAMES, result)
self.client.send_iter_request.assert_has_calls([
mock.call('aggr-get-iter', aggr_get_iter_args)])
def test_list_aggregates(self):
api_response = netapp_api.NaElement(fake.AGGR_GET_NAMES_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client._list_aggregates()
aggr_get_iter_args = {
'desired-attributes': {
'aggr-attributes': {
'aggregate-name': None,
},
},
}
self.assertSequenceEqual(
fake.ROOT_AGGREGATE_NAMES + fake.SHARE_AGGREGATE_NAMES, result)
self.client.send_iter_request.assert_has_calls([
mock.call('aggr-get-iter', aggr_get_iter_args)])
def test_list_aggregates_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NetAppException,
self.client._list_aggregates)
def test_list_vserver_aggregates(self):
self.mock_object(self.vserver_client,
'get_vserver_aggregate_capacities',
mock.Mock(return_value=fake.VSERVER_AGGREGATES))
result = self.vserver_client.list_vserver_aggregates()
self.assertListEqual(list(fake.VSERVER_AGGREGATES.keys()), result)
def test_list_vserver_aggregates_none_found(self):
self.mock_object(self.vserver_client,
'get_vserver_aggregate_capacities',
mock.Mock(return_value={}))
result = self.vserver_client.list_vserver_aggregates()
self.assertListEqual([], result)
@ddt.data((True, True), (True, False), (False, True), (False, False))
@ddt.unpack
def test_create_network_interface(self, broadcast_domains_supported,
use_vlans):
self.client.features.add_feature('BROADCAST_DOMAINS',
broadcast_domains_supported)
self.mock_object(self.client, '_ensure_broadcast_domain_for_port')
self.mock_object(self.client, '_create_vlan')
self.mock_object(self.client, 'send_request')
lif_create_args = {
'address': fake.IP_ADDRESS,
'administrative-status': 'up',
'data-protocols': [
{'data-protocol': 'nfs'},
{'data-protocol': 'cifs'}
],
'home-node': fake.NODE_NAME,
'home-port': fake.VLAN_PORT if use_vlans else fake.PORT,
'netmask': fake.NETMASK,
'interface-name': fake.LIF_NAME,
'role': 'data',
'vserver': fake.VSERVER_NAME,
}
self.client.create_network_interface(fake.IP_ADDRESS,
fake.NETMASK,
fake.VLAN if use_vlans else None,
fake.NODE_NAME,
fake.PORT,
fake.VSERVER_NAME,
fake.LIF_NAME,
fake.IPSPACE_NAME,
fake.MTU)
if use_vlans:
self.client._create_vlan.assert_called_with(
fake.NODE_NAME, fake.PORT, fake.VLAN)
else:
self.assertFalse(self.client._create_vlan.called)
if broadcast_domains_supported:
self.client._ensure_broadcast_domain_for_port.assert_called_with(
fake.NODE_NAME, fake.VLAN_PORT if use_vlans else fake.PORT,
fake.MTU, ipspace=fake.IPSPACE_NAME)
else:
self.assertFalse(
self.client._ensure_broadcast_domain_for_port.called)
self.client.send_request.assert_has_calls([
mock.call('net-interface-create', lif_create_args)])
def test_create_vlan(self):
self.mock_object(self.client, 'send_request')
vlan_create_args = {
'vlan-info': {
'parent-interface': fake.PORT,
'node': fake.NODE_NAME,
'vlanid': fake.VLAN
}
}
self.client._create_vlan(fake.NODE_NAME, fake.PORT, fake.VLAN)
self.client.send_request.assert_has_calls([
mock.call('net-vlan-create', vlan_create_args)])
def test_create_vlan_already_present(self):
self.mock_object(self.client,
'send_request',
self._mock_api_error(code=netapp_api.EDUPLICATEENTRY))
vlan_create_args = {
'vlan-info': {
'parent-interface': fake.PORT,
'node': fake.NODE_NAME,
'vlanid': fake.VLAN
}
}
self.client._create_vlan(fake.NODE_NAME, fake.PORT, fake.VLAN)
self.client.send_request.assert_has_calls([
mock.call('net-vlan-create', vlan_create_args)])
self.assertEqual(1, client_cmode.LOG.debug.call_count)
def test_create_vlan_api_error(self):
self.mock_object(self.client, 'send_request', self._mock_api_error())
self.assertRaises(exception.NetAppException,
self.client._create_vlan,
fake.NODE_NAME,
fake.PORT,
fake.VLAN)
def test_delete_vlan(self):
self.mock_object(self.client, 'send_request')
vlan_delete_args = {
'vlan-info': {
'parent-interface': fake.PORT,
'node': fake.NODE_NAME,
'vlanid': fake.VLAN
}
}
self.client.delete_vlan(fake.NODE_NAME, fake.PORT, fake.VLAN)
self.client.send_request.assert_has_calls([
mock.call('net-vlan-delete', vlan_delete_args)])
def test_delete_vlan_still_used(self):
self.mock_object(self.client,
'send_request',
self._mock_api_error(code=netapp_api.EAPIERROR,
message='Port already has a '
'lif bound. '))
vlan_delete_args = {
'vlan-info': {
'parent-interface': fake.PORT,
'node': fake.NODE_NAME,
'vlanid': fake.VLAN
}
}
self.client.delete_vlan(fake.NODE_NAME, fake.PORT, fake.VLAN)
self.client.send_request.assert_has_calls([
mock.call('net-vlan-delete', vlan_delete_args)])
self.assertEqual(1, client_cmode.LOG.debug.call_count)
def test_delete_vlan_api_error(self):
self.mock_object(self.client, 'send_request', self._mock_api_error())
self.assertRaises(exception.NetAppException,
self.client.delete_vlan,
fake.NODE_NAME,
fake.PORT,
fake.VLAN)
@ddt.data(('10.10.10.0/24', '10.10.10.1', False),
('fc00::/7', 'fe80::1', False),
('0.0.0.0/0', '10.10.10.1', True),
('::/0', 'fe80::1', True))
@ddt.unpack
def test_create_route(self, subnet, gateway, omit_destination):
api_response = netapp_api.NaElement(
fake.NET_ROUTES_CREATE_RESPONSE)
expected_api_args = {
'destination': subnet,
'gateway': gateway,
'return-record': 'true',
}
self.mock_object(
self.client, 'send_request', mock.Mock(return_value=api_response))
destination = None if omit_destination else subnet
self.client.create_route(gateway, destination=destination)
self.client.send_request.assert_called_once_with(
'net-routes-create', expected_api_args)
def test_create_route_duplicate(self):
self.mock_object(client_cmode.LOG, 'debug')
expected_api_args = {
'destination': fake.SUBNET,
'gateway': fake.GATEWAY,
'return-record': 'true',
}
self.mock_object(
self.client, 'send_request',
mock.Mock(side_effect=self._mock_api_error(
code=netapp_api.EAPIERROR, message='Duplicate route exists.')))
self.client.create_route(fake.GATEWAY, destination=fake.SUBNET)
self.client.send_request.assert_called_once_with(
'net-routes-create', expected_api_args)
self.assertEqual(1, client_cmode.LOG.debug.call_count)
def test_create_route_api_error(self):
expected_api_args = {
'destination': fake.SUBNET,
'gateway': fake.GATEWAY,
'return-record': 'true',
}
self.mock_object(
self.client, 'send_request',
mock.Mock(side_effect=self._mock_api_error()))
self.assertRaises(exception.NetAppException,
self.client.create_route,
fake.GATEWAY, destination=fake.SUBNET)
self.client.send_request.assert_called_once_with(
'net-routes-create', expected_api_args)
def test_create_route_without_gateway(self):
self.mock_object(self.client, 'send_request')
self.client.create_route(None, destination=fake.SUBNET)
self.assertFalse(self.client.send_request.called)
def test_ensure_broadcast_domain_for_port_domain_match(self):
port_info = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
}
self.mock_object(self.client,
'_get_broadcast_domain_for_port',
mock.Mock(return_value=port_info))
self.mock_object(self.client,
'_broadcast_domain_exists',
mock.Mock(return_value=True))
self.mock_object(self.client, '_create_broadcast_domain')
self.mock_object(self.client, '_modify_broadcast_domain')
self.mock_object(self.client, '_add_port_to_broadcast_domain')
self.client._ensure_broadcast_domain_for_port(
fake.NODE_NAME, fake.PORT, fake.MTU, ipspace=fake.IPSPACE_NAME)
self.client._get_broadcast_domain_for_port.assert_called_once_with(
fake.NODE_NAME, fake.PORT)
self.client._modify_broadcast_domain.assert_called_once_with(
fake.BROADCAST_DOMAIN, fake.IPSPACE_NAME, fake.MTU)
self.assertFalse(self.client._broadcast_domain_exists.called)
self.assertFalse(self.client._create_broadcast_domain.called)
self.assertFalse(self.client._add_port_to_broadcast_domain.called)
@ddt.data(fake.IPSPACE_NAME, client_cmode.DEFAULT_IPSPACE)
def test_ensure_broadcast_domain_for_port_other_domain(self, ipspace):
port_info = {
'ipspace': ipspace,
'broadcast-domain': 'other_domain',
}
self.mock_object(self.client,
'_get_broadcast_domain_for_port',
mock.Mock(return_value=port_info))
self.mock_object(self.client,
'_broadcast_domain_exists',
mock.Mock(return_value=True))
self.mock_object(self.client, '_create_broadcast_domain')
self.mock_object(self.client, '_modify_broadcast_domain')
self.mock_object(self.client, '_remove_port_from_broadcast_domain')
self.mock_object(self.client, '_add_port_to_broadcast_domain')
self.client._ensure_broadcast_domain_for_port(
fake.NODE_NAME, fake.PORT, ipspace=fake.IPSPACE_NAME, mtu=fake.MTU)
self.client._get_broadcast_domain_for_port.assert_called_once_with(
fake.NODE_NAME, fake.PORT)
self.client._remove_port_from_broadcast_domain.assert_called_once_with(
fake.NODE_NAME, fake.PORT, 'other_domain', ipspace)
self.client._broadcast_domain_exists.assert_called_once_with(
fake.BROADCAST_DOMAIN, fake.IPSPACE_NAME)
self.assertFalse(self.client._create_broadcast_domain.called)
self.client._modify_broadcast_domain.assert_called_once_with(
fake.BROADCAST_DOMAIN, fake.IPSPACE_NAME, fake.MTU)
self.client._add_port_to_broadcast_domain.assert_called_once_with(
fake.NODE_NAME, fake.PORT, fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
def test_ensure_broadcast_domain_for_port_no_domain(self):
port_info = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': None,
}
self.mock_object(self.client,
'_get_broadcast_domain_for_port',
mock.Mock(return_value=port_info))
self.mock_object(self.client,
'_broadcast_domain_exists',
mock.Mock(return_value=False))
self.mock_object(self.client, '_create_broadcast_domain')
self.mock_object(self.client, '_modify_broadcast_domain')
self.mock_object(self.client, '_remove_port_from_broadcast_domain')
self.mock_object(self.client, '_add_port_to_broadcast_domain')
self.client._ensure_broadcast_domain_for_port(
fake.NODE_NAME, fake.PORT, ipspace=fake.IPSPACE_NAME, mtu=fake.MTU)
self.client._get_broadcast_domain_for_port.assert_called_once_with(
fake.NODE_NAME, fake.PORT)
self.assertFalse(self.client._remove_port_from_broadcast_domain.called)
self.client._broadcast_domain_exists.assert_called_once_with(
fake.BROADCAST_DOMAIN, fake.IPSPACE_NAME)
self.client._create_broadcast_domain.assert_called_once_with(
fake.BROADCAST_DOMAIN, fake.IPSPACE_NAME, fake.MTU)
self.assertFalse(self.client._modify_broadcast_domain.called)
self.client._add_port_to_broadcast_domain.assert_called_once_with(
fake.NODE_NAME, fake.PORT, fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
def test_get_broadcast_domain_for_port(self):
api_response = netapp_api.NaElement(
fake.NET_PORT_GET_ITER_BROADCAST_DOMAIN_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
net_port_get_iter_args = {
'query': {
'net-port-info': {
'node': fake.NODE_NAME,
'port': fake.PORT,
},
},
'desired-attributes': {
'net-port-info': {
'broadcast-domain': None,
'ipspace': None,
},
},
}
result = self.client._get_broadcast_domain_for_port(fake.NODE_NAME,
fake.PORT)
expected = {
'broadcast-domain': fake.BROADCAST_DOMAIN,
'ipspace': fake.IPSPACE_NAME,
}
self.client.send_iter_request.assert_has_calls([
mock.call('net-port-get-iter', net_port_get_iter_args)])
self.assertEqual(expected, result)
def test_get_broadcast_domain_for_port_port_not_found(self):
api_response = netapp_api.NaElement(
fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NetAppException,
self.client._get_broadcast_domain_for_port,
fake.NODE_NAME,
fake.PORT)
def test_get_broadcast_domain_for_port_domain_not_found(self):
api_response = netapp_api.NaElement(
fake.NET_PORT_GET_ITER_BROADCAST_DOMAIN_MISSING_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client._get_broadcast_domain_for_port(fake.NODE_NAME,
fake.PORT)
expected = {
'broadcast-domain': None,
'ipspace': fake.IPSPACE_NAME,
}
self.assertEqual(expected, result)
def test_broadcast_domain_exists(self):
api_response = netapp_api.NaElement(
fake.NET_PORT_BROADCAST_DOMAIN_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client._broadcast_domain_exists(fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
net_port_broadcast_domain_get_iter_args = {
'query': {
'net-port-broadcast-domain-info': {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
},
},
'desired-attributes': {
'net-port-broadcast-domain-info': None,
},
}
self.client.send_iter_request.assert_has_calls([
mock.call('net-port-broadcast-domain-get-iter',
net_port_broadcast_domain_get_iter_args)])
self.assertTrue(result)
def test_broadcast_domain_exists_not_found(self):
api_response = netapp_api.NaElement(
fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client._broadcast_domain_exists(fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
self.assertFalse(result)
def test_create_broadcast_domain(self):
self.mock_object(self.client, 'send_request')
result = self.client._create_broadcast_domain(fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME,
fake.MTU)
net_port_broadcast_domain_create_args = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
'mtu': fake.MTU,
}
self.assertIsNone(result)
self.client.send_request.assert_has_calls([
mock.call('net-port-broadcast-domain-create',
net_port_broadcast_domain_create_args)])
def test_modify_broadcast_domain(self):
self.mock_object(self.client, 'send_request')
result = self.client._modify_broadcast_domain(fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME,
fake.MTU)
net_port_broadcast_domain_modify_args = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
'mtu': fake.MTU,
}
self.assertIsNone(result)
self.client.send_request.assert_called_once_with(
'net-port-broadcast-domain-modify',
net_port_broadcast_domain_modify_args)
def test_delete_broadcast_domain(self):
self.mock_object(self.client, 'send_request')
result = self.client._delete_broadcast_domain(fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
net_port_broadcast_domain_delete_args = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
}
self.assertIsNone(result)
self.client.send_request.assert_has_calls([
mock.call('net-port-broadcast-domain-destroy',
net_port_broadcast_domain_delete_args)])
def test_delete_broadcast_domains_for_ipspace_not_found(self):
self.mock_object(self.client,
'get_ipspaces',
mock.Mock(return_value=[]))
self.mock_object(self.client, '_delete_broadcast_domain')
self.client._delete_broadcast_domains_for_ipspace(fake.IPSPACE_NAME)
self.client.get_ipspaces.assert_called_once_with(
ipspace_name=fake.IPSPACE_NAME)
self.assertFalse(self.client._delete_broadcast_domain.called)
def test_delete_broadcast_domains_for_ipspace(self):
self.mock_object(self.client,
'get_ipspaces',
mock.Mock(return_value=fake.IPSPACES))
self.mock_object(self.client, '_delete_broadcast_domain')
self.client._delete_broadcast_domains_for_ipspace(fake.IPSPACE_NAME)
self.client.get_ipspaces.assert_called_once_with(
ipspace_name=fake.IPSPACE_NAME)
self.client._delete_broadcast_domain.assert_called_once_with(
fake.IPSPACES[0]['broadcast-domains'][0], fake.IPSPACE_NAME)
def test_add_port_to_broadcast_domain(self):
self.mock_object(self.client, 'send_request')
add_port_to_broadcast_domain_args = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
'ports': {
'net-qualified-port-name': ':'.join([fake.NODE_NAME,
fake.VLAN_PORT])
}
}
result = self.client._add_port_to_broadcast_domain(
fake.NODE_NAME, fake.VLAN_PORT, fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
self.assertIsNone(result)
self.client.send_request.assert_has_calls([
mock.call('net-port-broadcast-domain-add-ports',
add_port_to_broadcast_domain_args)])
def test_add_port_to_broadcast_domain_already_present(self):
self.mock_object(self.client, 'send_request', self._mock_api_error(
code=netapp_api.
E_VIFMGR_PORT_ALREADY_ASSIGNED_TO_BROADCAST_DOMAIN))
result = self.client._add_port_to_broadcast_domain(
fake.NODE_NAME, fake.VLAN_PORT, fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
self.assertIsNone(result)
def test_add_port_to_broadcast_domain_api_error(self):
self.mock_object(self.client, 'send_request', self._mock_api_error())
self.assertRaises(exception.NetAppException,
self.client._add_port_to_broadcast_domain,
fake.NODE_NAME,
fake.VLAN_PORT,
fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
def test_remove_port_from_broadcast_domain(self):
self.mock_object(self.client, 'send_request')
result = self.client._remove_port_from_broadcast_domain(
fake.NODE_NAME, fake.VLAN_PORT, fake.BROADCAST_DOMAIN,
fake.IPSPACE_NAME)
net_port_broadcast_domain_remove_ports_args = {
'ipspace': fake.IPSPACE_NAME,
'broadcast-domain': fake.BROADCAST_DOMAIN,
'ports': {
'net-qualified-port-name': ':'.join([fake.NODE_NAME,
fake.VLAN_PORT])
}
}
self.assertIsNone(result)
self.client.send_request.assert_has_calls([
mock.call('net-port-broadcast-domain-remove-ports',
net_port_broadcast_domain_remove_ports_args)])
def test_network_interface_exists(self):
api_response = netapp_api.NaElement(
fake.NET_INTERFACE_GET_ONE_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
net_interface_get_args = {
'query': {
'net-interface-info': {
'address': fake.IP_ADDRESS,
'home-node': fake.NODE_NAME,
'home-port': fake.VLAN_PORT,
'netmask': fake.NETMASK,
'vserver': fake.VSERVER_NAME}
},
'desired-attributes': {
'net-interface-info': {
'interface-name': None,
}
}
}
result = self.client.network_interface_exists(
fake.VSERVER_NAME, fake.NODE_NAME, fake.PORT, fake.IP_ADDRESS,
fake.NETMASK, fake.VLAN)
self.client.send_iter_request.assert_has_calls([
mock.call('net-interface-get-iter', net_interface_get_args)])
self.assertTrue(result)
def test_network_interface_exists_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
net_interface_get_args = {
'query': {
'net-interface-info': {
'address': fake.IP_ADDRESS,
'home-node': fake.NODE_NAME,
'home-port': fake.PORT,
'netmask': fake.NETMASK,
'vserver': fake.VSERVER_NAME}
},
'desired-attributes': {
'net-interface-info': {
'interface-name': None,
}
}
}
result = self.client.network_interface_exists(
fake.VSERVER_NAME, fake.NODE_NAME, fake.PORT, fake.IP_ADDRESS,
fake.NETMASK, None)
self.client.send_iter_request.assert_has_calls([
mock.call('net-interface-get-iter', net_interface_get_args)])
self.assertFalse(result)
def test_list_network_interfaces(self):
api_response = netapp_api.NaElement(
fake.NET_INTERFACE_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
net_interface_get_args = {
'desired-attributes': {
'net-interface-info': {
'interface-name': None,
}
}
}
result = self.client.list_network_interfaces()
self.client.send_iter_request.assert_has_calls([
mock.call('net-interface-get-iter', net_interface_get_args)])
self.assertSequenceEqual(fake.LIF_NAMES, result)
def test_list_network_interfaces_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.list_network_interfaces()
self.assertListEqual([], result)
def test_get_network_interfaces(self):
api_response = netapp_api.NaElement(
fake.NET_INTERFACE_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_network_interfaces()
self.client.send_iter_request.assert_has_calls([
mock.call('net-interface-get-iter', None)])
self.assertSequenceEqual(fake.LIFS, result)
def test_get_network_interfaces_filtered_by_protocol(self):
api_response = netapp_api.NaElement(
fake.NET_INTERFACE_GET_ITER_RESPONSE_NFS)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_network_interfaces(protocols=['NFS'])
net_interface_get_args = {
'query': {
'net-interface-info': {
'data-protocols': {
'data-protocol': 'nfs',
}
}
}
}
self.client.send_iter_request.assert_has_calls([
mock.call('net-interface-get-iter', net_interface_get_args)])
self.assertListEqual(fake.NFS_LIFS, result)
def test_get_network_interfaces_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_network_interfaces()
self.client.send_iter_request.assert_has_calls([
mock.call('net-interface-get-iter', None)])
self.assertListEqual([], result)
def test_get_ipspaces(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(
fake.NET_IPSPACES_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_ipspaces(ipspace_name=fake.IPSPACE_NAME)
net_ipspaces_get_iter_args = {
'query': {
'net-ipspaces-info': {
'ipspace': fake.IPSPACE_NAME,
},
},
}
self.client.send_iter_request.assert_has_calls([
mock.call('net-ipspaces-get-iter', net_ipspaces_get_iter_args)])
self.assertEqual(fake.IPSPACES, result)
def test_get_ipspaces_not_found(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_ipspaces()
net_ipspaces_get_iter_args = {}
self.client.send_iter_request.assert_has_calls([
mock.call('net-ipspaces-get-iter', net_ipspaces_get_iter_args)])
self.assertEqual([], result)
def test_get_ipspaces_not_supported(self):
self.mock_object(self.client, 'send_iter_request')
result = self.client.get_ipspaces()
self.assertFalse(self.client.send_iter_request.called)
self.assertEqual([], result)
@ddt.data((fake.NET_IPSPACES_GET_ITER_RESPONSE, True),
(fake.NO_RECORDS_RESPONSE, False))
@ddt.unpack
def test_ipspace_exists(self, api_response, expected):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(api_response)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.ipspace_exists(fake.IPSPACE_NAME)
net_ipspaces_get_iter_args = {
'query': {
'net-ipspaces-info': {
'ipspace': fake.IPSPACE_NAME,
},
},
'desired-attributes': {
'net-ipspaces-info': {
'ipspace': None,
},
},
}
self.client.send_iter_request.assert_has_calls([
mock.call('net-ipspaces-get-iter', net_ipspaces_get_iter_args)])
self.assertEqual(expected, result)
def test_ipspace_exists_not_supported(self):
result = self.client.ipspace_exists(fake.IPSPACE_NAME)
self.assertFalse(result)
def test_create_ipspace(self):
self.mock_object(self.client, 'send_request')
self.client.create_ipspace(fake.IPSPACE_NAME)
net_ipspaces_create_args = {'ipspace': fake.IPSPACE_NAME}
self.client.send_request.assert_has_calls([
mock.call('net-ipspaces-create', net_ipspaces_create_args)])
def test_delete_ipspace(self):
mock_delete_broadcast_domains_for_ipspace = self.mock_object(
self.client, '_delete_broadcast_domains_for_ipspace')
self.mock_object(self.client, 'send_request')
self.client.delete_ipspace(fake.IPSPACE_NAME)
net_ipspaces_destroy_args = {'ipspace': fake.IPSPACE_NAME}
mock_delete_broadcast_domains_for_ipspace.assert_called_once_with(
fake.IPSPACE_NAME)
self.client.send_request.assert_has_calls([
mock.call('net-ipspaces-destroy', net_ipspaces_destroy_args)])
def test_get_ipspace_name_for_vlan_port(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(fake.NET_PORT_GET_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
ipspace = self.client.get_ipspace_name_for_vlan_port(
fake.NODE_NAME, fake.PORT, fake.VLAN)
port = '%(port)s-%(id)s' % {'port': fake.PORT, 'id': fake.VLAN}
self.client.send_request.assert_called_once_with(
'net-port-get',
{'node': fake.NODE_NAME, 'port': port})
self.assertEqual(fake.IPSPACE_NAME, ipspace)
def test_get_ipspace_name_for_vlan_port_no_ipspace_feature(self):
self.mock_object(self.client, 'send_request')
ipspace = self.client.get_ipspace_name_for_vlan_port(
fake.NODE_NAME, fake.PORT, fake.VLAN)
self.client.send_request.assert_not_called()
self.assertIsNone(ipspace)
def test_get_ipspace_name_for_vlan_port_no_ipspace_found(self):
self.client.features.add_feature('IPSPACES')
self.mock_object(
self.client,
'send_request',
self._mock_api_error(code=netapp_api.EOBJECTNOTFOUND))
ipspace = self.client.get_ipspace_name_for_vlan_port(
fake.NODE_NAME, fake.PORT, fake.VLAN)
self.assertIsNone(ipspace)
def test_get_ipspace_name_for_vlan_port_no_vlan(self):
self.client.features.add_feature('IPSPACES')
api_response = netapp_api.NaElement(fake.NET_PORT_GET_RESPONSE_NO_VLAN)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
ipspace = self.client.get_ipspace_name_for_vlan_port(
fake.NODE_NAME, fake.PORT, None)
self.client.send_request.assert_called_once_with(
'net-port-get',
{'node': fake.NODE_NAME, 'port': fake.PORT})
self.assertEqual(fake.IPSPACE_NAME, ipspace)
def test_get_ipspace_name_for_vlan_port_raises_api_error(self):
self.client.features.add_feature('IPSPACES')
self.mock_object(self.client,
'send_request',
mock.Mock(side_effect=self._mock_api_error()))
self.assertRaises(netapp_api.NaApiError,
self.client.get_ipspace_name_for_vlan_port,
fake.NODE_NAME, fake.VLAN_PORT, None)
def test_add_vserver_to_ipspace(self):
self.mock_object(self.client, 'send_request')
self.client.add_vserver_to_ipspace(fake.IPSPACE_NAME,
fake.VSERVER_NAME)
net_ipspaces_assign_vserver_args = {
'ipspace': fake.IPSPACE_NAME,
'vserver': fake.VSERVER_NAME
}
self.client.send_request.assert_has_calls([
mock.call('net-ipspaces-assign-vserver',
net_ipspaces_assign_vserver_args)])
def test_get_node_for_aggregate(self):
api_response = netapp_api.NaElement(
fake.AGGR_GET_NODE_RESPONSE).get_child_by_name(
'attributes-list').get_children()
self.mock_object(self.client,
'_get_aggregates',
mock.Mock(return_value=api_response))
result = self.client.get_node_for_aggregate(fake.SHARE_AGGREGATE_NAME)
desired_attributes = {
'aggr-attributes': {
'aggregate-name': None,
'aggr-ownership-attributes': {
'home-name': None,
},
},
}
self.client._get_aggregates.assert_has_calls([
mock.call(
aggregate_names=[fake.SHARE_AGGREGATE_NAME],
desired_attributes=desired_attributes)])
self.assertEqual(fake.NODE_NAME, result)
def test_get_node_for_aggregate_none_requested(self):
result = self.client.get_node_for_aggregate(None)
self.assertIsNone(result)
def test_get_node_for_aggregate_api_not_found(self):
self.mock_object(self.client,
'send_iter_request',
mock.Mock(side_effect=self._mock_api_error(
netapp_api.EAPINOTFOUND)))
result = self.client.get_node_for_aggregate(fake.SHARE_AGGREGATE_NAME)
self.assertIsNone(result)
def test_get_node_for_aggregate_api_error(self):
self.mock_object(self.client,
'send_iter_request',
self._mock_api_error())
self.assertRaises(netapp_api.NaApiError,
self.client.get_node_for_aggregate,
fake.SHARE_AGGREGATE_NAME)
def test_get_node_for_aggregate_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client.get_node_for_aggregate(fake.SHARE_AGGREGATE_NAME)
self.assertIsNone(result)
def test_get_cluster_aggregate_capacities(self):
api_response = netapp_api.NaElement(
fake.AGGR_GET_SPACE_RESPONSE).get_child_by_name(
'attributes-list').get_children()
self.mock_object(self.client,
'_get_aggregates',
mock.Mock(return_value=api_response))
result = self.client.get_cluster_aggregate_capacities(
fake.SHARE_AGGREGATE_NAMES)
desired_attributes = {
'aggr-attributes': {
'aggregate-name': None,
'aggr-space-attributes': {
'size-available': None,
'size-total': None,
'size-used': None,
}
}
}
self.client._get_aggregates.assert_has_calls([
mock.call(
aggregate_names=fake.SHARE_AGGREGATE_NAMES,
desired_attributes=desired_attributes)])
expected = {
fake.SHARE_AGGREGATE_NAMES[0]: {
'available': 45670400,
'total': 943718400,
'used': 898048000,
},
fake.SHARE_AGGREGATE_NAMES[1]: {
'available': 4267659264,
'total': 7549747200,
'used': 3282087936,
},
}
self.assertDictEqual(expected, result)
def test_get_cluster_aggregate_capacities_not_found(self):
api_response = netapp_api.NaElement('none').get_children()
self.mock_object(self.client,
'_get_aggregates',
mock.Mock(return_value=api_response))
result = self.client.get_cluster_aggregate_capacities(
fake.SHARE_AGGREGATE_NAMES)
self.assertEqual({}, result)
def test_get_cluster_aggregate_capacities_none_requested(self):
result = self.client.get_cluster_aggregate_capacities([])
self.assertEqual({}, result)
def test_get_vserver_aggregate_capacities(self):
api_response = netapp_api.NaElement(fake.VSERVER_GET_RESPONSE)
self.mock_object(self.vserver_client,
'send_request',
mock.Mock(return_value=api_response))
result = self.vserver_client.get_vserver_aggregate_capacities()
vserver_args = {
'desired-attributes': {
'vserver-info': {
'vserver-name': None,
'vserver-aggr-info-list': {
'vserver-aggr-info': {
'aggr-name': None,
'aggr-availsize': None
}
}
}
}
}
self.vserver_client.send_request.assert_has_calls([
mock.call('vserver-get', vserver_args)])
self.assertDictEqual(fake.VSERVER_AGGREGATES, result)
def test_get_vserver_aggregate_capacities_partial_request(self):
api_response = netapp_api.NaElement(fake.VSERVER_GET_RESPONSE)
self.mock_object(self.vserver_client,
'send_request',
mock.Mock(return_value=api_response))
result = self.vserver_client.get_vserver_aggregate_capacities(
fake.SHARE_AGGREGATE_NAMES[0])
expected = {fake.SHARE_AGGREGATE_NAMES[0]:
fake.VSERVER_AGGREGATES[fake.SHARE_AGGREGATE_NAMES[0]]}
self.assertDictEqual(expected, result)
def test_get_vserver_aggregate_capacities_aggregate_not_found(self):
api_response = netapp_api.NaElement(
fake.VSERVER_GET_RESPONSE_NO_AGGREGATES)
self.mock_object(self.vserver_client,
'send_request',
mock.Mock(return_value=api_response))
result = self.vserver_client.get_vserver_aggregate_capacities()
self.assertDictEqual({}, result)
self.assertEqual(1, client_cmode.LOG.warning.call_count)
def test_get_vserver_aggregate_capacities_vserver_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.vserver_client,
'send_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NetAppException,
self.vserver_client.get_vserver_aggregate_capacities)
def test_get_vserver_aggregate_capacities_none_requested(self):
result = self.client.get_vserver_aggregate_capacities([])
self.assertEqual({}, result)
def test_get_aggregates(self):
api_response = netapp_api.NaElement(fake.AGGR_GET_ITER_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client._get_aggregates()
self.client.send_iter_request.assert_has_calls([
mock.call('aggr-get-iter', {})])
self.assertListEqual(
[aggr.to_string() for aggr in api_response.get_child_by_name(
'attributes-list').get_children()],
[aggr.to_string() for aggr in result])
def test_get_aggregates_with_filters(self):
api_response = netapp_api.NaElement(fake.AGGR_GET_SPACE_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
desired_attributes = {
'aggr-attributes': {
'aggregate-name': None,
'aggr-space-attributes': {
'size-total': None,
'size-available': None,
}
}
}
result = self.client._get_aggregates(
aggregate_names=fake.SHARE_AGGREGATE_NAMES,
desired_attributes=desired_attributes)
aggr_get_iter_args = {
'query': {
'aggr-attributes': {
'aggregate-name': '|'.join(fake.SHARE_AGGREGATE_NAMES),
}
},
'desired-attributes': desired_attributes
}
self.client.send_iter_request.assert_has_calls([
mock.call('aggr-get-iter', aggr_get_iter_args)])
self.assertListEqual(
[aggr.to_string() for aggr in api_response.get_child_by_name(
'attributes-list').get_children()],
[aggr.to_string() for aggr in result])
def test_get_aggregates_not_found(self):
api_response = netapp_api.NaElement(fake.NO_RECORDS_RESPONSE)
self.mock_object(self.client,
'send_iter_request',
mock.Mock(return_value=api_response))
result = self.client._get_aggregates()
self.client.send_iter_request.assert_has_calls([
mock.call('aggr-get-iter', {})])
self.assertListEqual([], result)
def test_get_performance_instance_uuids(self):
api_response = netapp_api.NaElement(
fake.PERF_OBJECT_INSTANCE_LIST_INFO_ITER_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.get_performance_instance_uuids(
'system', fake.NODE_NAME)
expected = [fake.NODE_NAME + ':kernel:system']
self.assertEqual(expected, result)
perf_object_instance_list_info_iter_args = {
'objectname': 'system',
'query': {
'instance-info': {
'uuid': fake.NODE_NAME + ':*',
}
}
}
self.client.send_request.assert_called_once_with(
'perf-object-instance-list-info-iter',
perf_object_instance_list_info_iter_args)
def test_get_performance_counter_info(self):
api_response = netapp_api.NaElement(
fake.PERF_OBJECT_COUNTER_LIST_INFO_WAFL_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
result = self.client.get_performance_counter_info('wafl',
'cp_phase_times')
expected = {
'name': 'cp_phase_times',
'base-counter': 'total_cp_msecs',
'labels': fake.PERF_OBJECT_COUNTER_TOTAL_CP_MSECS_LABELS,
}
self.assertEqual(expected, result)
perf_object_counter_list_info_args = {'objectname': 'wafl'}
self.client.send_request.assert_called_once_with(
'perf-object-counter-list-info',
perf_object_counter_list_info_args)
def test_get_performance_counter_info_not_found(self):
api_response = netapp_api.NaElement(
fake.PERF_OBJECT_COUNTER_LIST_INFO_WAFL_RESPONSE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
self.assertRaises(exception.NotFound,
self.client.get_performance_counter_info,
'wafl',
'invalid')
def test_get_performance_counters(self):
api_response = netapp_api.NaElement(
fake.PERF_OBJECT_GET_INSTANCES_SYSTEM_RESPONSE_CMODE)
self.mock_object(self.client,
'send_request',
mock.Mock(return_value=api_response))
instance_uuids = [
fake.NODE_NAMES[0] + ':kernel:system',
fake.NODE_NAMES[1] + ':kernel:system',
]
counter_names = ['avg_processor_busy']
result = self.client.get_performance_counters('system',