x
/
vmware-nsxlib
Code Issues Proposed changes
vmware-nsxlib/vmware_nsxlib/tests/unit/v3/test_resources.py

729 lines
25 KiB
Python
Raw Blame History

# Copyright 2015 VMware, Inc.
# 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 mock
from oslo_serialization import jsonutils
from vmware_nsxlib.tests.unit.v3 import mocks
from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase
from vmware_nsxlib.tests.unit.v3 import test_client
from vmware_nsxlib.tests.unit.v3 import test_constants
from vmware_nsxlib.v3 import exceptions
from vmware_nsxlib.v3 import resources
CLIENT_PKG = test_client.CLIENT_PKG
profile_types = resources.SwitchingProfileTypes
class TestSwitchingProfileTestCase(nsxlib_testcase.NsxClientTestCase):
def _mocked_switching_profile(self, session_response=None):
return self.mocked_resource(
resources.SwitchingProfile, session_response=session_response)
def test_switching_profile_create(self):
mocked_resource = self._mocked_switching_profile()
mocked_resource.create(profile_types.PORT_MIRRORING,
'pm-profile', 'port mirror prof')
test_client.assert_json_call(
'post', mocked_resource,
'https://1.2.3.4/api/v1/switching-profiles',
data=jsonutils.dumps({
'resource_type': profile_types.PORT_MIRRORING,
'display_name': 'pm-profile',
'description': 'port mirror prof'
}, sort_keys=True))
def test_switching_profile_update(self):
tags = [
{
'scope': 'os-project-id',
'tag': 'tenant-1'
},
{
'scope': 'os-api-version',
'tag': '2.1.1.0'
}
]
mocked_resource = self._mocked_switching_profile()
mocked_resource.update(
'a12bc1', profile_types.PORT_MIRRORING, tags=tags)
test_client.assert_json_call(
'put', mocked_resource,
'https://1.2.3.4/api/v1/switching-profiles/a12bc1',
data=jsonutils.dumps({
'resource_type': profile_types.PORT_MIRRORING,
'tags': tags
}, sort_keys=True))
def test_spoofgaurd_profile_create(self):
tags = [
{
'scope': 'os-project-id',
'tag': 'tenant-1'
},
{
'scope': 'os-api-version',
'tag': '2.1.1.0'
}
]
mocked_resource = self._mocked_switching_profile()
mocked_resource.create_spoofguard_profile(
'plugin-spoof', 'spoofguard-for-plugin',
whitelist_ports=True, tags=tags)
test_client.assert_json_call(
'post', mocked_resource,
'https://1.2.3.4/api/v1/switching-profiles',
data=jsonutils.dumps({
'resource_type': profile_types.SPOOF_GUARD,
'display_name': 'plugin-spoof',
'description': 'spoofguard-for-plugin',
'white_list_providers': ['LPORT_BINDINGS'],
'tags': tags
}, sort_keys=True))
def test_create_dhcp_profile(self):
tags = [
{
'scope': 'os-project-id',
'tag': 'tenant-1'
},
{
'scope': 'os-api-version',
'tag': '2.1.1.0'
}
]
mocked_resource = self._mocked_switching_profile()
mocked_resource.create_dhcp_profile(
'plugin-dhcp', 'dhcp-for-plugin',
tags=tags)
test_client.assert_json_call(
'post', mocked_resource,
'https://1.2.3.4/api/v1/switching-profiles',
data=jsonutils.dumps({
'bpdu_filter': {
'enabled': True,
'white_list': []
},
'resource_type': profile_types.SWITCH_SECURITY,
'display_name': 'plugin-dhcp',
'description': 'dhcp-for-plugin',
'tags': tags,
'dhcp_filter': {
'client_block_enabled': True,
'server_block_enabled': False
},
'rate_limits': {
'enabled': False,
'rx_broadcast': 0,
'tx_broadcast': 0,
'rx_multicast': 0,
'tx_multicast': 0
},
'block_non_ip_traffic': True
}, sort_keys=True))
def test_create_mac_learning_profile(self):
tags = [
{
'scope': 'os-project-id',
'tag': 'tenant-1'
},
{
'scope': 'os-api-version',
'tag': '2.1.1.0'
}
]
mocked_resource = self._mocked_switching_profile()
mocked_resource.create_mac_learning_profile(
'plugin-mac-learning', 'mac-learning-for-plugin',
tags=tags)
test_client.assert_json_call(
'post', mocked_resource,
'https://1.2.3.4/api/v1/switching-profiles',
data=jsonutils.dumps({
'mac_learning': {
'enabled': True,
},
'resource_type': profile_types.MAC_LEARNING,
'display_name': 'plugin-mac-learning',
'description': 'mac-learning-for-plugin',
'tags': tags,
'mac_change_allowed': True,
}, sort_keys=True))
def test_find_by_display_name(self):
resp_resources = {
'results': [
{'display_name': 'resource-1'},
{'display_name': 'resource-2'},
{'display_name': 'resource-3'}
]
}
session_response = mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources))
mocked_resource = self._mocked_switching_profile(
session_response=[session_response] * 3)
self.assertEqual([{'display_name': 'resource-1'}],
mocked_resource.find_by_display_name('resource-1'))
self.assertEqual([{'display_name': 'resource-2'}],
mocked_resource.find_by_display_name('resource-2'))
self.assertEqual([{'display_name': 'resource-3'}],
mocked_resource.find_by_display_name('resource-3'))
resp_resources = {
'results': [
{'display_name': 'resource-1'},
{'display_name': 'resource-1'},
{'display_name': 'resource-1'}
]
}
session_response = mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources))
mocked_resource = self._mocked_switching_profile(
session_response=session_response)
self.assertEqual(resp_resources['results'],
mocked_resource.find_by_display_name('resource-1'))
def test_list_all_profiles(self):
mocked_resource = self._mocked_switching_profile()
mocked_resource.list()
test_client.assert_json_call(
'get', mocked_resource,
'https://1.2.3.4/api/v1/switching-profiles/'
'?include_system_owned=True',
data=None)
class LogicalPortTestCase(nsxlib_testcase.NsxClientTestCase):
def _mocked_lport(self, mock_validate=True, session_response=None):
return self.mocked_resource(
resources.LogicalPort, mock_validate=True,
session_response=session_response)
def _get_profile_dicts(self, fake_port):
fake_profile_dicts = []
for profile_id in fake_port['switching_profile_ids']:
fake_profile_dicts.append({'resource_type': profile_id['key'],
'id': profile_id['value']})
return fake_profile_dicts
def _get_pktcls_bindings(self):
fake_pkt_classifiers = []
fake_binding_repr = []
for i in range(0, 3):
ip = "9.10.11.%s" % i
mac = "00:0c:29:35:4a:%sc" % i
fake_pkt_classifiers.append(resources.PacketAddressClassifier(
ip, mac, None))
fake_binding_repr.append({
'ip_address': ip,
'mac_address': mac
})
return fake_pkt_classifiers, fake_binding_repr
def test_create_logical_port(self):
"""Test creating a port
returns the correct response and 200 status
"""
fake_port = test_constants.FAKE_PORT.copy()
profile_dicts = self._get_profile_dicts(fake_port)
pkt_classifiers, binding_repr = self._get_pktcls_bindings()
fake_port['address_bindings'] = binding_repr
mocked_resource = self._mocked_lport()
switch_profile = resources.SwitchingProfile
mocked_resource.create(
fake_port['logical_switch_id'],
fake_port['attachment']['id'],
address_bindings=pkt_classifiers,
switch_profile_ids=switch_profile.build_switch_profile_ids(
mock.Mock(), *profile_dicts))
resp_body = {
'logical_switch_id': fake_port['logical_switch_id'],
'switching_profile_ids': fake_port['switching_profile_ids'],
'attachment': {
'attachment_type': 'VIF',
'id': fake_port['attachment']['id']
},
'admin_state': 'UP',
'address_bindings': fake_port['address_bindings']
}
test_client.assert_json_call(
'post', mocked_resource,
'https://1.2.3.4/api/v1/logical-ports',
data=jsonutils.dumps(resp_body, sort_keys=True))
def test_create_logical_port_with_attachtype_cif(self):
"""Test creating a port returns the correct response and 200 status
"""
fake_port = test_constants.FAKE_CONTAINER_PORT.copy()
profile_dicts = self._get_profile_dicts(fake_port)
pkt_classifiers, binding_repr = self._get_pktcls_bindings()
fake_port['address_bindings'] = binding_repr
mocked_resource = self._mocked_lport()
switch_profile = resources.SwitchingProfile
fake_port_ctx = fake_port['attachment']['context']
fake_container_host_vif_id = fake_port_ctx['container_host_vif_id']
mocked_resource.create(
fake_port['logical_switch_id'],
fake_port['attachment']['id'],
parent_vif_id=fake_container_host_vif_id,
parent_tag=fake_port_ctx['vlan_tag'],
address_bindings=pkt_classifiers,
switch_profile_ids=switch_profile.build_switch_profile_ids(
mock.Mock(), *profile_dicts))
resp_body = {
'logical_switch_id': fake_port['logical_switch_id'],
'switching_profile_ids': fake_port['switching_profile_ids'],
'attachment': {
'attachment_type': 'CIF',
'id': fake_port['attachment']['id'],
'context': {
'vlan_tag': fake_port_ctx['vlan_tag'],
'container_host_vif_id': fake_container_host_vif_id,
'resource_type': 'CifAttachmentContext'
}
},
'admin_state': 'UP',
'address_bindings': fake_port['address_bindings']
}
test_client.assert_json_call(
'post', mocked_resource,
'https://1.2.3.4/api/v1/logical-ports',
data=jsonutils.dumps(resp_body, sort_keys=True))
def test_create_logical_port_admin_down(self):
"""Test creating port with admin_state down"""
fake_port = test_constants.FAKE_PORT
fake_port['admin_state'] = "DOWN"
mocked_resource = self._mocked_lport(
session_response=mocks.MockRequestsResponse(
200, jsonutils.dumps(fake_port)))
result = mocked_resource.create(
test_constants.FAKE_PORT['logical_switch_id'],
test_constants.FAKE_PORT['attachment']['id'],
tags={}, admin_state=False)
self.assertEqual(fake_port, result)
def test_delete_logical_port(self):
"""Test deleting port"""
mocked_resource = self._mocked_lport()
uuid = test_constants.FAKE_PORT['id']
mocked_resource.delete(uuid)
test_client.assert_json_call(
'delete', mocked_resource,
'https://1.2.3.4/api/v1/logical-ports/%s?detach=true' % uuid)
def test_clear_port_bindings(self):
fake_port = copy.copy(test_constants.FAKE_PORT)
fake_port['address_bindings'] = ['a', 'b']
mocked_resource = self._mocked_lport()
def get_fake_port(*args):
return fake_port
mocked_resource.get = get_fake_port
mocked_resource.update(
fake_port['id'], fake_port['id'], address_bindings=[])
fake_port['address_bindings'] = []
test_client.assert_json_call(
'put', mocked_resource,
'https://1.2.3.4/api/v1/logical-ports/%s' % fake_port['id'],
data=jsonutils.dumps(fake_port, sort_keys=True))
def test_create_logical_port_fail(self):
"""
Test the failure of port creation
"""
fake_port = test_constants.FAKE_PORT.copy()
profile_dicts = self._get_profile_dicts(fake_port)
pkt_classifiers, binding_repr = self._get_pktcls_bindings()
fake_port['address_bindings'] = binding_repr
mocked_resource = self._mocked_lport(mock_validate=False)
switch_profile = resources.SwitchingProfile
try:
mocked_resource.create(
fake_port['logical_switch_id'],
fake_port['attachment']['id'],
address_bindings=pkt_classifiers,
switch_profile_ids=switch_profile.build_switch_profile_ids(
mock.Mock(), *profile_dicts))
except exceptions.ManagerError as e:
self.assertIn(nsxlib_testcase.NSX_MANAGER, e.msg)
class LogicalRouterTestCase(nsxlib_testcase.NsxClientTestCase):
def _mocked_lrouter(self, session_response=None):
return self.mocked_resource(
resources.LogicalRouter, session_response=session_response)
def test_create_logical_router(self):
"""Test creating a router returns the correct response and 201 status
"""
fake_router = test_constants.FAKE_ROUTER.copy()
router = self._mocked_lrouter()
tier0_router = True
router.create(fake_router['display_name'], None, None, tier0_router)
data = {
'display_name': fake_router['display_name'],
'router_type': 'TIER0' if tier0_router else 'TIER1',
'tags': None
}
test_client.assert_json_call(
'post', router,
'https://1.2.3.4/api/v1/logical-routers',
data=jsonutils.dumps(data, sort_keys=True))
def test_delete_logical_router(self):
"""Test deleting router"""
router = self._mocked_lrouter()
uuid = test_constants.FAKE_ROUTER['id']
router.delete(uuid)
test_client.assert_json_call(
'delete', router,
'https://1.2.3.4/api/v1/logical-routers/%s' % uuid)
class LogicalRouterPortTestCase(nsxlib_testcase.NsxClientTestCase):
def _mocked_lrport(self, session_response=None):
return self.mocked_resource(
resources.LogicalRouterPort, session_response=session_response)
def test_create_logical_router_port(self):
"""Test creating a router port
returns the correct response and 201 status
"""
fake_router_port = test_constants.FAKE_ROUTER_PORT.copy()
lrport = self._mocked_lrport()
lrport.create(fake_router_port['logical_router_id'],
fake_router_port['display_name'],
None,
fake_router_port['resource_type'],
None, None, None)
data = {
'display_name': fake_router_port['display_name'],
'logical_router_id': fake_router_port['logical_router_id'],
'resource_type': fake_router_port['resource_type'],
'tags': []
}
test_client.assert_json_call(
'post', lrport,
'https://1.2.3.4/api/v1/logical-router-ports',
data=jsonutils.dumps(data, sort_keys=True))
def test_logical_router_port_max_attempts(self):
"""
Test a router port api has the configured retries
"""
lrport = self._mocked_lrport()
self.assertEqual(nsxlib_testcase.NSX_MAX_ATTEMPTS,
lrport._client.max_attempts)
def test_delete_logical_router_port(self):
"""Test deleting router port"""
lrport = self._mocked_lrport()
uuid = test_constants.FAKE_ROUTER_PORT['id']
lrport.delete(uuid)
test_client.assert_json_call(
'delete', lrport,
'https://1.2.3.4/api/v1/logical-router-ports/%s' % uuid)
def test_get_logical_router_port_by_router_id(self):
"""Test getting a router port by router id"""
fake_router_port = test_constants.FAKE_ROUTER_PORT.copy()
resp_resources = {'results': [fake_router_port]}
lrport = self._mocked_lrport(
session_response=mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources)))
router_id = fake_router_port['logical_router_id']
result = lrport.get_by_router_id(router_id)
self.assertEqual(fake_router_port, result[0])
test_client.assert_json_call(
'get', lrport,
'https://1.2.3.4/api/v1/logical-router-ports/?'
'logical_router_id=%s' % router_id)
def test_get_logical_router_port_by_switch_id(self):
"""Test getting a router port by switch id"""
fake_router_port = test_constants.FAKE_ROUTER_PORT.copy()
resp_resources = {
'result_count': 1,
'results': [fake_router_port]
}
lrport = self._mocked_lrport(
session_response=mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources)))
switch_id = test_constants.FAKE_SWITCH_UUID
lrport.get_by_lswitch_id(switch_id)
test_client.assert_json_call(
'get', lrport,
'https://1.2.3.4/api/v1/logical-router-ports/?'
'logical_switch_id=%s' % switch_id)
class IpPoolTestCase(nsxlib_testcase.NsxClientTestCase):
def _mocked_pool(self, session_response=None):
return self.mocked_resource(
resources.IpPool, session_response=session_response)
def test_create_ip_pool_all_args(self):
"""Test creating an IP pool
returns the correct response and 201 status
"""
pool = self._mocked_pool()
display_name = 'dummy'
gateway_ip = '1.1.1.1'
ranges = [{'start': '2.2.2.0', 'end': '2.2.2.255'},
{'start': '3.2.2.0', 'end': '3.2.2.255'}]
cidr = '2.2.2.0/24'
description = 'desc'
dns_nameserver = '7.7.7.7'
pool.create(cidr, ranges=ranges,
display_name=display_name,
gateway_ip=gateway_ip,
description=description,
dns_nameservers=[dns_nameserver])
data = {
'display_name': display_name,
'description': description,
'subnets': [{
'gateway_ip': gateway_ip,
'allocation_ranges': ranges,
'cidr': cidr,
'dns_nameservers': [dns_nameserver]
}]
}
test_client.assert_json_call(
'post', pool,
'https://1.2.3.4/api/v1/pools/ip-pools',
data=jsonutils.dumps(data, sort_keys=True))
def test_create_ip_pool_minimal_args(self):
pool = self._mocked_pool()
ranges = [{'start': '2.2.2.0', 'end': '2.2.2.255'},
{'start': '3.2.2.0', 'end': '3.2.2.255'}]
cidr = '2.2.2.0/24'
pool.create(cidr, ranges=ranges)
data = {
'subnets': [{
'allocation_ranges': ranges,
'cidr': cidr,
}]
}
test_client.assert_json_call(
'post', pool,
'https://1.2.3.4/api/v1/pools/ip-pools',
data=jsonutils.dumps(data, sort_keys=True))
def test_create_ip_pool_no_ranges_with_gateway(self):
pool = self._mocked_pool()
cidr = '2.2.2.0/30'
gateway_ip = '2.2.2.1'
pool.create(cidr, ranges=None, gateway_ip=gateway_ip)
exp_ranges = [{'start': '2.2.2.0', 'end': '2.2.2.0'},
{'start': '2.2.2.2', 'end': '2.2.2.3'}]
data = {
'subnets': [{
'gateway_ip': gateway_ip,
'allocation_ranges': exp_ranges,
'cidr': cidr,
}]
}
test_client.assert_json_call(
'post', pool,
'https://1.2.3.4/api/v1/pools/ip-pools',
data=jsonutils.dumps(data, sort_keys=True))
def test_create_ip_pool_no_ranges_no_gateway(self):
pool = self._mocked_pool()
cidr = '2.2.2.0/30'
pool.create(cidr, ranges=None)
exp_ranges = [{'start': '2.2.2.0', 'end': '2.2.2.3'}]
data = {
'subnets': [{
'allocation_ranges': exp_ranges,
'cidr': cidr,
}]
}
test_client.assert_json_call(
'post', pool,
'https://1.2.3.4/api/v1/pools/ip-pools',
data=jsonutils.dumps(data, sort_keys=True))
def test_create_ip_pool_no_cidr(self):
pool = self._mocked_pool()
gateway_ip = '1.1.1.1'
ranges = [{'start': '2.2.2.0', 'end': '2.2.2.255'},
{'start': '3.2.2.0', 'end': '3.2.2.255'}]
cidr = None
try:
pool.create(cidr, ranges=ranges,
gateway_ip=gateway_ip)
except exceptions.InvalidInput:
# This call should fail
pass
else:
self.fail("shouldn't happen")
def test_get_ip_pool(self):
"""Test getting a router port by router id"""
fake_ip_pool = test_constants.FAKE_IP_POOL.copy()
resp_resources = fake_ip_pool
pool = self._mocked_pool(
session_response=mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources)))
uuid = fake_ip_pool['id']
result = pool.get(uuid)
self.assertEqual(fake_ip_pool, result)
test_client.assert_json_call(
'get', pool,
'https://1.2.3.4/api/v1/pools/ip-pools/%s' % uuid)
def test_delete_ip_pool(self):
"""Test deleting router port"""
pool = self._mocked_pool()
uuid = test_constants.FAKE_IP_POOL['id']
pool.delete(uuid)
test_client.assert_json_call(
'delete', pool,
'https://1.2.3.4/api/v1/pools/ip-pools/%s' % uuid)
def test_allocate_ip_from_pool(self):
pool = self._mocked_pool()
uuid = test_constants.FAKE_IP_POOL['id']
addr = '1.1.1.1'
pool.allocate(uuid, ip_addr=addr)
data = {'allocation_id': addr}
test_client.assert_json_call(
'post', pool,
'https://1.2.3.4/api/v1/pools/ip-pools/%s?action=ALLOCATE' % uuid,
data=jsonutils.dumps(data, sort_keys=True))
def test_release_ip_to_pool(self):
pool = self._mocked_pool()
uuid = test_constants.FAKE_IP_POOL['id']
addr = '1.1.1.1'
pool.release(uuid, addr)
data = {'allocation_id': addr}
test_client.assert_json_call(
'post', pool,
'https://1.2.3.4/api/v1/pools/ip-pools/%s?action=RELEASE' % uuid,
data=jsonutils.dumps(data, sort_keys=True))
def test_get_ip_pool_allocations(self):
"""Test getting a router port by router id"""
fake_ip_pool = test_constants.FAKE_IP_POOL.copy()
resp_resources = fake_ip_pool
pool = self._mocked_pool(
session_response=mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources)))
uuid = fake_ip_pool['id']
result = pool.get_allocations(uuid)
self.assertEqual(fake_ip_pool, result)
test_client.assert_json_call(
'get', pool,
'https://1.2.3.4/api/v1/pools/ip-pools/%s/allocations' % uuid)
View Git Blame Copy Permalink