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

New Cases from RCCA PRs

Browse Source 
{0} vmware_nsx_tempest_plugin.tests.nsxv3.scenario.test_allowed_address_pair.TestAllowedAddressPair.test_allowed_address_pair_with_newly_created_port [376.970521s] ... ok
{0} vmware_nsx_tempest_plugin.tests.nsxv3.scenario.test_ipv6_router.IPv6RoutersTest.test_ipv4_interface_and_ipv6_interface_to_rtr [287.663608s] ... ok
{0} vmware_nsx_tempest_plugin.tests.nsxv3.scenario.test_ipv6_router.IPv6RoutersTest.test_ipv6_interface_and_ipv4_interface_to_rtr [309.487982s] ... ok
{0} vmware_nsx_tempest_plugin.tests.scenario.test_new_case_coverage.TestNewCase.test_ipv4_vlan_network_interface_to_router_with_snat_disabled [83.359754s] ... FAILED
{0} vmware_nsx_tempest_plugin.tests.scenario.test_new_case_coverage.TestNewCase.test_switching_gateway_tier1_router  [264.083448s] ... ok
{0} vmware_nsx_tempest_plugin.tests.nsxv3.scenario.test_allowed_address_pair.TestAllowedAddressPair.test_allowed_address_pair_with_newly_created_port [359.579611s] ... ok
{0} vmware_nsx_tempest_plugin.tests.nsxv3.scenario.test_allowed_address_pair.TestAllowedAddressPair.test_allowed_address_pair_for_update_port_with_single_cidr_ip [259.324594s] ... ok
{0} vmware_nsx_tempest_plugin.tests.nsxv3.scenario.test_octavia_loadbalancers.OctaviaRoundRobin.test_create_verify_octavia_lb_when_qos_policy_delete [922.771589s] ... ok
{0} vmware_nsx_tempest_plugin.tests.nsxv3.api.test_nsx_security_groups.NSXv3SecGroupTest.test_create_icmp_security_rule_with_range [39.644244s] ... ok

Change-Id: I81d4685dc41db1a647683e62dbfa07e14ae8b0be
This commit is contained in:
Shubham Naik 2021-09-27 06:20:10 +00:00
parent 211d5f96e1
commit fca44216f1
6 changed files with 433 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
vmware_nsx_tempest_plugin/lib/appliance_manager.py
View File

@ -227,7 +227,7 @@ class ApplianceManager(manager.NetworkScenarioTest):
return network
def create_provider_network(self, net_type,
zone_name,
zone_name=None,
admin_state_up=True,
tz_id=None,
vlan_id_unique=None):
@ -255,6 +255,8 @@ class ApplianceManager(manager.NetworkScenarioTest):
"provider:physical_network": tz_id,
"admin_state_up": admin_state_up,
"dns_domain": zone_name}
if not zone_name:
body.pop("dns_domain")
network = self.create_topology_network(name,
networks_client=networks_client,
**body)

28
vmware_nsx_tempest_plugin/tests/nsxv3/api/test_nsx_security_groups.py
View File

@ -22,6 +22,7 @@ from tempest.api.network import base_security_groups as base
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as exception
from tempest import test
from vmware_nsx_tempest_plugin.common import constants
@ -433,3 +434,30 @@ class NSXv3SecGroupTest(base.BaseSecGroupTest):
self.assertIsNotNone(nsx_dfw_rule)
self.assertEqual(nsx_dfw_rule['rule_tag'],
secgroup_rule['project_id'])
@decorators.attr(type='nsxv3',)
@decorators.attr(type=['negative'])
@decorators.idempotent_id('91c298c0-1234-4597-b4c6-1a7ecfb8a2de')
def test_create_icmp_security_rule_with_range(self):
# Create a security group
create_body, _ = self._create_security_group()
secgroup = create_body['security_group']
self.assertIsNotNone(secgroup['id'])
dfw_error_msg = "Firewall section not found for %s!" % secgroup['name']
if CONF.network.backend == 'nsxp':
self._wait_till_firewall_gets_realize(
secgroup, dfw_error_msg, 'default')
else:
self._wait_till_firewall_gets_realize(secgroup,
dfw_error_msg)
# Create rules for ICMP with port range
client = self.security_group_rules_client
rule = {
'direction': 'ingress',
'protocol': 'icmp',
'port_range_min': 0,
'port_range_max': 255,
}
self.assertRaises(exception.BadRequest,
client.create_security_group_rule,
security_group_id=secgroup['id'], **rule)

