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

Merge "Tempest: Fixed TestRouterNoNATOps bugs and enhanced the test cases"

Browse Source
This commit is contained in:
Jenkins 2017-03-29 01:01:02 +00:00 committed by Gerrit Code Review
commit f361c7df14
1 changed files with 84 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
vmware_nsx_tempest/tests/nsxv3/scenario/test_router_nonat_ops.py
View File

@ -1,4 +1,4 @@
# Copyright 2016 VMware Inc # Copyright 2017 VMware Inc
# All Rights Reserved. # All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -43,9 +43,10 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
- Create a NATed topology and check end to end traffic. - Create a NATed topology and check end to end traffic.
Update the router to NoNAT and check end to end traffic. Update the router to NoNAT and check end to end traffic.
Note: For NoNAT use case, we need to add the static route on external Note: For NoNAT use case, Enable CONF.network.project_networks_reachable
VM in order for NSX connected network to be reachable from external. and add the static route on external VM in order for NSX connected
$ route add -net 192.168.1.0 netmask 255.255.255.0 gw 172.20.1.60 eth1 network to be reachable from external.
route add -net 192.168.1.0 netmask 255.255.255.0 gw 172.20.1.60 eth1
""" """
@classmethod @classmethod
@ -88,9 +89,10 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
self.router['id'], subnet_id=self.subnet['id']) self.router['id'], subnet_id=self.subnet['id'])
server_name = data_utils.rand_name('server-smoke') server_name = data_utils.rand_name('server-smoke')
self.server = self._create_server(server_name, self.network) self.server = self._create_server(server_name, self.network)
floating_ip = self.create_floating_ip(self.server) if enable_snat:
self.floating_ip_tuple = Floating_IP_tuple(floating_ip, floating_ip = self.create_floating_ip(self.server)
self.server) self.floating_ip_tuple = Floating_IP_tuple(floating_ip,
self.server)
def _cleanup_router(self, router): def _cleanup_router(self, router):
self._delete_router(router) self._delete_router(router)
@ -162,10 +164,10 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
should_connect=True): should_connect=True):
floating_ip, server = self.floating_ip_tuple floating_ip, server = self.floating_ip_tuple
# test internal connectivity to the network ports on the network # test internal connectivity to the network ports on the network
network_ips = (p['fixed_ips'][0]['ip_address'] for p in network_ips = [p['fixed_ips'][0]['ip_address'] for p in
self._list_ports(tenant_id=server['tenant_id'], self._list_ports(tenant_id=server['tenant_id'],
network_id=network['id']) network_id=network['id'])
if p['device_owner'].startswith('network')) if p['device_owner'].startswith('network')]
self._check_server_connectivity(floating_ip, self._check_server_connectivity(floating_ip,
network_ips, network_ips,
should_connect) should_connect)
@ -174,30 +176,34 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
should_connect=True): should_connect=True):
floating_ip, server = self.floating_ip_tuple floating_ip, server = self.floating_ip_tuple
# test internal connectivity to the other VM on the same network # test internal connectivity to the other VM on the same network
compute_ips = (p['fixed_ips'][0]['ip_address'] for p in compute_ips = [p['fixed_ips'][0]['ip_address'] for p in
self._list_ports(tenant_id=server['tenant_id'], self._list_ports(tenant_id=server['tenant_id'],
network_id=network['id']) network_id=network['id'])
if p['device_owner'].startswith('compute')) if p['device_owner'].startswith('compute')]
self._check_server_connectivity(floating_ip, self._check_server_connectivity(floating_ip,
compute_ips, compute_ips,
should_connect) should_connect)
def _check_nonat_network_connectivity(self): def _check_nonat_network_connectivity(self, should_connect=True):
server_ip = self._get_server_ip(self.server)
# test internal connectivity to the network ports on the network # test internal connectivity to the network ports on the network
network_ips = (p['fixed_ips'][0]['ip_address'] for p in network_ips = [p['fixed_ips'][0]['ip_address'] for p in
self._list_ports(tenant_id=self.server['tenant_id'], self._list_ports(tenant_id=self.server['tenant_id'],
network_id=self.network['id']) network_id=self.network['id'])
if p['device_owner'].startswith('network')) if p['device_owner'].startswith('network')]
self._check_server_connectivity(server_ip, network_ips.append(self._get_server_ip(self.server))
network_ips, self._check_fixed_ip_connectivity_from_ext_vm(
should_connect=True, network_ips, should_connect=should_connect)
enable_snat=False)
def _check_fixed_ip_connectivity_from_ext_vm(self, fixed_ips,
should_connect=True):
if not CONF.network.project_networks_reachable and should_connect:
return
for ip in fixed_ips:
self.ping_ip_address(ip, should_succeed=should_connect)
def _check_server_connectivity(self, floating_ip, address_list, def _check_server_connectivity(self, floating_ip, address_list,
should_connect=True, enable_snat=True): should_connect=True):
ip_address = floating_ip['floating_ip_address'] if enable_snat \ ip_address = floating_ip['floating_ip_address']
else floating_ip
private_key = self._get_server_key(self.server) private_key = self._get_server_key(self.server)
ssh_source = self.get_remote_client(ip_address, ssh_source = self.get_remote_client(ip_address,
private_key=private_key) private_key=private_key)
@ -217,8 +223,9 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
{'dest': remote_ip, 'src': floating_ip}) {'dest': remote_ip, 'src': floating_ip})
raise raise
def _test_router_nat_update(self, snat=True): def _test_router_nat_update_when_snat(self):
"""Test update router from NATed to NoNAT scenario""" """Test update router from NATed to NoNAT scenario"""
snat = True
self._setup_network_topo(enable_snat=snat) self._setup_network_topo(enable_snat=snat)
nsx_router = self.nsx.get_logical_router( nsx_router = self.nsx.get_logical_router(
self.router['name'], self.router['id']) self.router['name'], self.router['id'])
@ -226,46 +233,85 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
self.assertEqual(nsx_router['router_type'], 'TIER1') self.assertEqual(nsx_router['router_type'], 'TIER1')
# Check nat rules created correctly # Check nat rules created correctly
nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router) nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router)
self.assertTrue((len(nat_rules) == 3) == snat)
# Check router advertisement is correctly set # Check router advertisement is correctly set
router_adv = self.nsx.get_logical_router_advertisement(nsx_router) router_adv = self.nsx.get_logical_router_advertisement(nsx_router)
adv_msg = "Tier1 router's advertise_nsx_connected_routes is not %s" adv_msg = "Tier1 router's advertise_nsx_connected_routes is not True"
nat_msg = "Tier1 router's advertise_nat_routes is not %s" nat_msg = "Tier1 router's advertise_nat_routes is not False"
self.assertTrue(router_adv['advertise_nat_routes'], adv_msg % True) self.assertTrue(len(nat_rules) == 3)
self.assertFalse(router_adv['advertise_nsx_connected_routes'], self.assertTrue(router_adv['advertise_nat_routes'], nat_msg)
adv_msg % (not snat)) self.assertFalse(router_adv['advertise_nsx_connected_routes'], adv_msg)
self._check_network_internal_connectivity(network=self.network) self._check_network_internal_connectivity(network=self.network)
self._check_network_vm_connectivity(network=self.network) self._check_network_vm_connectivity(network=self.network)
self._check_nonat_network_connectivity(should_connect=False)
# Update router to disable snat # Update router to disable snat and disassociate floating ip
self.routers_client.update_router( self.routers_client.update_router(
self.router['id'], self.router['id'],
external_gateway_info={ external_gateway_info={
'network_id': CONF.network.public_network_id, 'network_id': CONF.network.public_network_id,
'enable_snat': (not snat)}) 'enable_snat': (not snat)})
floating_ip, server = self.floating_ip_tuple
self._disassociate_floating_ip(floating_ip)
nsx_router = self.nsx.get_logical_router( nsx_router = self.nsx.get_logical_router(
self.router['name'], self.router['id']) self.router['name'], self.router['id'])
self.assertNotEqual(nsx_router, None) self.assertNotEqual(nsx_router, None)
self.assertEqual(nsx_router['router_type'], 'TIER1') self.assertEqual(nsx_router['router_type'], 'TIER1')
# Check nat rules created correctly # Check nat rules created correctly
nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router) nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router)
self.assertTrue((len(nat_rules) == 2) == snat)
# Check router advertisement is correctly set # Check router advertisement is correctly set
router_adv = self.nsx.get_logical_router_advertisement(nsx_router) router_adv = self.nsx.get_logical_router_advertisement(nsx_router)
self.assertTrue(router_adv['advertise_nat_routes'], self.assertTrue(len(nat_rules) == 0)
nat_msg % True) self.assertFalse(router_adv['advertise_nat_routes'], nat_msg)
self.assertTrue(router_adv['advertise_nsx_connected_routes'], self.assertTrue(router_adv['advertise_nsx_connected_routes'], adv_msg)
adv_msg % snat)
self._check_nonat_network_connectivity() self._check_nonat_network_connectivity()
def _test_router_nat_update_when_no_snat(self):
"""Test update router from NATed to NoNAT scenario"""
snat = False
self._setup_network_topo(enable_snat=snat)
nsx_router = self.nsx.get_logical_router(
self.router['name'], self.router['id'])
self.assertNotEqual(nsx_router, None)
self.assertEqual(nsx_router['router_type'], 'TIER1')
# Check nat rules created correctly
nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router)
# Check router advertisement is correctly set
router_adv = self.nsx.get_logical_router_advertisement(nsx_router)
adv_msg = "Tier1 router's advertise_nsx_connected_routes is not True"
nat_msg = "Tier1 router's advertise_nat_routes is not False"
self.assertTrue(len(nat_rules) == 0)
self.assertFalse(router_adv['advertise_nat_routes'], nat_msg)
self.assertTrue(router_adv['advertise_nsx_connected_routes'], adv_msg)
self._check_nonat_network_connectivity()
# Update router to Enable snat and associate floating ip
self.routers_client.update_router(
self.router['id'],
external_gateway_info={
'network_id': CONF.network.public_network_id,
'enable_snat': (not snat)})
floating_ip = self.create_floating_ip(self.server)
self.floating_ip_tuple = Floating_IP_tuple(floating_ip, self.server)
nsx_router = self.nsx.get_logical_router(
self.router['name'], self.router['id'])
self.assertNotEqual(nsx_router, None)
self.assertEqual(nsx_router['router_type'], 'TIER1')
# Check nat rules created correctly
nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router)
# Check router advertisement is correctly set
router_adv = self.nsx.get_logical_router_advertisement(nsx_router)
self.assertTrue(len(nat_rules) == 3)
self.assertTrue(router_adv['advertise_nat_routes'], nat_msg)
self.assertFalse(router_adv['advertise_nsx_connected_routes'], adv_msg)
self._check_network_internal_connectivity(network=self.network)
self._check_network_vm_connectivity(network=self.network)
@test.attr(type='nsxv3') @test.attr(type='nsxv3')
@decorators.idempotent_id('5e5bfdd4-0962-47d3-a89b-7ce64322b53e') @decorators.idempotent_id('5e5bfdd4-0962-47d3-a89b-7ce64322b53e')
def test_router_nat_to_nonat_ops(self): def test_router_nat_to_nonat_ops(self):
"""Test update router from NATed to NoNAT scenario""" """Test update router from NATed to NoNAT scenario"""
self._test_router_nat_update(snat=True) self._test_router_nat_update_when_snat()
@test.attr(type='nsxv3') @test.attr(type='nsxv3')
@decorators.idempotent_id('a0274738-d3e7-49db-bf10-a5563610940d') @decorators.idempotent_id('a0274738-d3e7-49db-bf10-a5563610940d')
def test_router_nonat_to_nat_ops(self): def test_router_nonat_to_nat_ops(self):
"""Test update router from NoNAT to NATed scenario""" """Test update router from NoNAT to NATed scenario"""
self._test_router_nat_update(snat=False) self._test_router_nat_update_when_no_snat()