x/vmware-nsx-tempest-plugin
Code Issues Proposed changes

Merge "L2 Gateway scenario cases."

Browse Source
This commit is contained in:
Zuul 2019-08-09 10:05:00 +00:00 committed by Gerrit Code Review
commit f8c76d680a
4 changed files with 671 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
vmware_nsx_tempest_plugin/common/constants.py
View File

@ -94,7 +94,7 @@ PEER_ADDRESS = "172.24.4.12"
SITE_CONNECTION_STATE = 'True'
PSK = "secret"
CIDR = "22.0.9.0/24"
#BARBICAN
# BARBICAN
SECRET_TYPE = "opaque"
ALGORITHM = "aes"
PAYLOAD_CONTENT_TYPE = "text/plain"
@ -103,3 +103,8 @@ BIT_LENGTH = 256
CERT_FILE = "/root/server.crt"
KEY_FILE = "/root/server.key"
CONTAINER_TYPE = "certificate"
BRIDGE_VM1 = "192.168.1.201"
BRIDGE_VM2 = "192.168.1.202"
BRIDGE_VM3 = "192.168.2.203"

6
vmware_nsx_tempest_plugin/config.py
View File

@ -166,12 +166,6 @@ L2gwGroup = [
default="192.168.1.0/24",
help="Subnet 1 network cidr."
"Example: 1.1.1.0/24"),
cfg.StrOpt('vm1_on_vds_vlan16_ip',
default="192.168.1.201",
help="IPv4 IP address of VM1"),
cfg.StrOpt('vm2_on_vds_vlan17_ip',
default="192.168.1.202",
help="IPv4 IP address of VM2"),
cfg.StrOpt('vm_on_vds_tz1_vlan16_ip',
default="192.168.1.203",
help="IPv4 IP address of VM3"),

7
vmware_nsx_tempest_plugin/tests/nsxv3/api/test_l2_gateway.py
View File

@ -1361,13 +1361,16 @@ class L2GatewayConnectionNegative(L2GatewayBase):
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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_param2)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("14606e74-4f65-402e-ae50-a0adcd877a83")

662
vmware_nsx_tempest_plugin/tests/nsxv3/scenario/test_l2_gateway_scenario.py
View File