135
vmware_nsx_tempest_plugin/tests/nsxv3/scenario/test_allowed_address_pair.py
View File

@ -323,6 +323,55 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
(ssh_source, ip_address_vm2, 'True'),
'Destination is reachable')
def _test_connectivity_between_allowed_address_pair_update_ports_cidr(
self, network_topo):
server_name_default = data_utils.rand_name('server-default')
network = network_topo['network']
server_default = self._create_server(server_name_default, network)
server_name_default1 = \
data_utils.rand_name('server-default1-sec-group')
server_default1 = self._create_server(server_name_default1, network)
floating_ip_default = self.create_floating_ip(server_default)
floating_ip_default1 = self.create_floating_ip(server_default1)
ip_address_default_vm = floating_ip_default['floating_ip_address']
ip_address_default1_vm = floating_ip_default1['floating_ip_address']
private_key_default_vm = self._get_server_key(server_default)
private_key_default1_vm = self._get_server_key(server_default1)
port_client = self.ports_client
# Allowed Address pair
ip_address_vm1 = '83.0.0.3'
ip_address_vm2 = '83.0.0.4'
ip_subnet_vms = '83.0.0.0/24'
port_id = self.get_port_id(network['id'],
network_topo['subnet']['id'],
server_default,
network_name=network['name'])
# Update allowed address pair cidr attribute of port
allowed_address_pairs = [{'ip_address': ip_subnet_vms}]
port_client.update_port(
port_id, allowed_address_pairs=allowed_address_pairs)
port1_id = self.get_port_id(network['id'],
network_topo['subnet']['id'],
server_default1,
network_name=network['name'])
# Update allowed address pair cidr attribute of port
allowed_address_pairs = [{'ip_address': ip_subnet_vms}]
port_client.update_port(
port1_id, allowed_address_pairs=allowed_address_pairs)
ssh_source = self.get_remote_client(
ip_address_default_vm, private_key=private_key_default_vm)
ssh_source1 = self.get_remote_client(
ip_address_default1_vm, private_key=private_key_default1_vm)
# Assign Allowed pair ip to vm's
self._assign_ip_address(ssh_source, 'eth0:1', ip_address_vm1)
self._assign_ip_address(ssh_source1, 'eth0:1', ip_address_vm2)
self.assertTrue(self._check_remote_connectivity
(ssh_source1, ip_address_vm1, 'True'),
'Destination is reachable')
self.assertTrue(self._check_remote_connectivity
(ssh_source, ip_address_vm2, 'True'),
'Destination is reachable')
def _test_connectivity_between_allowed_adddress_pair_ports_cidr(
self, network_topo):
server_name_default = data_utils.rand_name('server-default')
@ -373,6 +422,70 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
(ssh_source, ip_address_vm2, 'True'),
'Destination is reachable')
def _test_connectivity_between_allowed_address_pair_on_existing_ports(
self, network_topo):
server_name_default = data_utils.rand_name('server-default')
network = network_topo['network']
port_client = self.ports_client
# Create port with allowed address pair attribute
ip_address_vm1 = '77.0.0.3'
ip_address_vm2 = '77.0.0.4'
ip_subnet_vms = '77.0.0.0/24'
body = {"network_id": network['id'],
"admin_state_up": 'true',
"allowed_address_pairs": [{'ip_address': ip_subnet_vms}]}
port_id = self._create_port(**body)
port_client.update_port(port_id['port']['id'],
security_groups=[self.security_group['id']])
# Create port1 with allowed address pair attribute
body = {"network_id": network['id'],
"admin_state_up": 'true',
"allowed_address_pairs": [{'ip_address': ip_subnet_vms}]}
port1_id = self._create_port(**body)
port_client.update_port(port1_id['port']['id'],
security_groups=[self.security_group['id']])
kwargs = {'port_id': port_id['port']['id']}
# Create servers for port with allowed address pair
server_default = self._create_server(server_name_default, network)
server_name_default1 = \
data_utils.rand_name('server-default1-sec-group')
server_default1 = self._create_server(server_name_default1, network)
floating_ip_default = self.create_floating_ip(server_default)
floating_ip_default1 = self.create_floating_ip(server_default1)
ip_address_default_vm = floating_ip_default['floating_ip_address']
ip_address_default1_vm = floating_ip_default1['floating_ip_address']
private_key_default_vm = self._get_server_key(server_default)
private_key_default1_vm = self._get_server_key(server_default1)
# Attach interface to vm
self.interface_client.create_interface(server_default['id'], **kwargs)
time.sleep(constants.NSX_BACKEND_VERY_SMALL_TIME_INTERVAL)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.interface_client.delete_interface,
server_default['id'], port_id['port']['id'])
kwargs = {'port_id': port1_id['port']['id']}
# Attach interface to vm
self.interface_client.create_interface(server_default1['id'], **kwargs)
time.sleep(constants.NSX_BACKEND_VERY_SMALL_TIME_INTERVAL)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.interface_client.delete_interface,
server_default1['id'], port1_id['port']['id'])
# Allowed Address pair
ssh_source = self.get_remote_client(
ip_address_default_vm, private_key=private_key_default_vm)
ssh_source1 = self.get_remote_client(
ip_address_default1_vm, private_key=private_key_default1_vm)
# Assign Allowed pair ip to vm's
self._assign_ip_address(ssh_source, 'eth1', ip_address_vm1)
self._assign_ip_address(ssh_source1, 'eth1', ip_address_vm2)
self.assertTrue(self._check_remote_connectivity
(ssh_source1, ip_address_vm1, 'True'),
'Destination is reachable')
self.assertTrue(self._check_remote_connectivity
(ssh_source, ip_address_vm2, 'True'),
'Destination is reachable')
def _test_v6_connectivity_between_allowed_adddress_pair_ports(
self, network_topo):
server_name_default = data_utils.rand_name('server-default')
@ -525,14 +638,14 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
kwargs = {'port_id': port_id['port']['id']}
# Attach interface to vm
self.interface_client.create_interface(server_default['id'], **kwargs)
time.sleep(10)
time.sleep(constants.NSX_BACKEND_VERY_SMALL_TIME_INTERVAL)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.interface_client.delete_interface,
server_default['id'], port_id['port']['id'])
kwargs = {'port_id': port1_id['port']['id']}
# Attach interface to vm
self.interface_client.create_interface(server_default1['id'], **kwargs)
time.sleep(10)
time.sleep(constants.NSX_BACKEND_VERY_SMALL_TIME_INTERVAL)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.interface_client.delete_interface,
server_default1['id'], port1_id['port']['id'])
@ -598,14 +711,14 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
time.sleep(constants.TIME["SEC"]["SIXTY"])
# Attach interface to vm
self.interface_client.create_interface(server_default['id'], **kwargs)
time.sleep(10)
time.sleep(constants.NSX_BACKEND_VERY_SMALL_TIME_INTERVAL)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.interface_client.delete_interface,
server_default['id'], port_id['port']['id'])
kwargs = {'port_id': port1_id['port']['id']}
# Attach interface to vm
self.interface_client.create_interface(server_default1['id'], **kwargs)
time.sleep(10)
time.sleep(constants.NSX_BACKEND_VERY_SMALL_TIME_INTERVAL)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.interface_client.delete_interface,
server_default1['id'], port1_id['port']['id'])
@ -1019,3 +1132,17 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
self.network_topo = self.create_ipv6_network_topo()
self._test_v6_allowed_address_pair_on_vms_with_multiple_cidr_ips(
self.network_topo)
@decorators.attr(type='nsxv3')
@decorators.idempotent_id('c0f0f446-65f5-40fa-8b05-b5798e8dd676')
def test_allowed_address_pair_with_newly_created_port(self):
self.network_topo = self.create_network_topo()
self._test_connectivity_between_allowed_address_pair_on_existing_ports(
self.network_topo)
@decorators.attr(type='nsxv3')
@decorators.idempotent_id('c0f0f446-65f5-40fa-8b05-b5798e8dd676')
def test_allowed_address_pair_for_update_port_with_single_cidr_ip(self):
self.network_topo = self.create_network_topo()
self._test_connectivity_between_allowed_address_pair_update_ports_cidr(
self.network_topo)

