x/vmware-nsx
Code Issues Proposed changes

refactor NSX v3 UTs

Browse Source 
this patch contains a major refactor of the nsx v3 tests
which have nsx rest api calls in their execution stack.
in particular it removes the global module level patching
that's current done as this type of global patching has
intermittent failures when testr runs tests in parallel.
these failures become more prevalent when the psec
code + tests are added. thus this patch is needed for
the psec drop I have.

NB: as noted in the code, most of the module level
patching / mocking magic can go away once we move
the current functional logic to a rest resource model.

Change-Id: Idec72c464a4a2771f089b840aae129d42816a35f
This commit is contained in:
Boden R 2015-09-30 12:00:17 -06:00
parent 8b8fdf022d
commit eac5453721
12 changed files with 876 additions and 1007 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
vmware_nsx/nsxlib/v3/client.py
View File

@ -19,6 +19,7 @@ from neutron.i18n import _LW, _
from oslo_config import cfg
from oslo_log import log
from oslo_serialization import jsonutils
from vmware_nsx.common import exceptions as nsx_exc
LOG = log.getLogger(__name__)
@ -204,20 +205,25 @@ class NSX3Client(JSONRESTClient):
# NOTE(boden): tmp until all refs use client class
def get_resource(resource):
return NSX3Client().get(resource)
def _get_client(client, *args, **kwargs):
return client or NSX3Client(*args, **kwargs)
# NOTE(boden): tmp until all refs use client class
def create_resource(resource, data):
return NSX3Client(url_prefix=resource).create(body=data)
def get_resource(resource, client=None):
return _get_client(client).get(resource)
# NOTE(boden): tmp until all refs use client class
def update_resource(resource, data):
return NSX3Client().update(resource, body=data)
def create_resource(resource, data, client=None):
return _get_client(client).url_post(resource, body=data)
# NOTE(boden): tmp until all refs use client class
def delete_resource(resource):
return NSX3Client().delete(resource)
def update_resource(resource, data, client=None):
return _get_client(client).update(resource, body=data)
# NOTE(boden): tmp until all refs use client class
def delete_resource(resource, client=None):
return _get_client(client).delete(resource)

2
vmware_nsx/nsxlib/v3/security.py
View File