@ -19,6 +19,7 @@ import netaddr
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from tempest import test
from vmware_nsx_tempest_plugin.common import constants
@ -48,8 +49,9 @@ class L2GatewayBase(feature_manager.FeatureManager):
super(L2GatewayBase, cls).resource_setup()
cls.VLAN_1 = CONF.l2gw.vlan_1
cls.VLAN_2 = CONF.l2gw.vlan_2
cls.VM_1 = CONF.l2gw.vm1_on_vds_vlan16_ip
cls.VM_2 = CONF.l2gw.vm2_on_vds_vlan17_ip
cls.VM_1 = constants.BRIDGE_VM1
cls.VM_2 = constants.BRIDGE_VM2
cls.VM_3 = constants.BRIDGE_VM3
# Create subnet on the network just created.
cls.SUBNET_1_NETWORK_CIDR = CONF.l2gw.subnet_1_cidr
cls.SUBNET_1_MASK = cls.SUBNET_1_NETWORK_CIDR.split("/")[1]
@ -211,3 +213,659 @@ class L2GatewayConnectionTest(L2GatewayBase):
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=True)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("81edfb9e-4722-4345-939c-6593b8406ff4")
def test_l2_gateway_two_connections_create_different_bridge_profile(self):
"""
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_name2, interface_name = cluster_info[1][0], transport_zone
device_2 = {"dname": device_name2, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_2}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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")
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param2["l2_gateway_id"],
l2gwc_rsp2[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param2["network_id"],
l2gwc_rsp2[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
compute_ips1 = []
compute_ips1.append(self.VM_2)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=True)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("81edfb9e-4722-4345-939c-6593b8408f64")
def test_l2_gateway_two_connections_create_delete_two(self):
"""
Create l2 gateway connection using one vlan. Vlan parameter is
passed into L2GW create.
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_2 = {"dname": device_name, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_2}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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})
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
compute_ips1 = []
compute_ips1.append(self.VM_2)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=True)
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})
l2gwc_id2 = l2gwc_rsp2[constants.L2GWC]["id"]
# Delete l2gw.
rsp2 = self.delete_l2gw_connection(l2gwc_id2)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=False)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=False)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("81edfb9e-4722-4345-939c-6593b8406f64")
def test_l2_gateway_two_connections_create_delete_one(self):
"""
Create l2 gateway connection using one vlan. Vlan parameter is
passed into L2GW create.
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_2 = {"dname": device_name, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_2}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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})
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
compute_ips1 = []
compute_ips1.append(self.VM_2)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=True)
l2gwc_id2 = l2gwc_rsp2[constants.L2GWC]["id"]
# Delete l2gw.
rsp = self.delete_l2gw_connection(l2gwc_id2)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=False)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("81edfb9e-4722-4345-939c-6593b8a06ff4")
def test_l2_gateway_two_connections_create_diff_bridge_profile_del1(self):
"""
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_name2, interface_name = cluster_info[1][0], transport_zone
device_2 = {"dname": device_name2, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_2}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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")
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param2["l2_gateway_id"],
l2gwc_rsp2[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param2["network_id"],
l2gwc_rsp2[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
compute_ips1 = []
compute_ips1.append(self.VM_2)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=True)
l2gwc_id2 = l2gwc_rsp2[constants.L2GWC]["id"]
# Delete l2gw.
rsp2 = self.delete_l2gw_connection(l2gwc_id2)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=False)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("81edfb9e-4722-4345-939c-6593b8a06ff4")
def test_l2_gateway_two_connections_create_diff_bridge_profile_del2(self):
"""
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_name2, interface_name = cluster_info[1][0], transport_zone
device_2 = {"dname": device_name2, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_2}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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")
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param2["l2_gateway_id"],
l2gwc_rsp2[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param2["network_id"],
l2gwc_rsp2[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
compute_ips1 = []
compute_ips1.append(self.VM_2)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=True)
l2gwc_id1 = l2gwc_rsp[constants.L2GWC]["id"]
# Delete l2gw.
rsp1 = self.delete_l2gw_connection(l2gwc_id1)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp1.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
l2gwc_id2 = l2gwc_rsp2[constants.L2GWC]["id"]
# Delete l2gw.
rsp2 = self.delete_l2gw_connection(l2gwc_id2)
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_204,
rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_204})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=False)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips1, should_connect=False)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("de70d6a2-d454-4349-b66b-8f39be67b635")
def test_l2_gateway_connection_with_seg_id_recreate(self):
"""
Create l2 gateway connection using one vlan. Vlan parameter is
passed into L2GW connection create.
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["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})
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
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})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=False)
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.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("de70d6a2-d454-4349-b66b-8f39be64b635")
def test_l2_gateway_connection_with_seg_id_recreate_diff_profile(self):
"""
Create l2 gateway connection using one vlan. Vlan parameter is
passed into L2GW connection create.
"""
LOG.info("Testing test_l2_gateway_connection_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["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})
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
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})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=False)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_name2, interface_name = cluster_info[1][0], transport_zone
device_2 = {"dname": device_name2, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
@decorators.attr(type="nsxv4")
@decorators.idempotent_id("e86bd8e9-b32b-434d-86fa-cd866138d028")
def test_active_l2_gateway_delete(self):
"""
Delete l2 gateway with active mapping.
"""
LOG.info("Testing test_l2_gateway_create api")
self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["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})
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
l2gw_id = l2gw_rsp[constants.L2GW]["id"]
# Delete l2gw must raise Conflict exception.
self.assertRaises(lib_exc.Conflict, self.delete_l2gw, l2gw_id)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
@decorators.attr(type="nsxv3")
@decorators.idempotent_id("81edfb9e-4722-4345-939c-6593b8436ff4")
def test_l2_gateway_two_connections_create_diff_bridge_profile_vlan(self):
"""
"""
LOG.info("Testing test_l2_gateway_connection_create api")
topology_dict = self.create_topo_single_network("admin_state")
if CONF.nsxv3.bridge_cluster:
cluster_info = self.nsx_bridge_cluster_info()
else:
cluster_info = self.nsx_bridge_profile_info()
transport_zone = self.nsx_transport_zone_info()
device_name, interface_name = cluster_info[0][0], transport_zone
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.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp = self.create_l2gw_connection(l2gwc_param)
network_l2gateway = self.create_topology_network("network_l2gateway")
# cidr must be presented & in IPNetwork structure.
self.CIDR = netaddr.IPNetwork('192.168.2.0/24')
self.create_topology_subnet(
"subnet1_l2gateway", network_l2gateway, cidr=self.CIDR,
mask_bits=int(self.SUBNET_1_MASK),
router_id=topology_dict['router_state']['id'])
image_id = self.get_glance_image_id(["cirros", "esx"])
self.create_topology_instance("state_vm_2", [network_l2gateway],
create_floating_ip=True,
image_id=image_id)
l2gw_name2 = data_utils.rand_name(constants.L2GW)
device_name2, interface_name = cluster_info[1][0], transport_zone
device_2 = {"dname": device_name2, "iname": interface_name}
l2gw_param2 = [device_2]
l2gw_rsp2, _ = self.create_l2gw(l2gw_name2, l2gw_param2)
l2gwc_param2 = {"l2_gateway_id": l2gw_rsp2[constants.L2GW]["id"],
"network_id":
self.topology_networks["network_l2gateway"]["id"],
"segmentation_id": self.VLAN_1}
l2gwc_rsp2 = self.create_l2gw_connection(l2gwc_param2)
# 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")
# Assert if create fails.
self.assertEqual(constants.EXPECTED_HTTP_RESPONSE_201,
l2gwc_rsp2.response["status"],
"Response code is not %(code)s" % {
"code": constants.EXPECTED_HTTP_RESPONSE_201})
self.assertEqual(l2gwc_param2["l2_gateway_id"],
l2gwc_rsp2[constants.L2GWC]["l2_gateway_id"],
"l2gw id is not same as expected in "
"create l2gw connection response")
self.assertEqual(l2gwc_param2["network_id"],
l2gwc_rsp2[constants.L2GWC]["network_id"],
"network id is not same as expected in "
"create l2gw connection response")
compute_ips = []
compute_ips.append(self.VM_1)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server,
compute_ips, should_connect=True)
compute_ips = []
compute_ips.append(self.VM_3)
self.check_server_internal_ips_using_floating_ip(
self.servers_details.get("state_vm_2").floating_ips[0],
self.servers_details.get("state_vm_2").server,
compute_ips, should_connect=True)