141
vmware_nsx_tempest_plugin/tests/nsxv3/scenario/test_ipv6_router.py
View File

@ -74,6 +74,25 @@ class IPv6RoutersTest(feature_manager.FeatureManager):
cidr=cidr, router_id=router_id)
return subnet
def _create_security_group(self, network):
sec_rule_client = self.cmgr_adm.security_group_rules_client
sec_client = self.cmgr_adm.security_groups_client
sg = self._create_empty_security_group(
namestart="tempest-ipv6-", client=sec_client)
common_ruleset = [dict(direction='ingress', ethertype='IPv6'),
dict(direction='ingress', ethertype='IPv4'),
dict(direction='ingress', protocol='tcp',
port_range_min=22, port_range_max=22),
dict(direction='egress', protocol='icmp'),
dict(direction='ingress', protocol='icmp')]
for rule in common_ruleset:
self._create_security_group_rule(
sec_group_rules_client=sec_rule_client,
security_groups_client=sec_client,
secgroup=sg,
**rule)
return sg
def create_topo_single_network(self, namestart, create_instance=True,
set_gateway=True, slaac=True, **kwargs):
"""
@ -735,3 +754,125 @@ class IPv6RoutersTest(feature_manager.FeatureManager):
self.assertRaises(exceptions.BadRequest,
self.cmgr_adm.routers_client.update_router,
router_id, routes=routes)
@decorators.attr(type=['nsxv3', 'positive'])
@decorators.idempotent_id('3564b971-6033-43cc-a13a-93b467bca803')
def test_ipv6_interface_and_ipv4_interface_to_rtr(self):
"""
Create ipv6 subnet under network1 & attach it to router
then create an ipv4 subnet under same network1 & attach to rtr
"""
name = data_utils.rand_name("dual-ipv6-ipv4-net")
networks_client = self.cmgr_adm.networks_client
network = self.create_topology_network(
name, networks_client=networks_client)
sg = self._create_security_group(network)
subnet_ipv6 = self._create_ipv6_subnet(
network, cidr="4200::/64",
ipv6_ra_mode="dhcpv6-stateful",
ipv6_address_mode="dhcpv6-stateful",
slaac=False)
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
rtr_name = data_utils.rand_name("dula-ipv4-ipv6-rtr")
router = self.create_topology_router(
rtr_name, routers_client=self.cmgr_adm.routers_client)
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv6['id'])
self.assertEqual(subnet_ipv6['id'], interface['subnet_id'])
subnet_client = self.cmgr_adm.subnets_client
subnet_ipv4_name = network['name'] + 'dual-ipv4-sub'
subnet_ipv4 = self.create_topology_subnet(
subnet_ipv4_name, network, subnets_client=subnet_client,
cidr='20.21.0.0/16')
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv4['id'])
self.assertEqual(subnet_ipv4['id'], interface['subnet_id'])
# Boot two VMs with the same network
image_id = self.get_glance_image_id(['cirros', 'esx'])
self.create_topology_instance(
"vm_1", [network],
security_groups=[{'name': sg['name']}],
create_floating_ip=True, image_id=image_id,
clients=self.cmgr_adm)
vm1_server_floatingip = self.topology_servers["vm_1"][
"floating_ips"][0]
self.create_topology_instance(
"vm_2", [network],
security_groups=[{'name': sg['name']}],
create_floating_ip=True, image_id=image_id,
clients=self.cmgr_adm)
vm2_server_floatingip = self.topology_servers["vm_2"][
"floating_ips"][0]
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
# Check IPv4 connectivity across the same network
vm1_server = self.topology_servers["vm_1"]
vm2_server = self.topology_servers["vm_2"]
self.check_vm_internal_connectivity(
network, vm1_server_floatingip, vm1_server)
self.check_vm_internal_connectivity(
network, vm2_server_floatingip, vm2_server)
@decorators.attr(type=['nsxv3', 'positive'])
@decorators.idempotent_id('3564b971-6033-43cc-a13a-93b467bca803')
def test_ipv4_interface_and_ipv6_interface_to_rtr(self):
"""
Create ipv4 subnet under network1 & attach it to router
then create an ipv6 subnet under same network1 & attach to rtr
"""
name = data_utils.rand_name("dual-ipv4-ipv6-net")
networks_client = self.cmgr_adm.networks_client
network = self.create_topology_network(
name, networks_client=networks_client)
sg = self._create_security_group(network)
subnet_client = self.cmgr_adm.subnets_client
subnet_ipv4_name = network['name'] + 'dual-ipv4-sub'
subnet_ipv4 = self.create_topology_subnet(
subnet_ipv4_name, network, subnets_client=subnet_client,
cidr='20.22.0.0/16')
rtr_name = data_utils.rand_name("dula-ipv4-ipv6-rtr")
router = self.create_topology_router(
rtr_name, routers_client=self.cmgr_adm.routers_client)
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv4['id'])
self.assertEqual(subnet_ipv4['id'], interface['subnet_id'])
subnet_ipv6 = self._create_ipv6_subnet(
network, cidr="4300::/64",
ipv6_ra_mode="dhcpv6-stateful",
ipv6_address_mode="dhcpv6-stateful",
slaac=False)
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv6['id'])
self.assertEqual(subnet_ipv6['id'], interface['subnet_id'])
# Boot two VMs with the same network
image_id = self.get_glance_image_id(['cirros', 'esx'])
self.create_topology_instance(
"vm_1", [network],
security_groups=[{'name': sg['name']}],
create_floating_ip=True, image_id=image_id,
clients=self.cmgr_adm)
vm1_server_floatingip = self.topology_servers["vm_1"][
"floating_ips"][0]
self.create_topology_instance(
"vm_2", [network],
security_groups=[{'name': sg['name']}],
create_floating_ip=True, image_id=image_id,
clients=self.cmgr_adm)
vm2_server_floatingip = self.topology_servers["vm_2"][
"floating_ips"][0]
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
# Check IPv4 connectivity across the same network
vm1_server = self.topology_servers["vm_1"]
vm2_server = self.topology_servers["vm_2"]
self.check_vm_internal_connectivity(
network, vm1_server_floatingip, vm1_server)
self.check_vm_internal_connectivity(
network, vm2_server_floatingip, vm2_server)

28
vmware_nsx_tempest_plugin/tests/nsxv3/scenario/test_octavia_loadbalancers.py
View File

@ -1363,3 +1363,31 @@ class OctaviaRoundRobin(feature_manager.FeatureManager):
get_loadbalancers(backend_name=backend_name)[0]
noerr = self.verifyBackendStatus_octavia_lb(backend_status)
self.assertTrue(noerr, "Backend Verification failed")
@decorators.attr(type='nsxv3')
@decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b21b1a34')
def test_create_verify_octavia_lb_when_qos_policy_delete(self):
"""
This testcase creates an octavia Loadbalancer with vip-net-ip
option, and verifies the traffic on the loadbalancer vip when
qos policy is deleted
"""
diction = self.deploy_octavia_topology()
self.start_web_servers(constants.HTTP_PORT)
net_id = diction['network']['id']
policy = self.cmgr_adm.qos_client.create_qos_policy(
name='test-policy', description='test policy desc1',
shared=False)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.cmgr_adm.qos_client.delete_qos_policy,
policy['policy']['id'])
lb_response = self.create_project_octavia(
protocol_type="HTTP", protocol_port="80",
lb_algorithm="LEAST_CONNECTIONS",
vip_net_id=net_id, hm_type='PING',
timeout=self.hm_timeout, max_retries=self.hm_max_retries,
delay=self.hm_delay, qos_policy_id=policy['policy']['id'])
self.assertEqual(lb_response['loadbalancer']['vip_qos_policy_id'],
policy['policy']['id'])
self.cmgr_adm.qos_client.delete_qos_policy(policy['policy']['id'])
self.check_project_lbaas()

102
vmware_nsx_tempest_plugin/tests/scenario/test_new_case_coverage.py
View File

@ -1959,3 +1959,105 @@ class TestNewCase(feature_manager.FeatureManager):
def test_verify_mdproxy_member_status(self):
"""verify mdproxy edges has member status is up"""
self.assertTrue(self.vsm.verify_member_status_of_md_proxy_edges())
@decorators.idempotent_id('6857fa0a-f7c2-400e-a8bf-a2a708eeaae1')
def test_ipv4_vlan_network_interface_to_router_with_snat_disabled(self):
network = self.create_provider_network(
net_type=constants.VLAN_TYPE, zone_name=None,
vlan_id_unique=int(CONF.nsxv.provider_vlan_id))
subnet_client = self.cmgr_adm.subnets_client
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
router = self.create_topology_router(
"ipv4-rtr-provider", set_gateway=True, enable_snat=True)
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
router_dict = {'network_id': CONF.network.public_network_id,
'enable_snat': False}
self.cmgr_adm.routers_client.update_router(
router['id'], external_gateway_info=router_dict)['router']
subnet_name = network['name'] + '_subnet'
subnet_ipv4 = self.create_topology_subnet(
subnet_name, network, enable_dhcp=True,
subnets_client=subnet_client)
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv4['id'])
self.assertEqual(subnet_ipv4['id'], interface['subnet_id'])
@decorators.idempotent_id('b3881f7a-4d7a-464d-9ecd-dc829e346e95')
def test_switching_gateway_tier1_router(self):
"""
"""
name = "ipv4-ext-network"
networks_client_ex = self.cmgr_adm.networks_client
ext_network = self.create_topology_network(
name, networks_client=networks_client_ex,
**{'router:external': True})
self.assertIsNotNone(ext_network['id'])
self.assertTrue(ext_network['router:external'])
subnet_client_ex = self.cmgr_adm.subnets_client
subnet_name_ex = ext_network['name'] + 'sub'
allocation_pools = {'allocation_pools': [{
'start': '100.20.0.2',
'end': '100.20.0.14'}]}
self.create_topology_subnet(subnet_name_ex, ext_network,
subnets_client=subnet_client_ex,
ip_version=4, enable_dhcp=False,
**allocation_pools)
# create new network1 and subnet1
name = data_utils.rand_name("ipv4-net11")
networks_client1 = self.cmgr_adm.networks_client
network = self.create_topology_network(
name, networks_client=networks_client1)
subnet_client1 = self.cmgr_adm.subnets_client
subnet_ipv4_name1 = network['name'] + 'ipv4-sub1'
subnet_ipv4_1 = self.create_topology_subnet(
subnet_ipv4_name1, network, subnets_client=subnet_client1)
rtr_name = data_utils.rand_name("ipv4--rtr")
router = self.create_topology_router(
rtr_name, routers_client=self.cmgr_adm.routers_client,
set_gateway=True)
time.sleep(constants.NSX_NETWORK_REALISE_TIMEOUT)
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv4_1['id'])
self.assertEqual(subnet_ipv4_1['id'], interface['subnet_id'])
# create new network2 and subnet2
name = data_utils.rand_name("ipv4-net12")
networks_client2 = self.cmgr_adm.networks_client
network1 = self.create_topology_network(
name, networks_client=networks_client2)
kwargs = dict(tenant_id=network1['tenant_id'],
security_group_rules_client=self.sec_rule_client,
security_groups_client=self.sec_client)
sg = self.create_topology_security_group(**kwargs)
subnet_client2 = self.cmgr_adm.subnets_client
subnet_ipv4_name2 = network1['name'] + 'ipv4-sub2'
subnet_ipv4_2 = self.create_topology_subnet(
subnet_ipv4_name2, network1, subnets_client=subnet_client2)
interface = self.add_router_interface(
router_client=self.cmgr_adm.routers_client,
router_id=router['id'], subnet_id=subnet_ipv4_2['id'])
self.assertEqual(subnet_ipv4_2['id'], interface['subnet_id'])
# Boot two VMs on the same network
image_id = self.get_glance_image_id(['cirros', 'esx'])
self.create_topology_instance(
"vm_11", [network],
security_groups=[{'name': sg['name']}],
create_floating_ip=False, image_id=image_id,
clients=self.cmgr_adm)
self.create_topology_instance(
"vm_21", [network1],
security_groups=[{'name': sg['name']}],
create_floating_ip=False, image_id=image_id,
clients=self.cmgr_adm)
# Update the external network of router
router_dict = {'network_id': ext_network['id'], 'enable_snat': False}
router_update = self.cmgr_adm.routers_client.update_router(
router['id'], external_gateway_info=router_dict)['router']
self.assertEqual(router_update['external_gateway_info']['network_id'],
ext_network['id'])