x/vmware-nsx
Code Issues Proposed changes

Tempest: Added L2GW API tests

Browse Source 
- Restructured function to reuse some of the function in various module
- Move pass and fail test verification within the test functions.
- Added L2GW connection API tests
- Added negative test cases

Change-Id: I9ac67b8ae0473dfc5b855744d2a4ca429b0045a6
This commit is contained in:
Devang Doshi 2016-03-28 22:00:13 +00:00 committed by Kobi Samoray
parent 9f50df6f7e
commit c12b31ee03
6 changed files with 1017 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
vmware_nsx_tempest/common/constants.py
View File

@ -24,3 +24,4 @@ L2_GWS_BASE_URI = "/l2-gateways"
EXPECTED_HTTP_RESPONSE_200 = "200" EXPECTED_HTTP_RESPONSE_200 = "200"
EXPECTED_HTTP_RESPONSE_201 = "201" EXPECTED_HTTP_RESPONSE_201 = "201"
EXPECTED_HTTP_RESPONSE_204 = "204" EXPECTED_HTTP_RESPONSE_204 = "204"
L2GWC = "l2_gateway_connection"

4
vmware_nsx_tempest/config.py
View File

@ -117,6 +117,10 @@ L2gwGroup = [
cfg.StrOpt('vlan_2', cfg.StrOpt('vlan_2',
default="17", default="17",
help="VLAN id"), help="VLAN id"),
cfg.StrOpt("subnet_1_cidr",
default="192.168.1.0/24",
help="Subnet 1 network cidr."
"Example: 1.1.1.0/24"),
] ]
nsxv3_group = cfg.OptGroup(name='nsxv3', nsxv3_group = cfg.OptGroup(name='nsxv3',

189
vmware_nsx_tempest/services/base_l2gw.py
View File

@ -1,3 +1,6 @@
# Copyright 2016 VMware Inc
# All Rights Reserved.
#
# Copyright 2015 Hewlett-Packard Development Company, L.P. # Copyright 2015 Hewlett-Packard Development Company, L.P.
# Copyright 2015 OpenStack Foundation # Copyright 2015 OpenStack Foundation
# #
@ -13,7 +16,20 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from tempest.api.network import base
from tempest import config
from tempest import test
from vmware_nsx_tempest._i18n import _LI
from vmware_nsx_tempest._i18n import _LW
from vmware_nsx_tempest.common import constants
from vmware_nsx_tempest.services import l2_gateway_client
from vmware_nsx_tempest.services import l2_gateway_connection_client
from vmware_nsx_tempest.services import nsxv3_client
LOG = constants.log.getLogger(__name__)
CONF = config.CONF
SEGMENTATION_ID_DELIMITER = "#" SEGMENTATION_ID_DELIMITER = "#"
INTERFACE_SEG_ID_DELIMITER = "|" INTERFACE_SEG_ID_DELIMITER = "|"
DEVICE_INTERFACE_DELIMITER = "::" DEVICE_INTERFACE_DELIMITER = "::"
@ -92,3 +108,176 @@ def form_dict_devices(devices):
devices1.setdefault(device_name, []).append(int_seg) devices1.setdefault(device_name, []).append(int_seg)
int_seg = [] int_seg = []
return devices1 return devices1
class BaseL2GatewayTest(base.BaseAdminNetworkTest):
"""
L2Gateway base class. Extend this class to get basics of L2GW.
"""
credentials = ["primary", "admin"]
@classmethod
def skip_checks(cls):
"""
Skip running test if we do not meet criteria to run the tests.
"""
super(BaseL2GatewayTest, cls).skip_checks()
if not test.is_extension_enabled("l2-gateway", "network"):
raise cls.skipException("l2-gateway extension not enabled.")
@classmethod
def setup_clients(cls):
"""
Create various client connections. Such as NSXv3 and L2 Gateway.
"""
super(BaseL2GatewayTest, cls).setup_clients()
cls.l2gw_created = {}
cls.l2gwc_created = {}
try:
manager = getattr(cls.os_adm, "manager", cls.os_adm)
net_client = getattr(manager, "networks_client")
_params = manager.default_params_withy_timeout_values.copy()
except AttributeError as attribute_err:
LOG.warning(
_LW("Failed to locate the attribute, Error: %(err_msg)s") %
{"err_msg": attribute_err.__str__()})
_params = {}
cls.l2gw_client = l2_gateway_client.L2GatewayClient(
net_client.auth_provider,
net_client.service,
net_client.region,
net_client.endpoint_type,
**_params)
cls.nsxv3_client_obj = nsxv3_client.NSXV3Client(
CONF.nsxv3.nsx_manager,
CONF.nsxv3.nsx_user,
CONF.nsxv3.nsx_password)
cls.l2gwc_client = \
l2_gateway_connection_client.L2GatewayConnectionClient(
net_client.auth_provider,
net_client.service,
net_client.region,
net_client.endpoint_type,
**_params)
@classmethod
def resource_setup(cls):
"""
Setting up the resources for the test.
"""
super(BaseL2GatewayTest, cls).resource_setup()
cls.VLAN_1 = CONF.l2gw.vlan_1
cls.VLAN_2 = CONF.l2gw.vlan_2
@classmethod
def resource_cleanup(cls):
"""
Clean all the resources used during the test.
"""
for l2gw_id in cls.l2gw_created.keys():
cls.l2gw_client.delete_l2_gateway(l2gw_id)
cls.l2gw_created.pop(l2gw_id)
for l2gwc_id in cls.l2gwc_created.keys():
cls.l2gwc_client.delete_l2_gateway_connection(l2gwc_id)
cls.l2gwc_created.pop(l2gwc_id)
def create_l2gw(self, l2gw_name, l2gw_param):
"""
Creates L2GW and return the response.
:param l2gw_name: name of the L2GW
:param l2gw_param: L2GW parameters
:return: response of L2GW create API
"""
LOG.info(_LI("l2gw name: %(name)s, l2gw_param: %(devices)s ") %
{"name": l2gw_name, "devices": l2gw_param})
devices = []
for device_dict in l2gw_param:
interface = [{"name": device_dict["iname"],
"segmentation_id": device_dict[
"vlans"]}] if "vlans" in device_dict else [
{"name": device_dict["iname"]}]
device = {"device_name": device_dict["dname"],
"interfaces": interface}
devices.append(device)
l2gw_request_body = {"devices": devices}
LOG.info(_LI(" l2gw_request_body: %s") % l2gw_request_body)
rsp = self.l2gw_client.create_l2_gateway(
name=l2gw_name, **l2gw_request_body)
LOG.info(_LI(" l2gw response: %s") % rsp)
self.l2gw_created[rsp[constants.L2GW]["id"]] = rsp[constants.L2GW]
return rsp, devices
def delete_l2gw(self, l2gw_id):
"""
Delete L2gw.
:param l2gw_id: L2GW id to delete l2gw.
:return: response of the l2gw delete API.
"""
LOG.info(_LI("L2GW id: %(id)s to be deleted.") % {"id": l2gw_id})
rsp = self.l2gw_client.delete_l2_gateway(l2gw_id)
LOG.info(_LI("response : %(rsp)s") % {"rsp": rsp})
return rsp
def update_l2gw(self, l2gw_id, l2gw_new_name, devices):
"""
Update existing L2GW.
:param l2gw_id: L2GW id to update its parameters.
:param l2gw_new_name: name of the L2GW.
:param devices: L2GW parameters.
:return: Response of the L2GW update API.
"""
rsp = self.l2gw_client.update_l2_gateway(l2gw_id,
name=l2gw_new_name, **devices)
return rsp
def nsx_bridge_cluster_info(self):
"""
Collect the device and interface name of the nsx brdige cluster.
:return: nsx bridge id and display name.
"""
response = self.nsxv3_client_obj.get_bridge_cluster_info()
if len(response) == 0:
raise RuntimeError("NSX bridge cluster information is null")
return [(x.get("id"), x.get("display_name")) for x in response]
def create_l2gw_connection(self, l2gwc_param):
"""
Creates L2GWC and return the response.
:param l2gwc_param: L2GWC parameters.
:return: response of L2GWC create API.
"""
LOG.info(_LI("l2gwc param: %(param)s ") % {"param": l2gwc_param})
l2gwc_request_body = {"l2_gateway_id": l2gwc_param["l2_gateway_id"],
"network_id": l2gwc_param["network_id"]}
if "segmentation_id" in l2gwc_param:
l2gwc_request_body["segmentation_id"] = l2gwc_param[
"segmentation_id"]
LOG.info(_LI("l2gwc_request_body: %s") % l2gwc_request_body)
rsp = self.l2gwc_client.create_l2_gateway_connection(
**l2gwc_request_body)
LOG.info(_LI("l2gwc response: %s") % rsp)
self.l2gwc_created[rsp[constants.L2GWC]["id"]] = rsp[constants.L2GWC]
return rsp
def delete_l2gw_connection(self, l2gwc_id):
"""
Delete L2GWC and returns the response.
:param l2gwc_id: L2GWC id to delete L2GWC.
:return: response of the l2gwc delete API.
"""
LOG.info(_LI("L2GW connection id: %(id)s to be deleted")
% {"id": l2gwc_id})
rsp = self.l2gwc_client.delete_l2_gateway_connection(l2gwc_id)
LOG.info(_LI("response : %(rsp)s") % {"rsp": rsp})
return rsp

444
vmware_nsx_tempest/tests/nsxv3/api/test_l2_gateway.py
View File

@ -15,217 +15,121 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from tempest.api.network import base
from tempest import config from tempest import config
from tempest.lib.common.utils import data_utils from tempest.lib.common.utils import data_utils
from tempest import test from tempest import test
from vmware_nsx_tempest._i18n import _LI from vmware_nsx_tempest._i18n import _LI
from vmware_nsx_tempest._i18n import _LW
from vmware_nsx_tempest.common import constants from vmware_nsx_tempest.common import constants
from vmware_nsx_tempest.services import l2_gateway_client from vmware_nsx_tempest.services import base_l2gw
from vmware_nsx_tempest.services import nsxv3_client
LOG = constants.log.getLogger(__name__) LOG = constants.log.getLogger(__name__)
CONF = config.CONF CONF = config.CONF
class L2GatewayTest(base.BaseAdminNetworkTest): class L2GatewayTest(base_l2gw.BaseL2GatewayTest):
""" """
Test l2 gateway operations. Test l2 gateway operations.
""" """
credentials = ["primary", "admin"]
@classmethod
def skip_checks(cls):
"""
Skip running test if we do not meet crietria to run the tests.
"""
super(L2GatewayTest, cls).skip_checks()
if not test.is_extension_enabled("l2-gateway", "network"):
raise cls.skipException("l2-gateway extension not enabled.")
@classmethod
def setup_clients(cls):
"""
Create various client connections. Such as NSXv3 and L2 Gateway.
"""
super(L2GatewayTest, cls).setup_clients()
cls.l2gw_created = {}
try:
manager = getattr(cls.os_adm, "manager", cls.os_adm)
net_client = getattr(manager, "networks_client")
_params = manager.default_params_withy_timeout_values.copy()
except AttributeError as attribute_err:
LOG.warning(
_LW("Failed to locate the attribute, Error: %(err_msg)s") %
{"err_msg": attribute_err.__str__()})
_params = {}
cls.l2gw_client = l2_gateway_client.L2GatewayClient(
net_client.auth_provider,
net_client.service,
net_client.region,
net_client.endpoint_type,
**_params)
cls.nsxv3_client_obj = nsxv3_client.NSXV3Client(
CONF.nsxv3.nsx_manager,
CONF.nsxv3.nsx_user,
CONF.nsxv3.nsx_password)
@classmethod
def resource_setup(cls):
"""
Setting up the resources for the test.
"""
super(L2GatewayTest, cls).resource_setup()
cls.VLAN_1 = getattr(CONF.l2gw, "vlan_1", None)
cls.VLAN_2 = getattr(CONF.l2gw, "vlan_2", None)
@classmethod
def resource_cleanup(cls):
"""
Clean all the resources used during the test.
"""
for l2gw_id in cls.l2gw_created.keys():
cls.l2gw_client.delete_l2_gateway(l2gw_id)
cls.l2gw_created.pop(l2gw_id)
def operate_l2gw(self, name, devices, task):
"""
l2gw operations such create, delete, update and show.
:param name: l2gw name.
:param devices: l2gw parameters information.
:param task: Operation to perform.
"""
LOG.info(_LI("name: %(name)s, devices: %(devices)s, "
"task: %(task)s") %
{"name": name, "devices": devices, "task": task})
if task == "create":
rsp = self.l2gw_client.create_l2_gateway(
name=name, **devices)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_201})
rsp_l2gw = rsp[constants.L2GW]
self.l2gw_created[rsp_l2gw["id"]] = rsp_l2gw
LOG.info(_LI("response : %(rsp_l2gw)s") % {"rsp_l2gw": rsp_l2gw})
self.assertEqual(name, rsp_l2gw["name"],
"l2gw name=%(rsp_name)s is not the same as "
"requested=%(name)s" % {"rsp_name": rsp_l2gw[
"name"], "name": name})
elif task == "delete":
l2gw_id, _ = self.l2gw_created.popitem()
rsp = self.l2gw_client.delete_l2_gateway(l2gw_id)
LOG.info(_LI("response : %(rsp)s") % {"rsp": rsp})
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_204})
elif task == "update":
l2gw_id, _ = self.l2gw_created.popitem()
rsp = self.l2gw_client.update_l2_gateway(l2gw_id,
name=name, **devices)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_200})
rsp_l2gw = rsp[constants.L2GW]
self.l2gw_created[rsp_l2gw["id"]] = rsp_l2gw
LOG.info(_LI("response : %(rsp_l2gw)s") % {"rsp_l2gw": rsp_l2gw})
self.assertEqual(name, rsp_l2gw["name"],
"l2gw name=%(rsp_name)s is not the same as "
"requested=%(name)s" % {
"rsp_name": rsp_l2gw["name"], "name": name})
if "name" in devices["devices"][0]["interfaces"][0]:
self.assertEqual(
devices["devices"][0]["interfaces"][0]["name"],
rsp_l2gw["devices"][0]["interfaces"][0][
"name"], "L2GW interface name update "
"failed!!!")
if "segmentation_id" in devices["devices"][0]["interfaces"][0]:
self.assertEqual(devices["devices"][0]["interfaces"][0][
"segmentation_id"][0],
rsp_l2gw["devices"][0]["interfaces"][0][
"segmentation_id"][0],
"L2GW segmentation id update failed!!!")
elif task == "show":
l2gw_id, l2gw_parameters = self.l2gw_created.popitem()
rsp = self.l2gw_client.show_l2_gateway(l2gw_id)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_200})
rsp_l2gw = rsp[constants.L2GW]
constants.LOG.info(
_LI("response : %(rsp_l2gw)s") % {"rsp_l2gw": rsp_l2gw})
self.assertEqual(name, rsp_l2gw["name"],
"l2gw name=%(rsp_name)s is not the same as "
"requested=%(name)s" % {
"rsp_name": rsp_l2gw["name"],
"name": name})
self.assertEqual(l2gw_parameters, rsp_l2gw,
"l2-gateway-show does not show parameter as it "
"was created.")
self.l2gw_created[l2gw_id] = l2gw_parameters
def nsx_bridge_cluster_info(self):
"""
Collect the device and interface name of the nsx brdige cluster.
:return: nsx bridge id and display name.
"""
response = self.nsxv3_client_obj.get_bridge_cluster_info()
if len(response) == 0:
raise RuntimeError("NSX bridge cluster information is null")
return response[0]["id"], response[0]["display_name"]
@test.attr(type="nsxv3") @test.attr(type="nsxv3")
@test.idempotent_id("e5e3a089-602c-496e-8c17-4ef613266924") @test.idempotent_id("e5e3a089-602c-496e-8c17-4ef613266924")
def test_l2_gateway_create(self): def test_l2_gateway_create_without_vlan(self):
""" """
Create l2gw based on UUID and bridge cluster name. It creates l2gw. Create l2gw based on UUID and bridge cluster name. It creates l2gw.
To create l2gw we need bridge cluster name (interface name) and To create l2gw we need bridge cluster name (interface name) and
bridge cluster UUID (device name) from NSX manager. bridge cluster UUID (device name) from NSX manager.
""" """
LOG.info(_LI("Testing l2_gateway_create api")) LOG.info(_LI("Testing l2_gateway_create api"))
device_name, interface_name = self.nsx_bridge_cluster_info() cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW) l2gw_name = data_utils.rand_name(constants.L2GW)
devices = {"devices": [{"device_name": device_name, device_1 = {"dname": device_name, "iname": interface_name}
"interfaces": [{"name": interface_name}]}] l2gw_param = [device_1]
} rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
self.operate_l2gw(name=l2gw_name, devices=devices, task="create") # Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(requested_devices[0]["device_name"],
rsp[constants.L2GW]["devices"][0]["device_name"],
"Device name is not the same as expected")
self.resource_cleanup() self.resource_cleanup()
@test.attr(type="nsxv3") @test.attr(type="nsxv3")
@test.idempotent_id("9968a529-e785-472f-8705-9b394a912e43") @test.idempotent_id("9968a529-e785-472f-8705-9b394a912e43")
def test_l2_gateway_create_with_segmentation_id(self): def test_l2_gateway_with_single_vlan(self):
""" """
Create l2gw based on UUID and bridge cluster name. It creates l2gw. Create l2gw based on UUID and bridge cluster name. It creates l2gw.
To create l2gw we need bridge cluster name (interface name) and To create l2gw we need bridge cluster name (interface name) and
bridge cluster UUID (device name) from NSX manager and vlan id. bridge cluster UUID (device name) from NSX manager and vlan id.
""" """
LOG.info(_LI("Testing l2_gateway_create api")) LOG.info(_LI("Testing l2_gateway_create api with segmentation ID"))
device_name, interface_name = self.nsx_bridge_cluster_info() cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW) l2gw_name = data_utils.rand_name(constants.L2GW)
devices = {"devices": [ device_1 = {"dname": device_name, "iname": interface_name,
{"device_name": device_name, "vlans": [self.VLAN_1]}
"interfaces": [{"name": interface_name, l2gw_param = [device_1]
"segmentation_id": [ rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
self.VLAN_1, # Assert if create fails.
self.VLAN_2 self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
] rsp.response["status"],
}] "Response code is not %(code)s" % {
}] "code": constants.EXPECTED_HTTP_RESPONSE_201})
} self.assertEqual(requested_devices[0]["device_name"],
self.operate_l2gw(name=l2gw_name, devices=devices, task="create") rsp[constants.L2GW]["devices"][0]["device_name"],
"Device name is not the same as expected")
self.assertEqual(requested_devices[0]["interfaces"][0][
"name"],
rsp[constants.L2GW]["devices"][0]["interfaces"][0][
"name"],
"Interface name is not the same as expected")
requested_vlans = \
requested_devices[0]["interfaces"][0]["segmentation_id"]
response_vlans = rsp[constants.L2GW]["devices"][0]["interfaces"][0][
"segmentation_id"]
for id in requested_vlans:
self.assertIn(id, response_vlans)
self.resource_cleanup()
@test.attr(type="nsxv3")
@test.idempotent_id("3861aab0-4f76-4472-ad0e-a255e6e42193")
def test_l2_gateway_with_multiple_vlans(self):
"""
Create l2gw based on UUID and bridge cluster name. It creates l2gw.
To create l2gw we need bridge cluster name (interface name) and
bridge cluster UUID (device name) from NSX manager and vlan id.
"""
LOG.info(_LI("Testing l2_gateway_create api with segmentation ID"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [self.VLAN_1, self.VLAN_2]}
l2gw_param = [device_1]
rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(requested_devices[0]["device_name"],
rsp[constants.L2GW]["devices"][0]["device_name"],
"Device name is not the same as expected")
self.assertEqual(requested_devices[0]["interfaces"][0][
"name"],
rsp[constants.L2GW]["devices"][0]["interfaces"][0][
"name"],
"Interface name is not the same as expected")
requested_vlans = \
requested_devices[0]["interfaces"][0]["segmentation_id"]
response_vlans = rsp[constants.L2GW]["devices"][0]["interfaces"][0][
"segmentation_id"]
for id in requested_vlans:
self.assertIn(id, response_vlans)
self.resource_cleanup() self.resource_cleanup()
@test.attr(type="nsxv3") @test.attr(type="nsxv3")
@ -236,15 +140,27 @@ class L2GatewayTest(base.BaseAdminNetworkTest):
delete l2gw we need l2gw id. delete l2gw we need l2gw id.
""" """
LOG.info(_LI("Testing l2_gateway_delete api")) LOG.info(_LI("Testing l2_gateway_delete api"))
device_name, interface_name = self.nsx_bridge_cluster_info() cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW) l2gw_name = data_utils.rand_name(constants.L2GW)
devices = {"devices": [{"device_name": device_name, device_1 = {"dname": device_name, "iname": interface_name}
"interfaces": [{"name": interface_name}]}] l2gw_param = [device_1]
} # Create l2gw to delete it.
# Creating l2gw. rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
self.operate_l2gw(name=l2gw_name, devices=devices, task="create") # Assert if create fails.
# Deleting already created l2gw. self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
self.operate_l2gw(name=l2gw_name, devices=devices, task="delete") rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_201})
l2gw_id = rsp[constants.L2GW]["id"]
# Delete l2gw.
rsp = self.delete_l2gw(l2gw_id)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
self.l2gw_created.pop(l2gw_id)
self.resource_cleanup() self.resource_cleanup()
@test.attr(type="nsxv3") @test.attr(type="nsxv3")
@ -255,16 +171,38 @@ class L2GatewayTest(base.BaseAdminNetworkTest):
update l2gw we need l2gw id and payload to update. update l2gw we need l2gw id and payload to update.
""" """
LOG.info(_LI("Testing l2_gateway_update api")) LOG.info(_LI("Testing l2_gateway_update api"))
device_name, interface_name = self.nsx_bridge_cluster_info() cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW) l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
# Create l2gw to update l2gw name.
rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_201})
devices = {"devices": [{"device_name": device_name, devices = {"devices": [{"device_name": device_name,
"interfaces": [{"name": interface_name}]}] "interfaces": [{"name": interface_name}]}]
} }
# Creating l2gw. l2gw_id = rsp[constants.L2GW]["id"]
self.operate_l2gw(name=l2gw_name, devices=devices, task="create") l2gw_new_name = "updated_name"
# Updating already created l2gw with new l2gw name. # Update l2gw name.
self.operate_l2gw(name=l2gw_name + "_updated", devices=devices, update_rsp = self.update_l2gw(l2gw_id, l2gw_new_name, devices)
task="update") self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
update_rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_200})
rsp_l2gw = update_rsp[constants.L2GW]
LOG.info(_LI("response : %(rsp_l2gw)s") % {"rsp_l2gw": rsp_l2gw})
# Assert if name is not updated.
self.assertEqual(l2gw_new_name, rsp_l2gw["name"],
"l2gw name=%(rsp_name)s is not the same as "
"requested=%(name)s" % {"rsp_name": rsp_l2gw["name"],
"name": l2gw_new_name})
self.resource_cleanup() self.resource_cleanup()
@test.attr(type="nsxv3") @test.attr(type="nsxv3")
@ -275,26 +213,42 @@ class L2GatewayTest(base.BaseAdminNetworkTest):
update l2gw we need l2gw id and payload to update. update l2gw we need l2gw id and payload to update.
""" """
LOG.info(_LI("Testing l2_gateway_update api")) LOG.info(_LI("Testing l2_gateway_update api"))
device_name, interface_name = self.nsx_bridge_cluster_info() cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW) l2gw_name = data_utils.rand_name(constants.L2GW)
devices = {"devices": [{"device_name": device_name, device_1 = {"dname": device_name, "iname": interface_name}
"interfaces": [{"name": interface_name}]}] l2gw_param = [device_1]
} # Create l2gw to update l2gw name.
# Creating l2gw. rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
self.operate_l2gw(name=l2gw_name, devices=devices, task="create") # Assert if create fails.
# Updating Interfaces. self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_201})
devices = {"devices": [ devices = {"devices": [
{"device_name": device_name, {"device_name": device_name,
"interfaces": [{"name": "new_name", "interfaces": [{"name": "new_name",
"segmentation_id": [ "segmentation_id": [self.VLAN_1]}],
self.VLAN_1]}], "deleted_interfaces": [{"name": interface_name}]}
"deleted_interfaces": [ ]}
{"name": interface_name}]} l2gw_id = rsp[constants.L2GW]["id"]
] update_rsp = self.update_l2gw(l2gw_id, l2gw_name, devices)
} self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
# Updating already created l2gw with new interface. update_rsp.response["status"],
self.operate_l2gw(name=l2gw_name, devices=devices, task="update") "Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_200})
rsp_l2gw = update_rsp[constants.L2GW]
self.l2gw_created[rsp_l2gw["id"]] = rsp_l2gw
LOG.info(_LI("response : %(rsp_l2gw)s") % {"rsp_l2gw": rsp_l2gw})
if "segmentation_id" in devices["devices"][0]["interfaces"][0]:
self.assertEqual(devices["devices"][0]["interfaces"][0][
"segmentation_id"][0],
rsp_l2gw["devices"][0]["interfaces"][0][
"segmentation_id"][0],
"L2GW segmentation id update failed!!!")
self.resource_cleanup() self.resource_cleanup()
@test.attr(type="nsxv3") @test.attr(type="nsxv3")
@ -304,18 +258,72 @@ class L2GatewayTest(base.BaseAdminNetworkTest):
show l2gw based on UUID. To see l2gw info we need l2gw id. show l2gw based on UUID. To see l2gw info we need l2gw id.
""" """
LOG.info(_LI("Testing l2_gateway_show api")) LOG.info(_LI("Testing l2_gateway_show api"))
device_name, interface_name = self.nsx_bridge_cluster_info() cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW) l2gw_name = data_utils.rand_name(constants.L2GW)
devices = {"devices": [ device_1 = {"dname": device_name, "iname": interface_name,
{"device_name": device_name, "vlans": [self.VLAN_1, self.VLAN_2]}
"interfaces": [{"name": interface_name, l2gw_param = [device_1]
"segmentation_id": [ rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
self.VLAN_1, # Assert if create fails.
self.VLAN_2]}] self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
}] rsp.response["status"],
} "Response code is not %(code)s" % {
# Creating l2gw. "code": constants.EXPECTED_HTTP_RESPONSE_201})
self.operate_l2gw(name=l2gw_name, devices=devices, task="create") l2gw_id = rsp[constants.L2GW]["id"]
# Show already created l2gw with l2gw id. l2gw_id = str(l2gw_id)
self.operate_l2gw(name=l2gw_name, devices=devices, task="show") show_rsp = self.l2gw_client.show_l2_gateway(l2gw_id)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
show_rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_200})
show_rsp = show_rsp[constants.L2GW]["devices"]
rsp = rsp[constants.L2GW]["devices"]
self.assertEqual(rsp[0]["device_name"],
show_rsp[0]["device_name"],
"Device name is not the same as expected")
self.assertEqual(
rsp[0]["interfaces"][0]["name"],
show_rsp[0]["interfaces"][0]["name"],
"Interface name is not the same as expected")
requested_vlans = \
rsp[0]["interfaces"][0]["segmentation_id"]
response_vlans = show_rsp[0]["interfaces"][0]["segmentation_id"]
for id in requested_vlans:
self.assertIn(id, response_vlans)
self.resource_cleanup()
@test.attr(type="nsxv3")
@test.idempotent_id("d4a7d3af-e637-45c5-a967-d179153a6e58")
def test_l2_gateway_list(self):
"""
list created l2gw.
"""
LOG.info(_LI("Testing l2_gateway_list api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [self.VLAN_1]}
l2gw_param = [device_1]
l2gw_rsp, requested_devices = self.create_l2gw(l2gw_name, l2gw_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gw_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
list_rsp = self.l2gw_client.list_l2_gateways()
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
list_rsp.response["status"],
"Response code is not %(code)s" % {
"code":
constants.EXPECTED_HTTP_RESPONSE_200})
for l2gw in list_rsp[constants.L2GWS]:
if l2gw["id"] == l2gw_rsp[constants.L2GW]["id"]:
list_rsp = l2gw
l2gw_rsp = l2gw_rsp[constants.L2GW]
break
self.assertEqual(l2gw_rsp, list_rsp, "L2GW create response and L2GW "
"list response does not match.")
self.resource_cleanup() self.resource_cleanup()

382
vmware_nsx_tempest/tests/nsxv3/api/test_l2_gateway_connection.py Normal file
View File

@ -0,0 +1,382 @@
# Copyright 2015 OpenStack Foundation
# Copyright 2016 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 netaddr
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest import test
from vmware_nsx_tempest._i18n import _LI
from vmware_nsx_tempest.common import constants
from vmware_nsx_tempest.services import base_l2gw
CONF = config.CONF
LOG = constants.log.getLogger(__name__)
class L2GatewayConnectionTest(base_l2gw.BaseL2GatewayTest):
"""
Test l2 gateway connection operations.
"""
@classmethod
def resource_setup(cls):
"""
Setting up the resources for the test.
"""
super(L2GatewayConnectionTest, cls).resource_setup()
# Create a network.
cls.network = cls.create_network()
# Create subnet on the network just created.
cls.SUBNET_1_NETWORK_CIDR = CONF.l2gw.subnet_1_cidr
network_cidr = cls.SUBNET_1_NETWORK_CIDR.split("/")
cls.SUBNET_1_MASK = network_cidr[1]
subnet_info = {}
# cidr must be presented & in IPNetwork structure.
cls.CIDR = netaddr.IPNetwork(cls.SUBNET_1_NETWORK_CIDR)
cls.subnet = cls.create_subnet(cls.network, cidr=cls.CIDR,
mask_bits=int(cls.SUBNET_1_MASK),
**subnet_info)
@classmethod
def resource_cleanup(cls):
"""
Clean all the resources used during the test.
"""
super(L2GatewayConnectionTest, cls).resource_cleanup()
cls._try_delete_resource(cls.networks_client.delete_network,
cls.network["id"])
@classmethod
def l2gw_cleanup(cls):
"""
Delete created L2GWs and L2GWCs.
"""
for l2gwc_id in cls.l2gwc_created.keys():
cls.l2gwc_client.delete_l2_gateway_connection(l2gwc_id)
cls.l2gwc_created.pop(l2gwc_id)
for l2gw_id in cls.l2gw_created.keys():
cls.l2gw_client.delete_l2_gateway(l2gw_id)
cls.l2gw_created.pop(l2gw_id)
@test.attr(type="nsxv3")
@decorators.skip_because(bug="634513")
@test.idempotent_id("81edfb9e-4722-4565-939c-6593b8405ff4")
def test_l2_gateway_connection_create(self):
"""
Create l2 gateway connection using one vlan. Vlan parameter is
passed into L2GW create.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [self.VLAN_1]}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"]}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param["l2_gateway_id"],
l2gwc_rsp[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param["network_id"],
l2gwc_rsp[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@decorators.skip_because(bug="634513")
@test.idempotent_id("7db4f6c9-18c5-4a99-93c1-68bc2ecb48a7")
def test_l2_gateway_connection_create_with_multiple_vlans(self):
"""
Create l2 gateway connection using multiple vlans. Vlan parameter is
passed into L2GW create.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [self.VLAN_1, self.VLAN_2]}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"]}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param["l2_gateway_id"],
l2gwc_rsp[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param["network_id"],
l2gwc_rsp[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("de70d6a2-d454-4a09-b06b-8f39be67b635")
def test_l2_gateway_connection_with_seg_id_create(self):
"""
Create l2 gateway connection using one vlan. Vlan parameter is
passed into L2GW connection create.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param["l2_gateway_id"],
l2gwc_rsp[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param["network_id"],
l2gwc_rsp[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param["segmentation_id"],
l2gwc_rsp[constants.L2GWC]["segmentation_id"],
"segmentation id is not same as expected in "
"create l2gw connection response")
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("819d9b50-9159-48d0-be2a-493ec686534c")
def test_l2_gateway_connection_show(self):
"""
Create l2 gateway connection using one vlan and tes l2 gateway
connection show api
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
l2gwc_id = l2gwc_rsp[constants.L2GWC]["id"]
show_rsp = self.l2gwc_client.show_l2_gateway_connection(l2gwc_id)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
show_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_200})
self.assertEqual(l2gwc_param["l2_gateway_id"],
show_rsp[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"show l2gw connection response")
self.assertEqual(l2gwc_param["network_id"],
show_rsp[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"show l2gw connection response")
show_rsp_seg_id = str(show_rsp[constants.L2GWC][
"segmentation_id"])
self.assertEqual(l2gwc_param["segmentation_id"],
show_rsp_seg_id,
"segmentation id is not same as expected in "
"show l2gw connection response")
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("4188f8e7-cd65-427e-92b8-2a9e0492ab21")
def test_l2_gateway_connection_list(self):
"""
Create l2 gateway connection using one vlan and test l2 gateway
connection list api.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
# Create 2 l2 gateways.
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
# Create 2 l2 gateway connections.
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
list_rsp = self.l2gwc_client.list_l2_gateway_connections()
LOG.info(_LI("l2gw connection list response: %s") % list_rsp)
# Assert in case of failure.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
list_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_200})
self.assertEqual(l2gwc_rsp["l2_gateway_connection"]["id"],
list_rsp["l2_gateway_connections"][0]["id"],
"l2gw connection list does not show proper id")
self.assertEqual(l2gwc_rsp["l2_gateway_connection"]["l2_gateway_id"],
list_rsp["l2_gateway_connections"][0][
"l2_gateway_id"],
"l2gw connection list does not show proper "
"l2_gateway_id")
self.assertEqual(l2gwc_rsp["l2_gateway_connection"]["network_id"],
list_rsp["l2_gateway_connections"][0]["network_id"],
"l2gw connection list does not show proper "
"network_id")
self.assertEqual(l2gwc_rsp["l2_gateway_connection"]["tenant_id"],
list_rsp["l2_gateway_connections"][0]["tenant_id"],
"l2gw connection list does not show proper tenant_id")
self.assertEqual(l2gwc_rsp["l2_gateway_connection"]["segmentation_id"],
str(list_rsp["l2_gateway_connections"][0][
"segmentation_id"]),
"l2gw connection list does not show proper "
"segmentation_id")
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("4d71111f-3d2b-4557-97c7-2e149a6f41fb")
def test_l2_gateway_connection_recreate(self):
"""
Recreate l2 gateway connection.
- Create l2GW.
- Create l2gw connection.
- delete l2gw connection.
- Recreate l2gw connection
- verify with l2gw connection list API.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
# List all the L2GW connection.
list_rsp = self.l2gwc_client.list_l2_gateway_connections()
LOG.info(_LI("l2gw connection list response: %s") % list_rsp)
# Assert in case of failure.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
list_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_200})
list_rsp = list_rsp["l2_gateway_connections"]
l2gwc_ids = [item.get("id") for item in list_rsp if "id"
in item]
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
l2gwc_id = l2gwc_rsp[constants.L2GWC]["id"]
# Delete l2gw.
rsp = self.delete_l2gw_connection(l2gwc_id)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
# Since we delete l2gwc pop that id from list.
self.l2gwc_created.pop(l2gwc_id)
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
# List all the L2GW connection.
list_rsp = self.l2gwc_client.list_l2_gateway_connections()
LOG.info(_LI("l2gw connection list response: %s") % list_rsp)
# Assert in case of failure.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_200,
list_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_200})
list_rsp = list_rsp["l2_gateway_connections"]
l2gwc_ids = l2gwc_ids + [item.get("id") for item in list_rsp if
"id" in item]
self.assertNotIn(l2gwc_id, l2gwc_ids, "l2gwc list api shows hanging "
"l2gwc id")
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("670cacb5-134e-467d-ba41-0d7cdbcf3903")
def test_l2_gateway_connection_delete(self):
"""
Delete l2gw will create l2gw and delete recently created l2gw. To
delete l2gw we need l2gw id.
"""
LOG.info(_LI("Testing l2_gateway_connection_delete api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
l2gwc_id = l2gwc_rsp[constants.L2GWC]["id"]
# Delete l2gw.
rsp = self.delete_l2gw_connection(l2gwc_id)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
# Since we delete l2gwc pop that id from list.
self.l2gwc_created.pop(l2gwc_id)
self.addCleanup(self.l2gw_cleanup)

215
vmware_nsx_tempest/tests/nsxv3/api/test_l2_gateway_connection_negative.py Normal file
View File

@ -0,0 +1,215 @@
# Copyright 2015 OpenStack Foundation
# Copyright 2016 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 netaddr
from tempest import config
from tempest import test
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from vmware_nsx_tempest._i18n import _LI
from vmware_nsx_tempest.common import constants
from vmware_nsx_tempest.services import base_l2gw
CONF = config.CONF
NON_EXIST_UUID = "12341234-0000-1111-2222-000000000000"
LOG = constants.log.getLogger(__name__)
class L2GatewayConnectionNegative(base_l2gw.BaseL2GatewayTest):
"""
Negative L2GW tests.
"""
@classmethod
def resource_setup(cls):
"""
Setting up the resources for the test.
"""
super(L2GatewayConnectionNegative, cls).resource_setup()
# Create a network.
cls.network = cls.create_network()
# Create subnet on the network just created.
cls.SUBNET_1_NETWORK_CIDR = CONF.l2gw.subnet_1_cidr
network_cidr = cls.SUBNET_1_NETWORK_CIDR.split("/")
cls.SUBNET_1_MASK = network_cidr[1]
subnet_info = {}
# cidr must be presented & in IPNetwork structure.
cls.CIDR = netaddr.IPNetwork(cls.SUBNET_1_NETWORK_CIDR)
cls.subnet = cls.create_subnet(cls.network, cidr=cls.CIDR,
mask_bits=int(cls.SUBNET_1_MASK),
**subnet_info)
@classmethod
def resource_cleanup(cls):
"""
Clean all the resources used during the test.
"""
super(L2GatewayConnectionNegative, cls).resource_cleanup()
cls._try_delete_resource(cls.networks_client.delete_network,
cls.network["id"])
@classmethod
def l2gw_cleanup(cls):
"""
Delete created L2GWs and L2GWCs.
"""
for l2gwc_id in cls.l2gwc_created.keys():
cls.l2gwc_client.delete_l2_gateway_connection(l2gwc_id)
cls.l2gwc_created.pop(l2gwc_id)
for l2gw_id in cls.l2gw_created.keys():
cls.l2gw_client.delete_l2_gateway(l2gw_id)
cls.l2gw_created.pop(l2gw_id)
@test.attr(type="nsxv3")
@test.idempotent_id("e86bd8e9-b32b-425d-86fa-cd866138d028")
def test_active_l2_gateway_delete(self):
"""
Delete l2 gateway with active mapping.
"""
LOG.info(_LI("Testing test_l2_gateway_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
l2gw_id = l2gw_rsp[constants.L2GW]["id"]
# Delete l2gw must raise Conflict exception.
self.assertRaises(lib_exc.Conflict, self.delete_l2gw, l2gw_id)
self.addCleanup(self.l2gw_cleanup)
@decorators.skip_because(bug="634513")
@test.attr(type="nsxv3")
@test.idempotent_id("488faaae-180a-4c48-8b7a-44c3a243369f")
def test_recreate_l2_gateway_connection(self):
"""
Recreate l2 gateway connection using same parameters.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [self.VLAN_1]}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": self.network["id"]}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertRaises(lib_exc.Conflict, self.create_l2gw_connection,
l2gwc_param)
self.addCleanup(self.l2gw_cleanup)
@decorators.skip_because(bug="1640042")
@test.attr(type="nsxv3")
@test.idempotent_id("14606e74-4f65-402e-ae50-a0adcd877a83")
def test_create_l2gwc_with_nonexist_l2gw(self):
"""
Create l2 gateway connection using non exist l2gw uuid.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
non_exist_l2gw_uuid = NON_EXIST_UUID
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": non_exist_l2gw_uuid,
"network_id": self.network["id"],
"segmentation_id": self.VLAN_1}
# Delete l2gw must raise Conflict exception.
self.assertRaises(lib_exc.NotFound, self.create_l2gw_connection,
l2gwc_param)
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("e6cb8973-fcbc-443e-a3cb-c6a82ae58b63")
def test_create_l2gwc_with_nonexist_network(self):
"""
Create l2 gateway connection using non exist l2gw uuid.
"""
LOG.info(_LI("Testing test_l2_gateway_connection_create api"))
non_exist_network_uuid = NON_EXIST_UUID
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name}
l2gw_param = [device_1]
l2gw_rsp, _ = self.create_l2gw(l2gw_name, l2gw_param)
l2gwc_param = {"l2_gateway_id": l2gw_rsp[constants.L2GW]["id"],
"network_id": non_exist_network_uuid,
"segmentation_id": self.VLAN_1}
# Delete l2gw must raise Conflict exception.
self.assertRaises(lib_exc.NotFound, self.create_l2gw_connection,
l2gwc_param)
self.addCleanup(self.l2gw_cleanup)
@test.attr(type="nsxv3")
@test.idempotent_id("27c7c64f-511f-421e-8b62-dfed143fc00b")
def test_create_l2gw_with_invalid_seg_id(self):
"""
Create l2 gateway connection using invalid seg id.
"""
LOG.info(_LI("Testing l2_gateway_create api with segmentation ID"))
invalid_seg_id = 20000
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [invalid_seg_id]}
l2gw_param = [device_1]
self.assertRaises(lib_exc.BadRequest, self.create_l2gw, l2gw_name,
l2gw_param)
self.addCleanup(self.l2gw_cleanup)
@decorators.skip_because(bug="1640033")
@test.attr(type="nsxv3")
@test.idempotent_id("000cc597-bcea-4539-af07-bd70357e8d82")
def test_create_l2gw_with_non_int_seg_id(self):
"""
Create l2 gateway connection using invalid seg id.
"""
LOG.info(_LI("Testing l2_gateway_create api with segmentation ID"))
invalid_seg_id = 2.45
cluster_info = self.nsx_bridge_cluster_info()
device_name, interface_name = cluster_info[0][0], cluster_info[0][1]
l2gw_name = data_utils.rand_name(constants.L2GW)
device_1 = {"dname": device_name, "iname": interface_name,
"vlans": [invalid_seg_id]}
l2gw_param = [device_1]
self.assertRaises(lib_exc.BadRequest, self.create_l2gw, l2gw_name,
l2gw_param)
self.addCleanup(self.l2gw_cleanup)