@ -217,7 +217,7 @@ def _init_nsgroup_container(name, description):
def _init_default_section(name, description, nsgroup_id):
fw_sections = firewall.list_sections()
for section in fw_sections:
if section['display_name'] == name:
if section.get('display_name') == name:
break
else:
section = firewall.create_empty_section(

26
vmware_nsx/plugins/nsx_v3/plugin.py
View File

@ -12,16 +12,8 @@
# 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 six
import netaddr
from oslo_config import cfg
from oslo_log import log
from oslo_utils import excutils
from oslo_utils import importutils
from oslo_utils import uuidutils
import six
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.api.rpc.handlers import dhcp_rpc
@ -56,7 +48,11 @@ from neutron.extensions import securitygroup as ext_sg
from neutron.i18n import _LE, _LI, _LW
from neutron.plugins.common import constants as plugin_const
from neutron.plugins.common import utils as n_utils
from oslo_config import cfg
from oslo_log import log
from oslo_utils import excutils
from oslo_utils import importutils
from oslo_utils import uuidutils
from vmware_nsx.common import config # noqa
from vmware_nsx.common import exceptions as nsx_exc
from vmware_nsx.common import nsx_constants
@ -69,6 +65,7 @@ from vmware_nsx.nsxlib.v3 import resources as nsx_resources
from vmware_nsx.nsxlib.v3 import router as routerlib
from vmware_nsx.nsxlib.v3 import security
LOG = log.getLogger(__name__)
@ -108,6 +105,7 @@ class NsxV3Plugin(db_base_plugin_v2.NeutronDbPluginV2,
self.tier0_groups_dict = {}
self._setup_rpc()
self._nsx_client = nsx_client.NSX3Client()
self._port_client = nsx_resources.LogicalPort(self._nsx_client)
self.nsgroup_container, self.default_section = (
security.init_nsgroup_container_and_default_section_rules())
@ -430,8 +428,7 @@ class NsxV3Plugin(db_base_plugin_v2.NeutronDbPluginV2,
# for ports plugged into a Bridge Endpoint.
vif_uuid = port_data.get('device_id')
attachment_type = port_data.get('device_owner')
port_client = nsx_resources.LogicalPort(self._nsx_client)
result = port_client.create(
result = self._port_client.create(
port_data['network_id'], vif_uuid,
tags=tags,
name=port_data['name'],
@ -519,7 +516,7 @@ class NsxV3Plugin(db_base_plugin_v2.NeutronDbPluginV2,
updated_port = {'port': {ext_sg.SECURITYGROUPS: [],
'admin_state_up': False}}
self.update_port(context, port_id, updated_port)
nsx_resources.LogicalPort(self._nsx_client).delete(nsx_port_id)
self._port_client.delete(nsx_port_id)
self.disassociate_floatingips(context, port_id)
ret_val = super(NsxV3Plugin, self).delete_port(context, port_id)
@ -537,7 +534,7 @@ class NsxV3Plugin(db_base_plugin_v2.NeutronDbPluginV2,
sec_grp_updated = self.update_security_group_on_port(
context, id, port, original_port, updated_port)
try:
nsx_resources.LogicalPort(self._nsx_client).update(
self._port_client.update(
nsx_lport_id, name=port['port'].get('name'),
admin_state=port['port'].get('admin_state_up'))
security.update_lport_with_security_groups(
@ -1083,6 +1080,7 @@ class NsxV3Plugin(db_base_plugin_v2.NeutronDbPluginV2,
tags = utils.build_v3_tags_payload(secgroup)
name = security.get_nsgroup_name(secgroup)
ns_group = None
firewall_section = None
try:
# NOTE(roeyc): We first create the nsgroup so that once the sg is

538
vmware_nsx/tests/unit/nsx_v3/mocks.py
View File

@ -12,15 +12,17 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import requests
import urlparse
import uuid
from oslo_serialization import jsonutils
from oslo_utils import uuidutils
from vmware_nsx.common import exceptions as nsx_exc
from vmware_nsx.common import nsx_constants
FAKE_NAME = "fake_name"
DEFAULT_TIER0_ROUTER_UUID = "fake_default_tier0_router_uuid"
DEFAULT_TIER0_ROUTER_UUID = "efad0078-9204-4b46-a2d8-d4dd31ed448f"
FAKE_MANAGER = "fake_manager_ip"
@ -65,128 +67,6 @@ def make_fake_switch(switch_uuid=None, tz_uuid=None, name=FAKE_NAME):
return fake_switch
def create_logical_switch(display_name, transport_zone_id, tags,
replication_mode=nsx_constants.MTEP,
admin_state=True, vlan_id=None):
return make_fake_switch()
def get_logical_switch(lswitch_id):
return make_fake_switch(switch_uuid=lswitch_id)
def update_logical_switch(lswitch_id, name=None, admin_state=None):
lswitch = get_logical_switch(lswitch_id)
if name is not None:
lswitch['display_name'] = name
if admin_state is not None:
if admin_state:
lswitch['admin_state'] = nsx_constants.ADMIN_STATE_UP
else:
lswitch['admin_state'] = nsx_constants.ADMIN_STATE_DOWN
return lswitch
def create_logical_port(client, lswitch_id, vif_uuid, tags=[],
attachment_type=nsx_constants.ATTACHMENT_VIF,
admin_state=True, name=None, address_bindings=None,
parent_name=None, parent_tag=None):
FAKE_SWITCH_UUID = uuidutils.generate_uuid()
FAKE_PORT_UUID = uuidutils.generate_uuid()
FAKE_PORT = {
"id": FAKE_PORT_UUID,
"display_name": FAKE_NAME,
"resource_type": "LogicalPort",
"address_bindings": [],
"logical_switch_id": FAKE_SWITCH_UUID,
"admin_state": nsx_constants.ADMIN_STATE_UP,
"attachment": {
"id": "9ca8d413-f7bf-4276-b4c9-62f42516bdb2",
"attachment_type": "VIF"
},
"switching_profile_ids": [
{
"value": "64814784-7896-3901-9741-badeff705639",
"key": "IpDiscoverySwitchingProfile"
},
{
"value": "fad98876-d7ff-11e4-b9d6-1681e6b88ec1",
"key": "SpoofGuardSwitchingProfile"
},
{
"value": "93b4b7e8-f116-415d-a50c-3364611b5d09",
"key": "PortMirroringSwitchingProfile"
},
{
"value": "fbc4fb17-83d9-4b53-a286-ccdf04301888",
"key": "SwitchSecuritySwitchingProfile"
},
{
"value": "f313290b-eba8-4262-bd93-fab5026e9495",
"key": "QosSwitchingProfile"
}
]
}
return FAKE_PORT
def get_logical_port(client, lport_id):
FAKE_SWITCH_UUID = uuidutils.generate_uuid()
FAKE_PORT = {
"id": lport_id,
"display_name": FAKE_NAME,
"resource_type": "LogicalPort",
"address_bindings": [],
"logical_switch_id": FAKE_SWITCH_UUID,
"admin_state": nsx_constants.ADMIN_STATE_UP,
"attachment": {
"id": "9ca8d413-f7bf-4276-b4c9-62f42516bdb2",
"attachment_type": "VIF"
},
"switching_profile_ids": [
{
"value": "64814784-7896-3901-9741-badeff705639",
"key": "IpDiscoverySwitchingProfile"
},
{
"value": "fad98876-d7ff-11e4-b9d6-1681e6b88ec1",
"key": "SpoofGuardSwitchingProfile"
},
{
"value": "93b4b7e8-f116-415d-a50c-3364611b5d09",
"key": "PortMirroringSwitchingProfile"
},
{
"value": "fbc4fb17-83d9-4b53-a286-ccdf04301888",
"key": "SwitchSecuritySwitchingProfile"
},
{
"value": "f313290b-eba8-4262-bd93-fab5026e9495",
"key": "QosSwitchingProfile"
}
]
}
return FAKE_PORT
def update_logical_port(client, lport_id, name=None, admin_state=None):
lport = get_logical_port(client, lport_id)
if name:
lport['display_name'] = name
if admin_state is not None:
if admin_state:
lport['admin_state'] = nsx_constants.ADMIN_STATE_UP
else:
lport['admin_state'] = nsx_constants.ADMIN_STATE_DOWN
return lport
def add_rules_in_section(rules, section_id):
for rule in rules:
rule['id'] = uuidutils.generate_uuid()
return {'rules': rules}
def get_resource(resource):
return {'id': resource.split('/')[-1]}
@ -204,266 +84,6 @@ def delete_resource(resource):
pass
def create_bridge_endpoint(device_name, seg_id, tags):
FAKE_BE = {
"id": uuidutils.generate_uuid(),
"display_name": FAKE_NAME,
"resource_type": "BridgeEndpoint",
"bridge_endpoint_id": device_name,
"vlan": seg_id,
}
return FAKE_BE
class NsxV3Mock(object):
def __init__(self, default_tier0_router_uuid=DEFAULT_TIER0_ROUTER_UUID):
self.logical_routers = {}
self.logical_router_ports = {}
self.logical_ports = {}
self.logical_router_nat_rules = {}
self.static_routes = {}
if default_tier0_router_uuid:
self.create_logical_router(
DEFAULT_TIER0_ROUTER_UUID, None,
edge_cluster_uuid="fake_edge_cluster_uuid",
tier_0=True)
def get_edge_cluster(self, edge_cluster_uuid):
FAKE_CLUSTER = {
"id": edge_cluster_uuid,
"members": [
{"member_index": 0},
{"member_index": 1}]}
return FAKE_CLUSTER
def create_logical_router(self, display_name, tags,
edge_cluster_uuid=None,
tier_0=False):
router_type = (nsx_constants.ROUTER_TYPE_TIER0 if tier_0 else
nsx_constants.ROUTER_TYPE_TIER1)
if display_name == DEFAULT_TIER0_ROUTER_UUID:
fake_router_uuid = DEFAULT_TIER0_ROUTER_UUID
else:
fake_router_uuid = uuidutils.generate_uuid()
result = {'display_name': display_name,
'router_type': router_type,
'tags': tags,
'id': fake_router_uuid}
if edge_cluster_uuid:
result['edge_cluster_id'] = edge_cluster_uuid
self.logical_routers[fake_router_uuid] = result
return result
def get_logical_router(self, lrouter_id):
if lrouter_id in self.logical_routers:
return self.logical_routers[lrouter_id]
else:
raise nsx_exc.ResourceNotFound(manager=FAKE_MANAGER,
operation="get_logical_router")
def update_logical_router(self, lrouter_id, **kwargs):
if lrouter_id in self.logical_routers:
payload = self.logical_routers[lrouter_id]
payload.update(kwargs)
return payload
else:
raise nsx_exc.ResourceNotFound(manager=FAKE_MANAGER,
operation="update_logical_router")
def delete_logical_router(self, lrouter_id):
if lrouter_id in self.logical_routers:
del self.logical_routers[lrouter_id]
else:
raise nsx_exc.ResourceNotFound(manager=FAKE_MANAGER,
operation="delete_logical_router")
def get_logical_router_port_by_ls_id(self, logical_switch_id):
router_ports = []
for router_port in self.logical_router_ports.values():
ls_port_id = router_port.get('linked_logical_switch_port_id')
if ls_port_id:
port = self.get_logical_port(ls_port_id)
if port['logical_switch_id'] == logical_switch_id:
router_ports.append(router_port)
if len(router_ports) >= 2:
raise nsx_exc.NsxPluginException(
err_msg=_("Can't support more than one logical router ports "
"on same logical switch %s ") % logical_switch_id)
elif len(router_ports) == 1:
return router_ports[0]
else:
err_msg = (_("Logical router link port not found on logical "
"switch %s") % logical_switch_id)
raise nsx_exc.ResourceNotFound(manager=FAKE_MANAGER,
operation=err_msg)
def create_logical_port(self, lswitch_id, vif_uuid, tags,
attachment_type=nsx_constants.ATTACHMENT_VIF,
admin_state=True, name=None, address_bindings=None,
parent_name=None, parent_tag=None):
fake_port = create_logical_port(
lswitch_id, vif_uuid, tags,
attachment_type=attachment_type,
admin_state=admin_state, name=name,
address_bindings=address_bindings,
parent_name=parent_name, parent_tag=parent_tag)
fake_port_uuid = fake_port['id']
self.logical_ports[fake_port_uuid] = fake_port
return fake_port
def get_logical_port(self, logical_port_id):
if logical_port_id in self.logical_ports:
return self.logical_ports[logical_port_id]
else:
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="get_logical_port")
def get_logical_router_ports_by_router_id(self, logical_router_id):
logical_router_ports = []
for port_id in self.logical_router_ports.keys():
if (self.logical_router_ports[port_id]['logical_router_id'] ==
logical_router_id):
logical_router_ports.append(self.logical_router_ports[port_id])
return logical_router_ports
def create_logical_router_port(self, logical_router_id,
display_name,
resource_type,
logical_port_id,
address_groups,
edge_cluster_member_index=None):
fake_router_port_uuid = uuidutils.generate_uuid()
body = {'display_name': display_name,
'resource_type': resource_type,
'logical_router_id': logical_router_id}
if address_groups:
body['subnets'] = address_groups
if resource_type in ["LogicalRouterUplinkPort",
"LogicalRouterDownLinkPort"]:
body['linked_logical_switch_port_id'] = logical_port_id
elif logical_port_id:
body['linked_logical_router_port_id'] = logical_port_id
if edge_cluster_member_index:
body['edge_cluster_member_index'] = edge_cluster_member_index
body['id'] = fake_router_port_uuid
self.logical_router_ports[fake_router_port_uuid] = body
return body
def update_logical_router_port(self, logical_port_id, **kwargs):
if logical_port_id in self.logical_router_ports:
payload = self.logical_router_ports[logical_port_id]
payload.update(kwargs)
return payload
else:
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="update_logical_router_port")
def delete_logical_router_port(self, logical_port_id):
if logical_port_id in self.logical_router_ports:
del self.logical_router_ports[logical_port_id]
else:
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="update_logical_router_port")
def add_nat_rule(self, logical_router_id, action, translated_network,
source_net=None, dest_net=None, enabled=True,
rule_priority=None):
fake_rule_id = uuidutils.generate_uuid()
if logical_router_id not in self.logical_routers.keys():
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="get_logical_router")
body = {'action': action,
'enabled': enabled,
'translated_network': translated_network}
if source_net:
body['match_source_network'] = source_net
if dest_net:
body['match_destination_network'] = dest_net
if rule_priority:
body['rule_priority'] = rule_priority
body['rule_id'] = fake_rule_id
if self.logical_router_nat_rules.get(logical_router_id):
self.logical_router_nat_rules[logical_router_id][fake_rule_id] = (
body)
else:
self.logical_router_nat_rules[logical_router_id] = {
fake_rule_id: body}
return body
def delete_nat_rule(self, logical_router_id, nat_rule_id):
if (self.logical_router_nat_rules.get(logical_router_id) and
self.logical_router_nat_rules[logical_router_id].get(nat_rule_id)):
del self.logical_router_nat_rules[logical_router_id][nat_rule_id]
else:
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="delete_nat_rule")
def delete_nat_rule_by_values(self, logical_router_id, **kwargs):
if self.logical_router_nat_rules.get(logical_router_id):
nat_rules = self.logical_router_nat_rules[logical_router_id]
remove_nat_rule_ids = []
for nat_id, nat_body in nat_rules.items():
remove_flag = True
for k, v in kwargs.items():
if nat_body[k] != v:
remove_flag = False
break
if remove_flag:
remove_nat_rule_ids.append(nat_id)
for nat_id in remove_nat_rule_ids:
del nat_rules[nat_id]
def add_static_route(self, logical_router_id, dest_cidr, nexthop):
fake_rule_id = uuidutils.generate_uuid()
if logical_router_id not in self.logical_routers.keys():
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="get_logical_router")
body = {}
if dest_cidr:
body['network'] = dest_cidr
if nexthop:
body['next_hops'] = [{"ip_address": nexthop}]
body['id'] = fake_rule_id
if self.static_routes.get(logical_router_id):
self.static_routes[logical_router_id][fake_rule_id] = body
else:
self.static_routes[logical_router_id] = {fake_rule_id: body}
return body
def delete_static_route(self, logical_router_id, static_route_id):
if (self.static_routes.get(logical_router_id) and
self.static_routes[logical_router_id].get(static_route_id)):
del self.static_routes[logical_router_id][static_route_id]
else:
raise nsx_exc.ResourceNotFound(
manager=FAKE_MANAGER, operation="delete_static_route")
def delete_static_route_by_values(self, logical_router_id,
dest_cidr=None, nexthop=None):
kwargs = {}
if dest_cidr:
kwargs['network'] = dest_cidr
if nexthop:
kwargs['next_hops'] = [{"ip_address": nexthop}]
if self.static_routes.get(logical_router_id):
static_rules = self.static_routes[logical_router_id]
remove_static_rule_ids = []
for rule_id, rule_body in static_rules.items():
remove_flag = True
for k, v in kwargs.items():
if rule_body[k] != v:
remove_flag = False
break
if remove_flag:
remove_static_rule_ids.append(rule_id)
for rule_id in remove_static_rule_ids:
del static_rules[rule_id]
def update_logical_router_advertisement(self, logical_router_id, **kwargs):
# TODO(berlin): implement this latter.
pass
class MockRequestsResponse(object):
def __init__(self, status_code, content=None):
self.status_code = status_code
@ -471,3 +91,151 @@ class MockRequestsResponse(object):
def json(self):
return jsonutils.loads(self.content)
class MockRequestSessionApi(object):
def __init__(self):
self._store = {}
def _is_uuid(self, item):
try:
uuid.UUID(item, version=4)
except ValueError:
return False
return True
def _format_uri(self, uri):
uri = urlparse.urlparse(uri).path
while uri.endswith('/'):
uri = uri[:-1]
while uri.startswith('/'):
uri = uri[1:]
if not self._is_uuid_uri(uri):
uri = "%s/" % uri
return uri
def _is_uuid_uri(self, uri):
return self._is_uuid(urlparse.urlparse(uri).path.split('/')[-1])
def _query(self, search_key, copy=True):
items = []
for uri, obj in self._store.items():
if uri.startswith(search_key):
items.append(obj.copy() if copy else obj)
return items
def _build_response(self, url, content=None,
status=requests.codes.ok, **kwargs):
if urlparse.urlparse(url).path.endswith('/static-routes/'):
content = {
'routes': content or []
}
elif type(content) is list:
content = {
'result_count': len(content),
'results': content
}
if (content is not None and kwargs.get('headers', {}).get(
'Content-Type') == 'application/json'):
content = jsonutils.dumps(content)
return MockRequestsResponse(status, content=content)
def _get_content(self, **kwargs):
content = kwargs.get('data', None)
if content and kwargs.get('headers', {}).get(
'Content-Type') == 'application/json':
content = jsonutils.loads(content)
return content
def get(self, url, **kwargs):
url = self._format_uri(url)
if self._is_uuid_uri(url):
item = self._store.get(url)
code = requests.codes.ok if item else requests.codes.not_found
return self._build_response(
url, content=item, status=code, **kwargs)
return self._build_response(
url, content=self._query(url), status=requests.codes.ok, **kwargs)
def _create(self, url, content, **kwargs):
resource_id = content.get('id', None)
if resource_id and self._store.get("%s%s" % (url, resource_id)):
return self._build_response(
url, content=None, status=requests.codes.bad, **kwargs)
resource_id = resource_id or uuidutils.generate_uuid()
content['id'] = resource_id
if urlparse.urlparse(url).path.endswith('/rules/'):
content['rule_id'] = resource_id
self._store["%s%s" % (url, resource_id)] = content.copy()
return content
def post(self, url, **kwargs):
parsed_url = urlparse.urlparse(url)
url = self._format_uri(url)
if self._is_uuid_uri(url):
if self._store.get(url, None) is None:
return self._build_response(
url, content=None, status=requests.codes.bad, **kwargs)
body = self._get_content(**kwargs)
if body is None:
return self._build_response(
url, content=None, status=requests.codes.bad, **kwargs)
response_content = None
if parsed_url.query and parsed_url.query == 'action=create_multiple':
response_content = {}
for resource_name, resource_body in body.items():
for new_resource in body[resource_name]:
created_resource = self._create(
url, new_resource, **kwargs)
if response_content.get(resource_name, None) is None:
response_content[resource_name] = []
response_content[resource_name].append(created_resource)
else:
response_content = self._create(url, body, **kwargs)
return self._build_response(
url, content=response_content, status=requests.codes.ok, **kwargs)
def put(self, url, **kwargs):
url = self._format_uri(url)
item = {}
if self._is_uuid_uri(url):
item = self._store.get(url, None)
if item is None:
return self._build_response(
url, content=None,
status=requests.codes.not_found, **kwargs)
body = self._get_content(**kwargs)
if body is None:
return self._build_response(
url, content=None, status=requests.codes.bad, **kwargs)
item.update(body)
self._store[url] = item
return self._build_response(
url, content=item, status=requests.codes.ok, **kwargs)
def delete(self, url, **kwargs):
url = self._format_uri(url)
if not self._store.get(url):
return self._build_response(
url, content=None, status=requests.codes.not_found, **kwargs)
del self._store[url]
return self._build_response(
url, content=None, status=requests.codes.ok, **kwargs)

5
vmware_nsx/tests/unit/nsx_v3/test_parent_tag.py
View File

@ -12,14 +12,11 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from neutron.extensions import portbindings as pbin
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_nsx_v3_plugin
class TestParentTagPortBinding(test_nsx_v3_plugin.NsxPluginV3TestCase):
class TestParentTagPortBinding(test_nsx_v3_plugin.NsxV3PluginTestCaseMixin):
# NOTE(arosen): commenting out this test for now for demo setup.
# def test_create_port_with_invalid_parent(self):

174
vmware_nsx/tests/unit/nsx_v3/test_plugin.py
View File

@ -12,9 +12,6 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
from oslo_config import cfg
import six
from neutron.api.v2 import attributes
@ -27,52 +24,65 @@ from neutron.extensions import l3
from neutron.extensions import l3_ext_gw_mode
from neutron.extensions import providernet as pnet
from neutron import manager
import neutron.tests.unit.db.test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.extensions import test_extra_dhcp_opt as test_dhcpopts
import neutron.tests.unit.extensions.test_extraroute as test_ext_route
import neutron.tests.unit.extensions.test_l3 as test_l3_plugin
import neutron.tests.unit.extensions.test_l3_ext_gw_mode as test_ext_gw_mode
import neutron.tests.unit.extensions.test_securitygroup as ext_sg
from neutron.tests.unit.extensions import test_extraroute as test_ext_route
from neutron.tests.unit.extensions import test_l3 as test_l3_plugin
from neutron.tests.unit.extensions \
import test_l3_ext_gw_mode as test_ext_gw_mode
from neutron.tests.unit.extensions import test_securitygroup as ext_sg
from neutron import version
from oslo_config import cfg
from oslo_serialization import jsonutils
from oslo_utils import uuidutils
from vmware_nsx.common import utils
from vmware_nsx.nsxlib import v3 as nsxlib
from vmware_nsx.nsxlib.v3 import dfw_api as firewall
from vmware_nsx.nsxlib.v3 import resources as nsx_resources
from vmware_nsx.nsxlib.v3 import client as nsx_client
from vmware_nsx.plugins.nsx_v3 import plugin as nsx_plugin
from vmware_nsx.tests import unit as vmware
from vmware_nsx.tests.unit.nsx_v3 import mocks as nsx_v3_mocks
PLUGIN_NAME = ('vmware_nsx.plugin.NsxV3Plugin')
from vmware_nsx.tests.unit.nsxlib.v3 import nsxlib_testcase
class NsxPluginV3TestCase(test_plugin.NeutronDbPluginV2TestCase):
PLUGIN_NAME = 'vmware_nsx.plugin.NsxV3Plugin'
def setUp(self,
plugin=PLUGIN_NAME,
class NsxV3PluginTestCaseMixin(test_plugin.NeutronDbPluginV2TestCase,
nsxlib_testcase.NsxClientTestCase):
def setUp(self, plugin=PLUGIN_NAME,
ext_mgr=None,
service_plugins=None):
cfg.CONF.set_override('nsx_manager', '1.2.3.4', 'nsx_v3')
# Mock entire nsxlib methods as this is the best approach to perform
# white-box testing on the plugin class
# TODO(salv-orlando): supply unit tests for nsxlib.v3
nsxlib.create_logical_switch = nsx_v3_mocks.create_logical_switch
nsxlib.delete_logical_switch = mock.Mock()
nsxlib.get_logical_switch = nsx_v3_mocks.get_logical_switch
nsxlib.update_logical_switch = nsx_v3_mocks.update_logical_switch
nsx_resources.LogicalPort.create = nsx_v3_mocks.create_logical_port
nsx_resources.LogicalPort.delete = mock.Mock()
nsx_resources.LogicalPort.get = nsx_v3_mocks.get_logical_port
nsx_resources.LogicalPort.update = nsx_v3_mocks.update_logical_port
firewall.add_rules_in_section = nsx_v3_mocks.add_rules_in_section
firewall.nsxclient.create_resource = nsx_v3_mocks.create_resource
firewall.nsxclient.update_resource = nsx_v3_mocks.update_resource
firewall.nsxclient.get_resource = nsx_v3_mocks.get_resource
firewall.nsxclient.delete_resource = nsx_v3_mocks.delete_resource
self._patchers = []
self.mock_api = nsx_v3_mocks.MockRequestSessionApi()
self.client = nsx_client.NSX3Client()
super(NsxPluginV3TestCase, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
self.v3_mock = nsx_v3_mocks.NsxV3Mock()
nsxlib.get_edge_cluster = self.v3_mock.get_edge_cluster
nsxlib.get_logical_router = self.v3_mock.get_logical_router
mocked = nsxlib_testcase.NsxClientTestCase.mocked_session_module(
nsx_plugin.security.firewall, self.client,
mock_session=self.mock_api)
mocked.start()
self._patchers.append(mocked)
mocked = nsxlib_testcase.NsxClientTestCase.mocked_session_module(
nsx_plugin.routerlib.nsxlib, self.client,
mock_session=self.mock_api)
mocked.start()
self._patchers.append(mocked)
super(NsxV3PluginTestCaseMixin, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
if getattr(self.plugin, '_nsx_client', None):
self.plugin._nsx_client = self.client
if getattr(self.plugin, '_port_client', None):
self.plugin._port_client._client._session = self.mock_api
self.maxDiff = None
def tearDown(self):
for patcher in self._patchers:
patcher.stop()
super(NsxV3PluginTestCaseMixin, self).tearDown()
def _create_network(self, fmt, name, admin_state_up,
arg_list=None, providernet_args=None, **kwargs):
@ -102,40 +112,20 @@ class NsxPluginV3TestCase(test_plugin.NeutronDbPluginV2TestCase):
return network_req.get_response(self.api)
class TestNetworksV2(test_plugin.TestNetworksV2, NsxPluginV3TestCase):
class TestNetworksV2(test_plugin.TestNetworksV2, NsxV3PluginTestCaseMixin):
pass
class TestPortsV2(test_plugin.TestPortsV2, NsxPluginV3TestCase):
class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin):
pass
class SecurityGroupsTestCase(ext_sg.SecurityGroupDBTestCase):
def setUp(self,
plugin=PLUGIN_NAME,
ext_mgr=None):
nsxlib.create_logical_switch = nsx_v3_mocks.create_logical_switch
nsxlib.delete_logical_switch = mock.Mock()
nsx_resources.LogicalPort.create = nsx_v3_mocks.create_logical_port
nsx_resources.LogicalPort.delete = mock.Mock()
nsx_resources.LogicalPort.get = nsx_v3_mocks.get_logical_port
nsx_resources.LogicalPort.update = nsx_v3_mocks.update_logical_port
firewall.add_rules_in_section = nsx_v3_mocks.add_rules_in_section
firewall.nsxclient.create_resource = nsx_v3_mocks.create_resource
firewall.nsxclient.update_resource = nsx_v3_mocks.update_resource
firewall.nsxclient.get_resource = nsx_v3_mocks.get_resource
firewall.nsxclient.delete_resource = nsx_v3_mocks.delete_resource
super(SecurityGroupsTestCase, self).setUp(plugin=PLUGIN_NAME,
ext_mgr=ext_mgr)
class TestSecurityGroups(ext_sg.TestSecurityGroups, SecurityGroupsTestCase):
class TestSecurityGroups(ext_sg.TestSecurityGroups, NsxV3PluginTestCaseMixin):
pass
class DHCPOptsTestCase(test_dhcpopts.TestExtraDhcpOpt, NsxPluginV3TestCase):
class DHCPOptsTestCase(test_dhcpopts.TestExtraDhcpOpt,
NsxV3PluginTestCaseMixin):
def setUp(self, plugin=None):
super(test_dhcpopts.ExtraDhcpOptDBTestCase, self).setUp(
@ -175,7 +165,7 @@ def restore_l3_attribute_map(map_to_restore):
l3.RESOURCE_ATTRIBUTE_MAP = map_to_restore
class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxPluginV3TestCase):
class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxV3PluginTestCaseMixin):
def _restore_l3_attribute_map(self):
l3.RESOURCE_ATTRIBUTE_MAP = self._l3_attribute_map_bk
@ -194,30 +184,6 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxPluginV3TestCase):
self.plugin_instance.__module__,
self.plugin_instance.__class__.__name__)
self._plugin_class = self.plugin_instance.__class__
nsx_resources.LogicalPort.create = self.v3_mock.create_logical_port
nsxlib.create_logical_router = self.v3_mock.create_logical_router
nsxlib.update_logical_router = self.v3_mock.update_logical_router
nsxlib.delete_logical_router = self.v3_mock.delete_logical_router
nsxlib.get_logical_router_port_by_ls_id = (
self.v3_mock.get_logical_router_port_by_ls_id)
nsxlib.create_logical_router_port = (
self.v3_mock.create_logical_router_port)
nsxlib.update_logical_router_port = (
self.v3_mock.update_logical_router_port)
nsxlib.delete_logical_router_port = (
self.v3_mock.delete_logical_router_port)
nsxlib.add_nat_rule = self.v3_mock.add_nat_rule
nsxlib.delete_nat_rule = self.v3_mock.delete_nat_rule
nsxlib.delete_nat_rule_by_values = (
self.v3_mock.delete_nat_rule_by_values)
nsxlib.get_logical_router_ports_by_router_id = (
self.v3_mock.get_logical_router_ports_by_router_id)
nsxlib.update_logical_router_advertisement = (
self.v3_mock.update_logical_router_advertisement)
nsxlib.add_static_route = self.v3_mock.add_static_route
nsxlib.delete_static_route = self.v3_mock.delete_static_route
nsxlib.delete_static_route_by_values = (
self.v3_mock.delete_static_route_by_values)
def _create_l3_ext_network(
self, physical_network=nsx_v3_mocks.DEFAULT_TIER0_ROUTER_UUID):
@ -234,11 +200,37 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxPluginV3TestCase):
class TestL3NatTestCase(L3NatTest,
test_l3_plugin.L3NatDBIntTestCase,
NsxPluginV3TestCase,
NsxV3PluginTestCaseMixin,
test_ext_route.ExtraRouteDBTestCaseBase):
def setUp(self, plugin=PLUGIN_NAME,
ext_mgr=None,
service_plugins=None):
super(TestL3NatTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
cluster_id = uuidutils.generate_uuid()
self.mock_api.post(
'api/v1/logical-routers',
data=jsonutils.dumps({
'display_name': nsx_v3_mocks.DEFAULT_TIER0_ROUTER_UUID,
'router_type': "TIER0",
'id': nsx_v3_mocks.DEFAULT_TIER0_ROUTER_UUID,
'edge_cluster_id': cluster_id}),
headers=nsx_client.JSONRESTClient._DEFAULT_HEADERS)
self.mock_api.post(
'api/v1/edge-clusters',
data=jsonutils.dumps({
'id': cluster_id,
'members': [
{'member_index': 0},
{'member_index': 1}
]}),
headers=nsx_client.JSONRESTClient._DEFAULT_HEADERS)
def _test_create_l3_ext_network(
self, physical_network=nsx_v3_mocks.DEFAULT_TIER0_ROUTER_UUID):
self, physical_network=nsx_v3_mocks.DEFAULT_TIER0_ROUTER_UUID):
name = 'l3_ext_net'
net_type = utils.NetworkTypes.L3_EXT
expected = [('subnets', []), ('name', name), ('admin_state_up', True),
@ -324,7 +316,7 @@ class ExtGwModeTestCase(L3NatTest,
pass
class TestNsxV3Utils(NsxPluginV3TestCase):
class TestNsxV3Utils(NsxV3PluginTestCaseMixin):
def test_build_v3_tags_payload(self):
result = utils.build_v3_tags_payload(

140
vmware_nsx/tests/unit/nsxlib/v3/nsxlib_testcase.py
View File

@ -13,13 +13,143 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import contextlib
import mock
import types
import unittest
from oslo_config import cfg
import unittest
from vmware_nsx.nsxlib.v3 import client as nsx_client
from vmware_nsx.tests.unit.nsx_v3 import mocks
NSX_USER = 'admin'
NSX_PASSWORD = 'default'
NSX_MANAGER = '1.2.3.4'
NSX_INSECURE = True
NSX_CERT = '/opt/stack/certs/nsx.pem'
V3_CLIENT_PKG = 'vmware_nsx.nsxlib.v3.client'
BRIDGE_FNS = ['create_resource', 'delete_resource',
'update_resource', 'get_resource']
class NsxLibTestCase(unittest.TestCase):
def setUp(self):
cfg.CONF.set_override('nsx_user', 'admin')
cfg.CONF.set_override('nsx_password', 'default')
cfg.CONF.set_override('nsx_manager', '1.2.3.4', 'nsx_v3')
def setUp(self, *args, **kwargs):
super(NsxLibTestCase, self).setUp()
cfg.CONF.set_override('nsx_user', NSX_USER)
cfg.CONF.set_override('nsx_password', NSX_PASSWORD)
cfg.CONF.set_override('nsx_user', NSX_USER, 'nsx_v3')
cfg.CONF.set_override('nsx_password', NSX_PASSWORD, 'nsx_v3')
cfg.CONF.set_override('nsx_manager', NSX_MANAGER, 'nsx_v3')
cfg.CONF.set_override('insecure', NSX_INSECURE, 'nsx_v3')
cfg.CONF.set_override('ca_file', NSX_CERT, 'nsx_v3')
# print diffs when assert comparisons fail
self.maxDiff = None
# NOTE(boden): a lot of the hackery and magic below can be removed
# once we move all v3 rest function calls to OO based on rest resource
class NsxClientTestCase(NsxLibTestCase):
class MockBridge(object):
def __init__(self, api_client):
self._client = api_client
def get_resource(self, resource):
return nsx_client.get_resource(
resource, client=self._client)
def create_resource(self, resource, data):
return nsx_client.create_resource(
resource, data, client=self._client)
def delete_resource(self, resource):
return nsx_client.delete_resource(
resource, client=self._client)
def update_resource(self, resource, data):
return nsx_client.create_resource(
resource, data, client=self._client)
def new_client(
self, clazz, host_ip=NSX_MANAGER,
user_name=NSX_USER,
password=NSX_PASSWORD,
insecure=NSX_INSECURE,
url_prefix=None,
default_headers=None,
cert_file=NSX_CERT):
return clazz(host_ip=host_ip, user_name=user_name,
password=password, insecure=insecure,
url_prefix=url_prefix, default_headers=default_headers,
cert_file=cert_file)
@contextlib.contextmanager
def mocked_client(self, client, mock_validate=True):
session = client._session
with mock.patch.object(session, 'get')as _get:
with mock.patch.object(session, 'post')as _post:
with mock.patch.object(session, 'delete')as _delete:
with mock.patch.object(session, 'put')as _put:
rep = {
'get': _get,
'put': _put,
'delete': _delete,
'post': _post
}
if mock_validate:
with mock.patch.object(client, '_validate_result'):
yield rep
else:
yield rep
@contextlib.contextmanager
def mocked_resource(self, resource, mock_validate=True):
with self.mocked_client(resource._client,
mock_validate=mock_validate) as _client:
yield _client
@contextlib.contextmanager
def mocked_client_bridge(self, client, module, attr, mock_validate=True):
mocked_bridge = NsxClientTestCase.MockBridge(client)
mocked_bridge.JSONRESTClient = nsx_client.JSONRESTClient
with self.mocked_client(client, mock_validate=mock_validate) as mocked:
with mock.patch.object(module, attr, new=mocked_bridge):
yield mocked
@classmethod
def patch_client_module(cls, in_module, fn_map):
mock_client = mock.Mock()
for name, clazz in in_module.__dict__.items():
if (isinstance(clazz, types.ModuleType) and
clazz.__name__ == V3_CLIENT_PKG):
for fn_name in BRIDGE_FNS:
mock_call = fn_map.get(fn_name, getattr(mocks, fn_name))
setattr(mock_client, fn_name, mock_call)
for fn_name, fn_call in fn_map.items():
if fn_name not in BRIDGE_FNS:
setattr(mock_client, fn_name, fn_call)
return mock.patch.object(in_module, name, new=mock_client)
return None
@classmethod
def mocked_session_module(cls, in_module, with_client,
mock_session=None):
mock_session = mock_session or mocks.MockRequestSessionApi()
with_client._session = mock_session
def _call_client(fn_name):
def _client(*args, **kwargs):
client_fn = getattr(nsx_client, fn_name)
kwargs['client'] = with_client
return client_fn(*args, **kwargs)
return _client
fn_map = {}
for fn in BRIDGE_FNS:
fn_map[fn] = _call_client(fn)
fn_map['NSX3Client'] = nsx_client.NSX3Client
return cls.patch_client_module(in_module, fn_map)

476
vmware_nsx/tests/unit/nsxlib/v3/test_client.py
View File

@ -13,365 +13,317 @@
# License for the specific language governing permissions and limitations
# under the License.
#
import mock
import vmware_nsx.common.exceptions as exep
from oslo_config import cfg
from oslo_log import log
from oslo_serialization import jsonutils
from vmware_nsx.common import exceptions as exep
from vmware_nsx.nsxlib.v3 import client
from vmware_nsx.tests.unit.nsx_v3 import mocks
from vmware_nsx.tests.unit.nsxlib.v3 import nsxlib_testcase
LOG = log.getLogger(__name__)
CLIENT_PKG = 'vmware_nsx.nsxlib.v3.client'
# Cache the pointers to those functions because they are overridden
# by tests/unit/nsx_v3/test_plugin.py. This is only needed when running
# tox on multiple tests.
client_get_resource = client.get_resource
client_create_resource = client.create_resource
client_update_resource = client.update_resource
client_delete_resource = client.delete_resource
def assert_session_call(mock_call, url, verify, data, headers, cert):
mock_call.assert_called_once_with(
url, verify=verify, data=data, headers=headers, cert=cert)
class BaseClientTestCase(nsxlib_testcase.NsxLibTestCase):
nsx_manager = '1.2.3.4'
nsx_user = 'testuser'
nsx_password = 'pass123'
ca_file = '/path/to/ca.pem'
insecure = True
def setUp(self):
cfg.CONF.set_override(
'nsx_manager', BaseClientTestCase.nsx_manager, 'nsx_v3')
cfg.CONF.set_override(
'nsx_user', BaseClientTestCase.nsx_user, 'nsx_v3')
cfg.CONF.set_override(
'nsx_password', BaseClientTestCase.nsx_password, 'nsx_v3')
cfg.CONF.set_override(
'ca_file', BaseClientTestCase.ca_file, 'nsx_v3')
cfg.CONF.set_override(
'insecure', BaseClientTestCase.insecure, 'nsx_v3')
super(BaseClientTestCase, self).setUp()
def new_client(
self, clazz, host_ip=nsx_manager,
user_name=nsx_user, password=nsx_password,
insecure=insecure, url_prefix=None,
default_headers=None, cert_file=ca_file):
return clazz(host_ip=host_ip, user_name=user_name,
password=password, insecure=insecure,
url_prefix=url_prefix, default_headers=default_headers,
cert_file=cert_file)
class NsxV3RESTClientTestCase(BaseClientTestCase):
class NsxV3RESTClientTestCase(nsxlib_testcase.NsxClientTestCase):
def test_client_conf_init(self):
api = self.new_client(client.RESTClient)
self.assertEqual((
BaseClientTestCase.nsx_user, BaseClientTestCase.nsx_password),
nsxlib_testcase.NSX_USER, nsxlib_testcase.NSX_PASSWORD),
api._session.auth)
self.assertEqual(BaseClientTestCase.nsx_manager, api._host_ip)
self.assertEqual(BaseClientTestCase.ca_file, api._cert_file)
self.assertEqual(nsxlib_testcase.NSX_MANAGER, api._host_ip)
self.assertEqual(nsxlib_testcase.NSX_CERT, api._cert_file)
def test_client_params_init(self):
api = self.new_client(
client.RESTClient, host_ip='11.12.13.14', password='mypass')
self.assertEqual((
BaseClientTestCase.nsx_user, 'mypass'),
nsxlib_testcase.NSX_USER, 'mypass'),
api._session.auth)
self.assertEqual('11.12.13.14', api._host_ip)
self.assertEqual(BaseClientTestCase.ca_file, api._cert_file)
self.assertEqual(nsxlib_testcase.NSX_CERT, api._cert_file)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_url_prefix(self, mock_validate, mock_get):
mock_get.return_value = {}
def test_client_url_prefix(self):
api = self.new_client(client.RESTClient, url_prefix='/cloud/api')
api.list()
with self.mocked_client(api) as mocked:
mock_get = mocked.get('get')
mock_get.return_value = {}
api.list()
assert_session_call(
mock_get,
'https://1.2.3.4/cloud/api',
False, None, {}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/cloud/api',
False, None, {}, nsxlib_testcase.NSX_CERT)
mock_get.reset_mock()
mock_get.reset_mock()
api.url_list('v1/ports')
assert_session_call(
mock_get,
'https://1.2.3.4/cloud/api/v1/ports', False, None, {},
BaseClientTestCase.ca_file)
api.url_list('v1/ports')
assert_session_call(
mock_get,
'https://1.2.3.4/cloud/api/v1/ports', False, None, {},
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_headers(self, mock_validate, mock_get):
def test_client_headers(self):
default_headers = {'Content-Type': 'application/golang'}
mock_get.return_value = {}
api = self.new_client(
client.RESTClient, default_headers=default_headers,
url_prefix='/v1/api')
api.list()
assert_session_call(
mock_get,
'https://1.2.3.4/v1/api',
False, None, default_headers, BaseClientTestCase.ca_file)
with self.mocked_client(api) as mocked:
mock_get = mocked.get('get')
mock_get.reset_mock()
mock_get.return_value = {}
method_headers = {'X-API-Key': 'strong-crypt'}
api.url_list('ports/33', headers=method_headers)
method_headers.update(default_headers)
assert_session_call(
mock_get,
'https://1.2.3.4/v1/api/ports/33', False, None,
method_headers,
BaseClientTestCase.ca_file)
api.list()
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_for(self, mock_validate, mock_get):
assert_session_call(
mock_get,
'https://1.2.3.4/v1/api',
False, None, default_headers, nsxlib_testcase.NSX_CERT)
mock_get.reset_mock()
method_headers = {'X-API-Key': 'strong-crypt'}
api.url_list('ports/33', headers=method_headers)
method_headers.update(default_headers)
assert_session_call(
mock_get,
'https://1.2.3.4/v1/api/ports/33', False, None,
method_headers,
nsxlib_testcase.NSX_CERT)
def test_client_for(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/')
sub_api = api.new_client_for('switch/ports')
sub_api.get('11a2b')
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/switch/ports/11a2b',
False, None, {}, BaseClientTestCase.ca_file)
with self.mocked_client(sub_api) as mocked:
sub_api.get('11a2b')
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_list(self, mock_validate, mock_get):
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/switch/ports/11a2b',
False, None, {}, nsxlib_testcase.NSX_CERT)
def test_client_list(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.list()
with self.mocked_client(api) as mocked:
api.list()
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/ports',
False, None, {}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/ports',
False, None, {}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_get(self, mock_validate, mock_get):
def test_client_get(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.get('unique-id')
with self.mocked_client(api) as mocked:
api.get('unique-id')
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/ports/unique-id',
False, None, {}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/ports/unique-id',
False, None, {}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.delete'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_delete(self, mock_validate, mock_delete):
def test_client_delete(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.delete('unique-id')
with self.mocked_client(api) as mocked:
api.delete('unique-id')
assert_session_call(
mock_delete,
'https://1.2.3.4/api/v1/ports/unique-id',
False, None, {}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('delete'),
'https://1.2.3.4/api/v1/ports/unique-id',
False, None, {}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.put'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_update(self, mock_validate, mock_put):
def test_client_update(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.update('unique-id', {'name': 'a-new-name'})
with self.mocked_client(api) as mocked:
api.update('unique-id', {'name': 'a-new-name'})
assert_session_call(
mock_put,
'https://1.2.3.4/api/v1/ports/unique-id',
False, {'name': 'a-new-name'},
{}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('put'),
'https://1.2.3.4/api/v1/ports/unique-id',
False, {'name': 'a-new-name'},
{}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_create(self, mock_validate, mock_post):
def test_client_create(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.create({'resource-name': 'port1'})
with self.mocked_client(api) as mocked:
api.create({'resource-name': 'port1'})
assert_session_call(
mock_post,
'https://1.2.3.4/api/v1/ports',
False, {'resource-name': 'port1'},
{}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/ports',
False, {'resource-name': 'port1'},
{}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_url_list(self, mock_validate, mock_get):
def test_client_url_list(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.url_list('/connections', {'Content-Type': 'application/json'})
with self.mocked_client(api) as mocked:
api.url_list('/connections', {'Content-Type': 'application/json'})
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/ports/connections',
False, None,
{'Content-Type': 'application/json'},
BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/ports/connections',
False, None,
{'Content-Type': 'application/json'},
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_url_get(self, mock_validate, mock_get):
def test_client_url_get(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.url_get('connections/1')
with self.mocked_client(api) as mocked:
api.url_get('connections/1')
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/ports/connections/1',
False, None, {}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/ports/connections/1',
False, None, {}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.delete'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_url_delete(self, mock_validate, mock_delete):
def test_client_url_delete(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.url_delete('1')
with self.mocked_client(api) as mocked:
api.url_delete('1')
assert_session_call(
mock_delete,
'https://1.2.3.4/api/v1/ports/1',
False, None, {}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('delete'),
'https://1.2.3.4/api/v1/ports/1',
False, None, {}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.put'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_url_put(self, mock_validate, mock_put):
def test_client_url_put(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.url_put('connections/1', {'name': 'conn1'})
with self.mocked_client(api) as mocked:
api.url_put('connections/1', {'name': 'conn1'})
assert_session_call(
mock_put,
'https://1.2.3.4/api/v1/ports/connections/1',
False, {'name': 'conn1'},
{}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('put'),
'https://1.2.3.4/api/v1/ports/connections/1',
False, {'name': 'conn1'},
{}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_client_url_post(self, mock_validate, mock_post):
def test_client_url_post(self):
api = self.new_client(client.RESTClient, url_prefix='api/v1/ports')
api.url_post('1/connections', {'name': 'conn1'})
with self.mocked_client(api) as mocked:
api.url_post('1/connections', {'name': 'conn1'})
assert_session_call(
mock_post,
'https://1.2.3.4/api/v1/ports/1/connections',
False, {'name': 'conn1'},
{}, BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/ports/1/connections',
False, {'name': 'conn1'},
{}, nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.put'))
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.delete'))
def test_client_validate_result(self, *args):
def test_client_validate_result(self):
def _verb_response_code(http_verb, status_code):
response = mocks.MockRequestsResponse(status_code, None)
api = self.new_client(client.RESTClient)
for mocked in args:
mocked.return_value = response
client_call = getattr(api, "url_%s" % http_verb)
client_call('', None)
api = self.new_client(client.RESTClient)
with self.mocked_client(api, mock_validate=False) as mocked:
def _verb_response_code(http_verb, status_code):
response = mocks.MockRequestsResponse(
status_code, None)
for _verb in ['get', 'post', 'put', 'delete']:
mocked.get(_verb).return_value = response
client_call = getattr(api, "url_%s" % http_verb)
client_call('', None)
for verb in ['get', 'post', 'put', 'delete']:
for code in client.RESTClient._VERB_RESP_CODES.get(verb):
_verb_response_code(verb, code)
self.assertRaises(
exep.ManagerError, _verb_response_code, verb, 500)
for verb in ['get', 'post', 'put', 'delete']:
for code in client.RESTClient._VERB_RESP_CODES.get(
verb):
_verb_response_code(verb, code)
self.assertRaises(
exep.ManagerError,
_verb_response_code, verb, 500)
class NsxV3JSONClientTestCase(BaseClientTestCase):
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_json_request(self, mock_validate, mock_post):
mock_post.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps({'result': {'ok': 200}}))
class NsxV3JSONClientTestCase(nsxlib_testcase.NsxClientTestCase):
def test_json_request(self):
api = self.new_client(client.JSONRESTClient, url_prefix='api/v2/nat')
resp = api.create(body={'name': 'mgmt-egress'})
with self.mocked_client(api) as mocked:
mock_post = mocked.get('post')
mock_post.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps({'result': {'ok': 200}}))
assert_session_call(
mock_post,
'https://1.2.3.4/api/v2/nat',
False, jsonutils.dumps({'name': 'mgmt-egress'}),
client.JSONRESTClient._DEFAULT_HEADERS,
BaseClientTestCase.ca_file)
resp = api.create(body={'name': 'mgmt-egress'})
self.assertEqual(resp, {'result': {'ok': 200}})
assert_session_call(
mock_post,
'https://1.2.3.4/api/v2/nat',
False, jsonutils.dumps({'name': 'mgmt-egress'}),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
self.assertEqual(resp, {'result': {'ok': 200}})
class NsxV3APIClientTestCase(BaseClientTestCase):
class NsxV3APIClientTestCase(nsxlib_testcase.NsxClientTestCase):
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_api_call(self, mock_validate, mock_get):
def test_api_call(self):
api = self.new_client(client.NSX3Client)
api.get('ports')
with self.mocked_client(api) as mocked:
api.get('ports')
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/ports',
False, None,
client.JSONRESTClient._DEFAULT_HEADERS,
NsxV3APIClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/ports',
False, None,
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
# NOTE(boden): remove this when tmp brigding removed
class NsxV3APIClientBridgeTestCase(BaseClientTestCase):
class NsxV3APIClientBridgeTestCase(nsxlib_testcase.NsxClientTestCase):
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_get_resource(self, mock_validate, mock_get):
client_get_resource('ports')
def test_get_resource(self):
api = self.new_client(client.NSX3Client)
with self.mocked_client(api) as mocked:
client.get_resource('ports', client=api)
assert_session_call(
mock_get,
'https://1.2.3.4/api/v1/ports',
False, None,
client.JSONRESTClient._DEFAULT_HEADERS,
NsxV3APIClientTestCase.ca_file)
assert_session_call(
mocked.get('get'),
'https://1.2.3.4/api/v1/ports',
False, None,
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_create_resource(self, mock_validate, mock_post):
client_create_resource('ports', {'resource-name': 'port1'})
def test_create_resource(self):
api = self.new_client(client.NSX3Client)
with self.mocked_client(api) as mocked:
client.create_resource(
'ports', {'resource-name': 'port1'},
client=api)
assert_session_call(
mock_post,
'https://1.2.3.4/api/v1/ports',
False, jsonutils.dumps({'resource-name': 'port1'}),
client.JSONRESTClient._DEFAULT_HEADERS,
BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/ports',
False, jsonutils.dumps({'resource-name': 'port1'}),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.put'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_update_resource(self, mock_validate, mock_put):
client_update_resource('ports/1', {'name': 'a-new-name'})
def test_update_resource(self):
api = self.new_client(client.NSX3Client)
with self.mocked_client(api) as mocked:
client.update_resource(
'ports/1', {'name': 'a-new-name'}, client=api)
assert_session_call(
mock_put,
'https://1.2.3.4/api/v1/ports/1',
False, jsonutils.dumps({'name': 'a-new-name'}),
client.JSONRESTClient._DEFAULT_HEADERS,
BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('put'),
'https://1.2.3.4/api/v1/ports/1',
False, jsonutils.dumps({'name': 'a-new-name'}),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.delete'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_delete_resource(self, mock_validate, mock_delete):
client_delete_resource('ports/11')
def test_delete_resource(self):
api = self.new_client(client.NSX3Client)
with self.mocked_client(api) as mocked:
client.delete_resource('ports/11', client=api)
assert_session_call(
mock_delete,
'https://1.2.3.4/api/v1/ports/11',
False, None, client.JSONRESTClient._DEFAULT_HEADERS,
BaseClientTestCase.ca_file)
assert_session_call(
mocked.get('delete'),
'https://1.2.3.4/api/v1/ports/11',
False, None, client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)

97
vmware_nsx/tests/unit/nsxlib/v3/test_qos_switching_profile.py
View File

@ -13,63 +13,86 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import mock
from oslo_log import log
from oslo_serialization import jsonutils
from vmware_nsx.nsxlib import v3 as nsxlib
from vmware_nsx.nsxlib.v3 import client
from vmware_nsx.tests.unit.nsx_v3 import test_constants as test_constants_v3
from vmware_nsx.tests.unit.nsxlib.v3 import nsxlib_testcase
from vmware_nsx.tests.unit.nsxlib.v3 import test_client
LOG = log.getLogger(__name__)
_JSON_HEADERS = client.JSONRESTClient._DEFAULT_HEADERS
class NsxLibQosTestCase(nsxlib_testcase.NsxLibTestCase):
class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase):
@mock.patch("vmware_nsx.nsxlib.v3"
".client.create_resource")
def test_create_qos_switching_profile_untrusted(
self, mock_create_resource):
def _body(self, qos_marking=None, dscp=None):
body = {
"resource_type": "QosSwitchingProfile",
"tags": []
}
if qos_marking:
body["dscp"] = {}
body["dscp"]["mode"] = qos_marking.upper()
if dscp:
body["dscp"]["priority"] = dscp
body["display_name"] = test_constants_v3.FAKE_NAME
body["description"] = test_constants_v3.FAKE_NAME
return body
def test_create_qos_switching_profile_untrusted(self):
"""
Test creating a qos-switching profile returns the correct response
"""
fake_qos_profile = test_constants_v3.FAKE_QOS_PROFILE
fake_qos_profile["dscp"]["mode"] = "UNTRUSTED"
fake_qos_profile["dscp"]["priority"] = 25
mock_create_resource.return_value = fake_qos_profile
api = self.new_client(client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
nsxlib.create_qos_switching_profile(
qos_marking="untrusted", dscp=25, tags=[],
name=test_constants_v3.FAKE_NAME,
description=test_constants_v3.FAKE_NAME)
result = nsxlib.create_qos_switching_profile(
qos_marking="untrusted", dscp=25, tags=[],
name=test_constants_v3.FAKE_NAME,
description=test_constants_v3.FAKE_NAME)
self.assertEqual(fake_qos_profile, result)
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/switching-profiles',
False,
jsonutils.dumps(self._body(qos_marking='UNTRUSTED', dscp=25)),
_JSON_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("vmware_nsx.nsxlib.v3"
".client.create_resource")
def test_create_qos_switching_profile_trusted(
self, mock_create_resource):
def test_create_qos_switching_profile_trusted(self):
"""
Test creating a qos-switching profile returns the correct response
"""
fake_qos_profile = test_constants_v3.FAKE_QOS_PROFILE
fake_qos_profile["dscp"]["mode"] = "TRUSTED"
fake_qos_profile["dscp"]["priority"] = 0
mock_create_resource.return_value = fake_qos_profile
api = self.new_client(client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
nsxlib.create_qos_switching_profile(
qos_marking="trusted", dscp=0, tags=[],
name=test_constants_v3.FAKE_NAME,
description=test_constants_v3.FAKE_NAME)
result = nsxlib.create_qos_switching_profile(
qos_marking="trusted", dscp=0, tags=[],
name=test_constants_v3.FAKE_NAME,
description=test_constants_v3.FAKE_NAME)
self.assertEqual(fake_qos_profile, result)
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/switching-profiles',
False,
jsonutils.dumps(self._body(qos_marking='trusted', dscp=0)),
_JSON_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("vmware_nsx.nsxlib.v3"
".client.delete_resource")
def test_delete_qos_switching_profile(self, mock_delete_resource):
def test_delete_qos_switching_profile(self):
"""
Test deleting qos-switching-profile
"""
mock_delete_resource.return_value = None
result = nsxlib.delete_qos_switching_profile(
test_constants_v3.FAKE_QOS_PROFILE['id'])
self.assertIsNone(result)
api = self.new_client(client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
nsxlib.delete_qos_switching_profile(
test_constants_v3.FAKE_QOS_PROFILE['id'])
test_client.assert_session_call(
mocked.get('delete'),
'https://1.2.3.4/api/v1/switching-profiles/%s'
% test_constants_v3.FAKE_QOS_PROFILE['id'],
False, None,
_JSON_HEADERS,
nsxlib_testcase.NSX_CERT)

271
vmware_nsx/tests/unit/nsxlib/v3/test_resources.py
View File

@ -13,13 +13,13 @@
# License for the specific language governing permissions and limitations
# under the License.
#
import mock
from oslo_serialization import jsonutils
from vmware_nsx.nsxlib.v3 import client
from vmware_nsx.nsxlib.v3 import resources
from vmware_nsx.tests.unit.nsx_v3 import mocks
from vmware_nsx.tests.unit.nsx_v3 import test_constants as test_constants_v3
from vmware_nsx.tests.unit.nsxlib.v3 import nsxlib_testcase
from vmware_nsx.tests.unit.nsxlib.v3 import test_client
@ -27,29 +27,26 @@ CLIENT_PKG = test_client.CLIENT_PKG
profile_types = resources.SwitchingProfileTypes
class TestSwitchingProfileTestCase(test_client.BaseClientTestCase):
class TestSwitchingProfileTestCase(nsxlib_testcase.NsxClientTestCase):
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_switching_profile_create(self, mock_validate, mock_post):
def test_switching_profile_create(self):
api = resources.SwitchingProfile(client.NSX3Client())
api.create(profile_types.PORT_MIRRORING,
'pm-profile', 'port mirror prof')
with self.mocked_resource(api) as mocked:
api.create(profile_types.PORT_MIRRORING,
'pm-profile', 'port mirror prof')
test_client.assert_session_call(
mock_post,
'https://1.2.3.4/api/v1/switching-profiles',
False, jsonutils.dumps({
'resource_type': profile_types.PORT_MIRRORING,
'display_name': 'pm-profile',
'description': 'port mirror prof'
}),
client.JSONRESTClient._DEFAULT_HEADERS,
test_client.BaseClientTestCase.ca_file)
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/switching-profiles',
False, jsonutils.dumps({
'resource_type': profile_types.PORT_MIRRORING,
'display_name': 'pm-profile',
'description': 'port mirror prof'
}),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.put'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_switching_profile_update(self, mock_validate, mock_put):
def test_switching_profile_update(self):
tags = [
{
@ -63,21 +60,20 @@ class TestSwitchingProfileTestCase(test_client.BaseClientTestCase):
]
api = resources.SwitchingProfile(client.NSX3Client())
api.update('a12bc1', profile_types.PORT_MIRRORING, tags=tags)
with self.mocked_resource(api) as mocked:
api.update('a12bc1', profile_types.PORT_MIRRORING, tags=tags)
test_client.assert_session_call(
mock_put,
'https://1.2.3.4/api/v1/switching-profiles/a12bc1',
False, jsonutils.dumps({
'resource_type': profile_types.PORT_MIRRORING,
'tags': tags
}),
client.JSONRESTClient._DEFAULT_HEADERS,
test_client.BaseClientTestCase.ca_file)
test_client.assert_session_call(
mocked.get('put'),
'https://1.2.3.4/api/v1/switching-profiles/a12bc1',
False, jsonutils.dumps({
'resource_type': profile_types.PORT_MIRRORING,
'tags': tags
}),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_spoofgaurd_profile_create(self, mock_validate, mock_post):
def test_spoofgaurd_profile_create(self):
tags = [
{
@ -91,27 +87,26 @@ class TestSwitchingProfileTestCase(test_client.BaseClientTestCase):
]
api = resources.SwitchingProfile(client.NSX3Client())
api.create_spoofguard_profile(
'neutron-spoof', 'spoofguard-for-neutron',
whitelist_ports=True, tags=tags)
with self.mocked_resource(api) as mocked:
api.create_spoofguard_profile(
'neutron-spoof', 'spoofguard-for-neutron',
whitelist_ports=True, tags=tags)
test_client.assert_session_call(
mock_post,
'https://1.2.3.4/api/v1/switching-profiles',
False,
jsonutils.dumps({
'resource_type': profile_types.SPOOF_GUARD,
'display_name': 'neutron-spoof',
'description': 'spoofguard-for-neutron',
'white_list_providers': ['LPORT_BINDINGS'],
'tags': tags
}),
client.JSONRESTClient._DEFAULT_HEADERS,
test_client.BaseClientTestCase.ca_file)
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/switching-profiles',
False,
jsonutils.dumps({
'resource_type': profile_types.SPOOF_GUARD,
'display_name': 'neutron-spoof',
'description': 'spoofguard-for-neutron',
'white_list_providers': ['LPORT_BINDINGS'],
'tags': tags
}),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.get'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_find_by_display_name(self, mock_validate, mock_get):
def test_find_by_display_name(self):
resp_resources = {
'results': [
{'display_name': 'resource-1'},
@ -119,123 +114,113 @@ class TestSwitchingProfileTestCase(test_client.BaseClientTestCase):
{'display_name': 'resource-3'}
]
}
mock_get.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources))
api = resources.SwitchingProfile(client.NSX3Client())
self.assertEqual([{'display_name': 'resource-1'}],
api.find_by_display_name('resource-1'))
self.assertEqual([{'display_name': 'resource-2'}],
api.find_by_display_name('resource-2'))
self.assertEqual([{'display_name': 'resource-3'}],
api.find_by_display_name('resource-3'))
with self.mocked_resource(api) as mocked:
mock_get = mocked.get('get')
mock_get.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources))
self.assertEqual([{'display_name': 'resource-1'}],
api.find_by_display_name('resource-1'))
self.assertEqual([{'display_name': 'resource-2'}],
api.find_by_display_name('resource-2'))
self.assertEqual([{'display_name': 'resource-3'}],
api.find_by_display_name('resource-3'))
mock_get.reset_mock()
mock_get.reset_mock()
resp_resources = {
'results': [
{'display_name': 'resource-1'},
{'display_name': 'resource-1'},
{'display_name': 'resource-1'}
]
}
mock_get.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources))
api = resources.SwitchingProfile(client.NSX3Client())
self.assertEqual(resp_resources['results'],
api.find_by_display_name('resource-1'))
resp_resources = {
'results': [
{'display_name': 'resource-1'},
{'display_name': 'resource-1'},
{'display_name': 'resource-1'}
]
}
mock_get.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(resp_resources))
self.assertEqual(resp_resources['results'],
api.find_by_display_name('resource-1'))
class LogicalPortTestCase(test_client.BaseClientTestCase):
class LogicalPortTestCase(nsxlib_testcase.NsxClientTestCase):
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_create_logical_port(self, mock_validate, mock_post):
def test_create_logical_port(self):
"""
Test creating a port returns the correct response and 200 status
"""
fake_port = test_constants_v3.FAKE_PORT
mock_post.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(fake_port))
fake_port = test_constants_v3.FAKE_PORT.copy()
profile_client = resources.SwitchingProfile(client.NSX3Client())
profile_dicts = []
for profile_id in fake_port['switching_profile_ids']:
profile_dicts.append({'resource_type': profile_id['key'],
'id': profile_id['value']})
# Reload resources because resources.LogicalPort.create is overridden
# by tests/unit/nsx_v3/test_plugin.py. This is only needed when running
# tox on multiple tests.
reload(resources)
result = resources.LogicalPort(client.NSX3Client()).create(
fake_port['logical_switch_id'],
fake_port['attachment']['id'],
switch_profile_ids=profile_client.build_switch_profile_ids(
*profile_dicts))
api = resources.LogicalPort(client.NSX3Client())
with self.mocked_resource(api) as mocked:
resp_body = {
'logical_switch_id': fake_port['logical_switch_id'],
'attachment': {
'attachment_type': 'VIF',
'id': fake_port['attachment']['id']
},
'admin_state': 'UP',
'switching_profile_ids': fake_port['switching_profile_ids']
}
mocked.get('post').return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(fake_port))
self.assertEqual(fake_port, result)
self.assertEqual(fake_port['switching_profile_ids'],
resp_body['switching_profile_ids'])
test_client.assert_session_call(
mock_post,
'https://1.2.3.4/api/v1/logical-ports',
False,
jsonutils.dumps(resp_body),
client.JSONRESTClient._DEFAULT_HEADERS,
test_client.BaseClientTestCase.ca_file)
result = api.create(
fake_port['logical_switch_id'],
fake_port['attachment']['id'],
switch_profile_ids=[
{'value': profile['id'],
'key': profile['resource_type']}
for profile in profile_dicts])
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.post'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_create_logical_port_admin_down(self, mock_validate, mock_post):
resp_body = {
'logical_switch_id': fake_port['logical_switch_id'],
'admin_state': 'UP',
'switching_profile_ids': fake_port['switching_profile_ids'],
'attachment': {
'attachment_type': 'VIF',
'id': fake_port['attachment']['id']
}
}
self.assertEqual(fake_port, result)
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/logical-ports',
False,
jsonutils.dumps(resp_body),
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
def test_create_logical_port_admin_down(self):
"""
Test creating port with admin_state down
"""
fake_port = test_constants_v3.FAKE_PORT
fake_port['admin_state'] = "DOWN"
mock_post.return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(fake_port))
api = resources.LogicalPort(client.NSX3Client())
with self.mocked_resource(api) as mocked:
mocked.get('post').return_value = mocks.MockRequestsResponse(
200, jsonutils.dumps(fake_port))
# Reload resources because resources.LogicalPort.create is overridden
# by tests/unit/nsx_v3/test_plugin.py. This is only needed when running
# tox on multiple tests.
reload(resources)
result = resources.LogicalPort(client.NSX3Client()).create(
test_constants_v3.FAKE_PORT['logical_switch_id'],
test_constants_v3.FAKE_PORT['attachment']['id'],
tags={}, admin_state=False)
result = api.create(
test_constants_v3.FAKE_PORT['logical_switch_id'],
test_constants_v3.FAKE_PORT['attachment']['id'],
tags={}, admin_state=False)
self.assertEqual(fake_port, result)
self.assertEqual(fake_port, result)
@mock.patch("%s.%s" % (CLIENT_PKG, 'requests.Session.delete'))
@mock.patch(CLIENT_PKG + '.RESTClient._validate_result')
def test_delete_logical_port(self, mock_validate, mock_delete):
def test_delete_logical_port(self):
"""
Test deleting port
"""
mock_delete.return_value = mocks.MockRequestsResponse(
200, None)
api = resources.LogicalPort(client.NSX3Client())
with self.mocked_resource(api) as mocked:
mocked.get('delete').return_value = mocks.MockRequestsResponse(
200, None)
uuid = test_constants_v3.FAKE_PORT['id']
# Reload resources because resources.LogicalPort.delete is overridden
# by tests/unit/nsx_v3/test_plugin.py. This is only needed when running
# tox on multiple tests.
reload(resources)
result = resources.LogicalPort(client.NSX3Client()).delete(uuid)
self.assertIsNone(result.content)
test_client.assert_session_call(
mock_delete,
'https://1.2.3.4/api/v1/logical-ports/%s?detach=true' % uuid,
False,
None,
client.JSONRESTClient._DEFAULT_HEADERS,
test_client.BaseClientTestCase.ca_file)
uuid = test_constants_v3.FAKE_PORT['id']
result = api.delete(uuid)
self.assertIsNone(result.content)
test_client.assert_session_call(
mocked.get('delete'),
'https://1.2.3.4/api/v1/logical-ports/%s?detach=true' % uuid,
False,
None,
client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)

116
vmware_nsx/tests/unit/nsxlib/v3/test_switch.py
View File

@ -13,80 +13,96 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import mock
from oslo_log import log
from oslo_utils import uuidutils
from oslo_serialization import jsonutils
from vmware_nsx.common import nsx_constants
from vmware_nsx.nsxlib import v3 as nsxlib
from vmware_nsx.tests.unit.nsx_v3 import mocks as nsx_v3_mocks
from vmware_nsx.tests.unit.nsxlib.v3 import nsxlib_testcase
from vmware_nsx.tests.unit.nsxlib.v3 import test_client
LOG = log.getLogger(__name__)
# Cache the pointers to those functions because they are overridden
# by tests/unit/nsx_v3/test_plugin.py. This is only needed when running
# tox on multiple tests.
nsxlib_create_logical_switch = nsxlib.create_logical_switch
nsxlib_delete_logical_switch = nsxlib.delete_logical_switch
class NsxLibSwitchTestCase(nsxlib_testcase.NsxClientTestCase):
_tz_id = "8f602f97-ee3e-46b0-9d9f-358955f03608"
class NsxLibSwitchTestCase(nsxlib_testcase.NsxLibTestCase):
def _create_body(self, admin_state=nsx_constants.ADMIN_STATE_UP,
vlan_id=None):
body = {
"transport_zone_id": NsxLibSwitchTestCase._tz_id,
"replication_mode": "MTEP",
"display_name": "fake_name",
"tags": [],
"admin_state": admin_state
}
if vlan_id:
body['vlan'] = vlan_id
return body
@mock.patch("vmware_nsx.nsxlib.v3"
".client.create_resource")
def test_create_logical_switch(self, mock_create_resource):
def test_create_logical_switch(self):
"""
Test creating a switch returns the correct response and 200 status
"""
tz_uuid = uuidutils.generate_uuid()
fake_switch = nsx_v3_mocks.make_fake_switch(tz_uuid=tz_uuid)
mock_create_resource.return_value = fake_switch
api = self.new_client(nsxlib.client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
nsxlib.create_logical_switch(
nsx_v3_mocks.FAKE_NAME, NsxLibSwitchTestCase._tz_id, [])
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/logical-switches',
False, jsonutils.dumps(self._create_body()),
nsxlib.client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
result = nsxlib_create_logical_switch(nsx_v3_mocks.FAKE_NAME, tz_uuid,
[])
self.assertEqual(fake_switch, result)
@mock.patch("vmware_nsx.nsxlib.v3"
".client.create_resource")
def test_create_logical_switch_admin_down(self, mock_create_resource):
def test_create_logical_switch_admin_down(self):
"""
Test creating switch with admin_state down
"""
tz_uuid = uuidutils.generate_uuid()
fake_switch = nsx_v3_mocks.make_fake_switch(tz_uuid=tz_uuid)
fake_switch['admin_state'] = nsx_constants.ADMIN_STATE_DOWN
mock_create_resource.return_value = fake_switch
api = self.new_client(nsxlib.client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
nsxlib.create_logical_switch(
nsx_v3_mocks.FAKE_NAME, NsxLibSwitchTestCase._tz_id,
[], admin_state=False)
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/logical-switches',
False,
jsonutils.dumps(self._create_body(
admin_state=nsx_constants.ADMIN_STATE_DOWN)),
nsxlib.client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
result = nsxlib_create_logical_switch(nsx_v3_mocks.FAKE_NAME, tz_uuid,
[], admin_state=False)
self.assertEqual(fake_switch, result)
@mock.patch("vmware_nsx.nsxlib.v3"
".client.create_resource")
def test_create_logical_switch_vlan(self, mock_create_resource):
def test_create_logical_switch_vlan(self):
"""
Test creating switch with provider:network_type VLAN
"""
tz_uuid = uuidutils.generate_uuid()
fake_switch = nsx_v3_mocks.make_fake_switch()
fake_switch['vlan_id'] = '123'
mock_create_resource.return_value = fake_switch
api = self.new_client(nsxlib.client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
nsxlib.create_logical_switch(
nsx_v3_mocks.FAKE_NAME, NsxLibSwitchTestCase._tz_id,
[], vlan_id='123')
test_client.assert_session_call(
mocked.get('post'),
'https://1.2.3.4/api/v1/logical-switches',
False, jsonutils.dumps(self._create_body(vlan_id='123')),
nsxlib.client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)
result = nsxlib_create_logical_switch(nsx_v3_mocks.FAKE_NAME, tz_uuid,
[])
self.assertEqual(fake_switch, result)
@mock.patch("vmware_nsx.nsxlib.v3"
".client.delete_resource")
def test_delete_logical_switch(self, mock_delete_resource):
def test_delete_logical_switch(self):
"""
Test deleting switch
"""
mock_delete_resource.return_value = None
fake_switch = nsx_v3_mocks.make_fake_switch()
result = nsxlib_delete_logical_switch(fake_switch['id'])
self.assertIsNone(result)
api = self.new_client(nsxlib.client.NSX3Client)
with self.mocked_client_bridge(api, nsxlib, 'client') as mocked:
fake_switch = nsx_v3_mocks.make_fake_switch()
nsxlib.delete_logical_switch(fake_switch['id'])
test_client.assert_session_call(
mocked.get('delete'),
'https://1.2.3.4/api/v1/logical-switches/%s'
'?detach=true&cascade=true' % fake_switch['id'],
False, None,
nsxlib.client.JSONRESTClient._DEFAULT_HEADERS,
nsxlib_testcase.NSX_CERT)

16
vmware_nsx/tests/unit/services/l2gateway/test_nsxv3_driver.py
View File

@ -26,32 +26,34 @@ from neutron import context
from neutron.tests import base
from vmware_nsx.common import nsx_constants
from vmware_nsx.nsxlib import v3 as nsxlib
from vmware_nsx.services.l2gateway.common import plugin as l2gw_plugin
from vmware_nsx.services.l2gateway.nsx_v3 import driver as nsx_v3_driver
from vmware_nsx.tests.unit.nsx_v3 import mocks as nsx_v3_mocks
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_nsx_v3_plugin
NSX_V3_PLUGIN_CLASS = ('vmware_nsx.plugins.nsx_v3.plugin.NsxV3Plugin')
NSX_V3_L2GW_DRIVER_CLASS_PATH = ('vmware_nsx.services.l2gateway.'
'nsx_v3.driver.NsxV3Driver')
class TestNsxV3L2GatewayDriver(test_l2gw_db.L2GWTestCase,
test_nsx_v3_plugin.NsxPluginV3TestCase,
test_nsx_v3_plugin.NsxV3PluginTestCaseMixin,
base.BaseTestCase):
def setUp(self):
super(TestNsxV3L2GatewayDriver, self).setUp()
cfg.CONF.set_override("nsx_l2gw_driver",
NSX_V3_L2GW_DRIVER_CLASS_PATH)
self.core_plugin = importutils.import_object(NSX_V3_PLUGIN_CLASS)
self.l2gw_plugin = l2gw_plugin.NsxL2GatewayPlugin()
self.core_plugin._nsx_client = self.client
self.core_plugin._port_client._client._session = self.mock_api
self.maxDiff = None
self.driver = nsx_v3_driver.NsxV3Driver()
self.l2gw_plugin = l2gw_plugin.NsxL2GatewayPlugin()
self.context = context.get_admin_context()
# Mock NSX lib backend calls
nsxlib.create_bridge_endpoint = nsx_v3_mocks.create_bridge_endpoint
nsxlib.delete_bridge_endpoint = mock.Mock()
def test_nsxl2gw_driver_init(self):
with mock.patch.object(nsx_v3_driver.NsxV3Driver,