160 lines
6.6 KiB
Python
160 lines
6.6 KiB
Python
# Copyright 2018 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 os
|
|
import subprocess
|
|
|
|
from oslo_log import log as logging
|
|
from tempest import config
|
|
from tempest.lib.common.utils import data_utils
|
|
from tempest.lib.common.utils import test_utils
|
|
from tempest.lib import decorators
|
|
|
|
from vmware_nsx_tempest_plugin.lib import feature_manager
|
|
from vmware_nsx_tempest_plugin.services import nsxp_client
|
|
from vmware_nsx_tempest_plugin.services import nsxv3_client
|
|
|
|
CONF = config.CONF
|
|
CONF.validation.auth_method = 'None'
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class TestVRFBasic(feature_manager.FeatureManager):
|
|
|
|
@classmethod
|
|
def setup_clients(cls):
|
|
super(TestVRFBasic, cls).setup_clients()
|
|
cls.cmgr_adm = cls.get_client_manager('admin')
|
|
cls.rtr_client = cls.cmgr_adm.routers_client
|
|
cls.nw_client = cls.cmgr_adm.networks_client
|
|
cls.subnets_client = cls.cmgr_adm.subnets_client
|
|
cls.ports_client = cls.cmgr_adm.ports_client
|
|
cls.sec_rule_client = cls.cmgr_adm.security_group_rules_client
|
|
cls.sec_client = cls.cmgr_adm.security_groups_client
|
|
|
|
def setUp(self):
|
|
super(TestVRFBasic, self).setUp()
|
|
self.nsx = nsxv3_client.NSXV3Client(CONF.nsxv3.nsx_manager,
|
|
CONF.nsxv3.nsx_user,
|
|
CONF.nsxv3.nsx_password)
|
|
self.nsxp = nsxp_client.NSXPClient(CONF.nsxv3.nsx_manager,
|
|
CONF.nsxv3.nsx_user,
|
|
CONF.nsxv3.nsx_password)
|
|
|
|
def _create_router(self, router_name=None, admin_state_up=True,
|
|
external_network_id=None, enable_snat=None,
|
|
set_gateway=True, **kwargs):
|
|
ext_gw_info = {}
|
|
if external_network_id:
|
|
ext_gw_info['network_id'] = external_network_id
|
|
if enable_snat is not None:
|
|
ext_gw_info['enable_snat'] = enable_snat
|
|
if set_gateway:
|
|
body = self.routers_client.create_router(
|
|
name=router_name, external_gateway_info=ext_gw_info,
|
|
admin_state_up=admin_state_up, **kwargs)
|
|
else:
|
|
body = self.routers_client.create_router(
|
|
name=router_name,
|
|
admin_state_up=admin_state_up,
|
|
**kwargs)
|
|
router = body.get('router', body)
|
|
self.addCleanup(self._delete_router, router)
|
|
return router
|
|
|
|
def _delete_router(self, router):
|
|
body = self.ports_client.list_ports(device_id=router['id'])
|
|
interfaces = body['ports']
|
|
for i in interfaces:
|
|
test_utils.call_and_ignore_notfound_exc(
|
|
self.routers_client.remove_router_interface, router['id'],
|
|
subnet_id=i['fixed_ips'][0]['subnet_id'])
|
|
self.routers_client.delete_router(router['id'])
|
|
|
|
def _test_ping_from_external_network(self, fip_ip):
|
|
out = os.popen('ping -c 2 %s' % fip_ip).read().strip()
|
|
return out
|
|
|
|
def _add_static_route(self, vm1_real_ip, router):
|
|
r_ips = router['external_gateway_info']['external_fixed_ips']
|
|
r_ip = r_ips[0]['ip_address']
|
|
os.popen('route add -host %s gw %s' % (vm1_real_ip, r_ip))
|
|
|
|
def _test_tcp_port_from_external_network(self, fip_ip):
|
|
nc_cmd = subprocess.Popen(['nc -zvw01 192.166.1.2 22'],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, shell=True)
|
|
output1, output2 = nc_cmd.communicate()
|
|
return output2
|
|
|
|
def create_network_topo(self, create_instance=False, enable_snat=True,
|
|
set_gateway=True):
|
|
network = \
|
|
self.create_topology_network(network_name="fw-network",
|
|
networks_client=self.networks_client)
|
|
# Create router topo
|
|
kwargs = {"router:external": "True"}
|
|
ext_nw = \
|
|
self.create_topology_network(network_name="ext_new",
|
|
networks_client=self.nw_client,
|
|
**kwargs)
|
|
subnet_name = 'ext_subnet'
|
|
subnet_args = {'enable_dhcp': False}
|
|
self.create_topology_subnet(subnet_name, ext_nw,
|
|
cidr='172.24.5.0/24',
|
|
**subnet_args)
|
|
router = self._create_router(
|
|
router_name=data_utils.rand_name('router-default1'),
|
|
external_network_id=ext_nw['id'],
|
|
enable_snat=enable_snat, set_gateway=set_gateway)
|
|
subnet_name = 'fw-subnet'
|
|
self.create_topology_subnet(subnet_name, network,
|
|
router_id=router['id'])
|
|
if create_instance:
|
|
if set_gateway and enable_snat:
|
|
floating_ip = True
|
|
else:
|
|
floating_ip = False
|
|
vm1 = self.create_topology_instance(
|
|
"state_vm_1", [network],
|
|
create_floating_ip=floating_ip)
|
|
if set_gateway and enable_snat:
|
|
ips = self.topology_servers['state_vm_1']['floating_ips']
|
|
real_ip = ips[0]['fixed_ip_address']
|
|
float_ip = ips[0]['floating_ip_address']
|
|
else:
|
|
ips = self.topology_servers['state_vm_1']['addresses']
|
|
real_ip = ips[network['name']][0]['addr']
|
|
float_ip = None
|
|
else:
|
|
real_ip = None
|
|
float_ip = None
|
|
vm1 = None
|
|
net_topo = dict(project=network['project_id'], router=router,
|
|
vm1_real_ip=real_ip,
|
|
vm1_float_ip=float_ip, server=vm1)
|
|
return net_topo
|
|
|
|
@decorators.attr(type='nsxp')
|
|
@decorators.idempotent_id('131288d7-9513-4b1e-a11d-15840c8e3f13')
|
|
def test_vrf_basic_traffic_test_and_backend_verification(self):
|
|
"""
|
|
"""
|
|
net_topo = self.create_network_topo(create_instance=True)
|
|
# verify ext-VM to intVM floating IP ping
|
|
out = self._test_ping_from_external_network(net_topo['vm1_float_ip'])
|
|
self.assertIn("2 received", str(out))
|