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

tempest changes for python3 in nsxv related

Browse Source 
Change-Id: I72751f0e6663b933f2ff2d6c4218c1c30b3688a3
This commit is contained in:
dkumbhar 2020-03-31 07:52:38 +00:00
parent 3800e954ce
commit 68242e8c15
10 changed files with 65 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
vmware_nsx_tempest_plugin/services/nsxv_client.py
View File

@ -175,7 +175,7 @@ class VSMClient(object):
LOG.debug(msg)
pages = utils.ceil(total_count, page_size)
LOG.debug("Total pages: %s" % pages)
for i in range(pages):
for i in range(int(pages) + 1):
start_index = page_size * i
params = {'startindex': start_index}
response = self.get(params=params)
@ -227,7 +227,7 @@ class VSMClient(object):
page_size)
LOG.debug(msg)
pages = utils.ceil(total_count, page_size)
for i in range(pages):
for i in range(int(pages) + 1):
start_index = page_size * i
params = {'startindex': start_index}
response = self.get(params=params)

28
vmware_nsx_tempest_plugin/tests/nsxv/api/test_router_sizes.py
View File

@ -22,7 +22,6 @@ from tempest.api.network import base
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib.services import network as net_clients
from tempest import test
from vmware_nsx_tempest_plugin.services import nsxv_client
@ -62,6 +61,8 @@ class RouterSizeBaseTest(base.BaseAdminNetworkTest):
@classmethod
def setup_clients(cls):
super(RouterSizeBaseTest, cls).setup_clients()
cls.cmgr_adm = cls.get_client_manager('admin')
cls.routers_client = cls.cmgr_adm.routers_client
@classmethod
def resource_setup(cls):
@ -75,19 +76,6 @@ class RouterSizeBaseTest(base.BaseAdminNetworkTest):
manager_ip, CONF.nsxv.user, CONF.nsxv.password)
cls.routers = []
def setUp(self):
super(RouterSizeBaseTest, self).setUp()
params = {'build_interval': self.routers_client.build_interval,
'build_timeout': self.routers_client.build_timeout}
http_timeout = CONF.nsxv.create_router_http_timeout
self.router_sizes_client = net_clients.RoutersClient(
self.routers_client.auth_provider,
self.routers_client.service,
self.routers_client.region,
self.routers_client.endpoint_type,
http_timeout=http_timeout,
**params)
def create_exclusive_router(self, router_size):
name = data_utils.rand_name('rtr1-%s' % router_size)
LOG.debug("create router with size=%s", router_size)
@ -98,7 +86,7 @@ class RouterSizeBaseTest(base.BaseAdminNetworkTest):
external_gateway_info=ext_gw_info,
router_type='exclusive',
router_size=router_size)
router = self.router_sizes_client.create_router(**rtr_cfg)
router = self.routers_client.create_router(**rtr_cfg)
router = router.get('router', router)
self.routers.append(router)
self.assertEqual(router['name'], name)
@ -107,7 +95,7 @@ class RouterSizeBaseTest(base.BaseAdminNetworkTest):
def change_router_size(self, router, new_router_size):
LOG.debug("update router to size=%s", new_router_size)
update_router = self.router_sizes_client.update_router(
update_router = self.routers_client.update_router(
router['id'], router_size=new_router_size)['router']
self.assertEqual(update_router['router_size'], new_router_size)
self.check_router_nsx_name(update_router, new_router_size)
@ -134,17 +122,17 @@ class RouterSizeBaseTest(base.BaseAdminNetworkTest):
del_interval=1.5):
router = self.create_exclusive_router(router_size)
updated_name = 'updated-' + router['name']
update_router = self.router_sizes_client.update_router(
update_router = self.routers_client.update_router(
router['id'], name=updated_name)['router']
self.assertEqual(update_router['name'], updated_name)
# change router name, the backend also change
router = self.router_sizes_client.show_router(
router = self.routers_client.show_router(
router['id'])['router']
nsxv_edge_name = self.check_router_nsx_name(router, router_size)
# Delete the exclusive router and verify it has been deleted
# from nsxv backend
self.router_sizes_client.delete_router(router['id'])
list_body = self.router_sizes_client.list_routers()
self.routers_client.delete_router(router['id'])
list_body = self.routers_client.list_routers()
routers_list = [r['id'] for r in list_body['routers']]
self.assertNotIn(router['id'], routers_list)
wait_till = time.time() + del_waitfor

4
vmware_nsx_tempest_plugin/tests/nsxv/api/test_v1_fwaas.py
View File

@ -616,7 +616,9 @@ class FWaaSTestJSON(base.BaseNetworkTest):
break
rules = self.vsm.get_edge_firewall_rules(edge_id)
for rule in rules:
if rule['name'] in ('VSERule', 'MDServiceIP', 'MDInterEdgeNet'):
rules_to_skip = ('VSERule', 'MDServiceIP', 'MDInterEdgeNet',
'VSEMDInterEdgeNet')
if rule['name'] in rules_to_skip:
continue
if rule_no == 1:
self.assertIn(rule['name'], firewall_rule_name_2,

10
vmware_nsx_tempest_plugin/tests/nsxv/scenario/manager_topo_deployment.py
View File

@ -263,7 +263,7 @@ class TopoDeployScenarioManager(manager.NetworkScenarioTest):
flavor = flavor or CONF.compute.flavor_ref
servers_client = servers_client or self.servers_client
create_kwargs = create_kwargs or {}
if type(tenant_id) in (str, unicode): # noqa
if type(tenant_id) in (bytes, str): # noqa
if servers_client.tenant_id != tenant_id:
create_kwargs['tenant_id'] = tenant_id
@ -449,7 +449,7 @@ class TopoDeployScenarioManager(manager.NetworkScenarioTest):
if network_name and server['addresses'].get(network_name):
s_if = network_name
else:
s_if = server['addresses'].keys()[0]
s_if = sorted(server['addresses'].keys())[0]
for s_address in server['addresses'][s_if]:
if s_address['OS-EXT-IPS:type'] == ip_type:
@ -465,7 +465,7 @@ class TopoDeployScenarioManager(manager.NetworkScenarioTest):
PING_INSESSION = 'ping-progress-in-session'
PING_DONE = 'ping-progress-completed'
PING_TIMEOUT = 'ping-progress-timeout'
if msg and type(msg) in (str, unicode): # noqa
if msg and type(msg) in (bytes, str): # noqa
xmsg = ("waitfor_host_connected ip=%(ip)s! %(msg)s" %
{'ip': host_ip, 'msg': msg})
LOG.debug(xmsg)
@ -678,7 +678,7 @@ def check_hosts_connectivity(host, dest_list, ignore_helper=None,
for dest in dest_list:
# caller can say to ignore dest ipaddr
if ('helper' in dest and
type(ignore_helper) in (str, unicode) and # noqa
type(ignore_helper) in (bytes, str) and # noqa
re.search(ignore_helper, dest['helper'], re.I)): # noqa
dest['reachable'] = None
continue
@ -803,7 +803,7 @@ def copy_file_to_host(file_from, dest, host, username, pkey):
"-i %(pkey)s %(file1)s %(dest)s" % {'pkey': pkey,
'file1': file_from,
'dest': dest}
args = shlex.split(cmd.encode('utf-8'))
args = shlex.split(cmd)
subprocess_args = {'stdout': subprocess.PIPE,
'stderr': subprocess.STDOUT}
proc = subprocess.Popen(args, **subprocess_args)

2
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_dvr_basic_ops.py
View File

@ -131,7 +131,7 @@ class TestDvrBasicOps(manager.NetworkScenarioTest):
namestart=namestart)
router_kwargs = dict(client=routers_client, namestart=namestart)
for k in kwargs.keys():
for k in sorted(kwargs.keys()):
if k in ('distributed', 'router_type', 'router_size'):
router_kwargs[k] = kwargs.pop(k)
router = self._create_router(**router_kwargs)

13
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_lbaas_round_robin_ops.py
View File

@ -228,13 +228,13 @@ class LBaasRoundRobinBaseTest(dmgr.TopoDeployScenarioManager):
# need to wait for web server to be able to response
time.sleep(self.web_service_start_delay)
for server in server_list:
server_name = server['name']
server_name = server['name'].encode()
fip = server['_floating_ip']
web_fip = fip['floating_ip_address']
response = self.send_request(web_fip)
# by design, each lbaas member server response its server_name
self.assertEqual(response, server_name)
self.server_names.append(server_name)
self.server_names.append(server_name.decode())
def start_web_server(self, server, server_fip, server_name):
"""start server's web service which return its server_name."""
@ -250,10 +250,12 @@ class LBaasRoundRobinBaseTest(dmgr.TopoDeployScenarioManager):
'charset=UTF-8\r\n\r\n%s"; cat >/dev/null')
with tempfile.NamedTemporaryFile() as script:
script.write(resp % (len(server_name), server_name))
server_name = server_name.encode('utf-8')
script.write(resp.encode('utf-8') % (len(server_name),
server_name))
script.flush()
with tempfile.NamedTemporaryFile() as key:
key.write(private_key)
key.write(private_key.encode('utf-8'))
key.flush()
dmgr.copy_file_to_host(script.name,
"/tmp/script",
@ -347,7 +349,7 @@ class LBaasRoundRobinBaseTest(dmgr.TopoDeployScenarioManager):
send_counts = send_counts or self.poke_counters
send_counts = (send_counts * 2) / 2
url_path = "http://{0}/{1}".format(self.vip_ip_address, start_path)
for x in range(send_counts):
for x in range(int(send_counts)):
resp = http.request('GET', url_path)
if resp.status == 200:
self.count_response(resp.data.strip())
@ -366,6 +368,7 @@ class LBaasRoundRobinBaseTest(dmgr.TopoDeployScenarioManager):
self.assertEqual(self.http_cnt[s0], self.http_cnt[s1])
def count_response(self, response):
response = response.decode()
if response in self.http_cnt:
self.http_cnt[response] += 1
else:

8
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_multiple_transport_zones_basic_ops.py
View File

@ -197,7 +197,7 @@ class TestMultipleTransportZonesBasicOps(dmgr.TopoDeployScenarioManager):
def run_servers_connectivity_test(self, servers):
# select one from the servers
net_id_list = servers.keys()
net_id_list = sorted(servers.keys())
net_id = net_id_list[0]
other_net_id_list = net_id_list[1:]
username, password = self.get_image_userpass()
@ -234,7 +234,7 @@ class TestMultipleTransportZonesBasicOps(dmgr.TopoDeployScenarioManager):
def wait_for_servers_become_active(self, servers, client=None):
servers_client = client or self.admin_manager.servers_client
net_id_list = servers.keys()
net_id_list = sorted(servers.keys())
for net_id in net_id_list:
nsv = self.tp_svrs[net_id]
serv = nsv['server']
@ -248,7 +248,7 @@ class TestMultipleTransportZonesBasicOps(dmgr.TopoDeployScenarioManager):
def delete_all_servers(self, client=None):
servers_client = client or self.admin_manager.servers_client
for net_id in six.iterkeys(self.tp_svrs):
for net_id in sorted(six.iterkeys(self.tp_svrs)):
server = self.tp_svrs[net_id]['server']
test_utils.call_and_ignore_notfound_exc(
servers_client.delete, server['id'])
@ -259,7 +259,7 @@ class TestMultipleTransportZonesBasicOps(dmgr.TopoDeployScenarioManager):
router, nets = self._test_router_with_network_and_mtz_networks(
router_type)
servers_client = self.manager.servers_client
for net_id in six.iterkeys(nets):
for net_id in sorted(six.iterkeys(nets)):
s_id, network, subnet, security_group = nets[net_id]
"""
servers_client = (self.manager.servers_client if s_id is None

24
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_spoofguard_policy.py
View File

@ -195,7 +195,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
for exclude_vm in exclude_list:
if vm_id in exclude_vm:
if vm_id.encode('utf-8') in exclude_vm:
LOG.info("Vm in exclude list")
# Update Port security to disabled
port_client.update_port(
@ -204,7 +204,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
if exclude_vm in exclude_list:
if vm_id not in exclude_vm:
if vm_id.encode('utf-8') not in exclude_vm:
LOG.info("Vm not in exclude list")
# Detach interface from vm
self.interface_client.delete_interface(vm_id, port_id)
@ -319,7 +319,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
for exclude_vm in exclude_list:
if vm_id in exclude_vm:
if vm_id.encode('utf-8') in exclude_vm:
LOG.info("Vm1 in exclude list")
vm2_id = t_serv2['id']
# Update vm2 port to disable port security
@ -331,7 +331,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
exclude_list = [item.encode('utf-8') for item in items]
# Check vm2 in exclude list or not
for exclude_vm in exclude_list:
if vm2_id in exclude_vm:
if vm2_id.encode('utf-8') in exclude_vm:
LOG.info("Vm2 in exclude list")
vm3_id = t_serv3['id']
# Update vm3 port to enable port security
@ -343,7 +343,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
exclude_list = [item.encode('utf-8') for item in items]
# Check vm3 in exclude list or not
for exclude_vm in exclude_list:
if vm3_id in exclude_vm:
if vm3_id.encode('utf-8') in exclude_vm:
LOG.info("Vm3 in exclude list")
# Update vm1 port to enable port security
port_client.update_port(
@ -353,7 +353,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
exclude_list = [item.encode('utf-8') for item in items]
# Check vm should not be in exclude list
for exclude_vm in exclude_list:
if vm_id not in exclude_vm:
if vm_id.encode('utf-8') not in exclude_vm:
LOG.info("Vm1 not in exclude list")
@decorators.attr(type='nsxv')
@ -372,7 +372,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
for exclude_vm in exclude_list:
if vm_id in exclude_vm:
if vm_id.encode('utf-8') in exclude_vm:
LOG.info("Vm in exclude list")
port_client.update_port(
port_id=port_id,
@ -380,7 +380,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
if exclude_vm in exclude_list:
if vm_id not in exclude_vm:
if vm_id.encode('utf-8') not in exclude_vm:
LOG.info("Vm not in exclude list")
self.interface_client.delete_interface(vm_id, port_id)
@ -433,7 +433,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
for exclude_vm in exclude_list:
if vm_id in exclude_vm:
if vm_id.encode('utf-8') in exclude_vm:
LOG.info("Vm in exclude list")
name = 'disabled-port-security-port2'
kwargs = {'name': name, 'network_id': net_id,
@ -444,7 +444,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
for exclude_vm in exclude_list:
if vm_id in exclude_vm:
if vm_id.encode('utf-8') in exclude_vm:
LOG.info("Vm in exclude list")
port_client.update_port(
port_id=port2_id,
@ -452,7 +452,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
for exclude_vm in exclude_list:
if vm_id in exclude_vm:
if vm_id.encode('utf-8') in exclude_vm:
LOG.info("Vm in exclude list")
port_client.update_port(
port_id=port1_id,
@ -460,7 +460,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps):
items = self.vsm.get_excluded_vm_name_list()
exclude_list = [item.encode('utf-8') for item in items]
if exclude_vm in exclude_list:
if vm_id not in exclude_vm:
if vm_id.encode('utf-8') not in exclude_vm:
LOG.info("Vm not in exclude list")
self.interface_client.delete_interface(vm_id, port1_id)
self.interface_client.delete_interface(vm_id, port2_id)

14
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_v1_fwaas_basic_ops.py
View File

@ -274,6 +274,14 @@ class FWaaSTestBasicOps(dmgr.TopoDeployScenarioManager):
self.keypairs = {}
router = self.create_router_by_type(router_type)
self.addCleanup(self._delete_router_if_exists, router)
if protocol_name != 'icmp':
body = self.fwaasv1_client.create_firewall_rule(
name=data_utils.rand_name("fw-rule-icmp"),
action="allow",
protocol=constants.ICMP_PROTOCOL)
fw_rule_id2 = body['firewall_rule']['id']
firewall_name = self._create_firewall_rule_name(body)
self.addCleanup(self._delete_rule_if_exists, fw_rule_id2)
body = self.fwaasv1_client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action="allow",
@ -290,6 +298,9 @@ class FWaaSTestBasicOps(dmgr.TopoDeployScenarioManager):
# Insert rule to firewall policy
self.fwaasv1_client.insert_firewall_rule_in_policy(
fw_policy_id, fw_rule_id1, '', '')
if protocol_name != 'icmp':
self.fwaasv1_client.insert_firewall_rule_in_policy(
fw_policy_id, fw_rule_id2, '', '')
else:
fw_policy_id = policy
# Create firewall
@ -313,6 +324,9 @@ class FWaaSTestBasicOps(dmgr.TopoDeployScenarioManager):
t_security_group['id'])
rule = {'direction': 'ingress', 'protocol': 'tcp'}
self._create_security_group_rule(secgroup=t_security_group, **rule)
icmp_rule = {'direction': 'ingress', 'protocol': 'icmp'}
self._create_security_group_rule(secgroup=t_security_group,
**icmp_rule)
rule = {'direction': 'ingress'}
rule_id = self._create_security_group_rule(secgroup=t_security_group,
**rule)['id']

16
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_v1_lbaas_basic_ops.py
View File

@ -17,9 +17,9 @@ import shlex
import subprocess
import tempfile
import time
#import urllib.error
#import urllib.parse
#import urllib.request
# import urllib.error
# import urllib.parse
# import urllib.request
import six
from tempest.common import utils
@ -270,7 +270,7 @@ class TestLBaaSBasicOps(manager.NetworkScenarioTest):
try:
resp = None
# TODO(Puneet)
#resp = urllib.request.urlopen(
# resp = urllib.request.urlopen(
# "http://{0}:{1}/".format(ip, port))
if resp.getcode() == 200:
return True
@ -278,7 +278,7 @@ class TestLBaaSBasicOps(manager.NetworkScenarioTest):
except IOError:
return False
# TODO(Puneet)
#except urllib.error.HTTPError:
# except urllib.error.HTTPError:
# return False
timeout = CONF.validation.ping_timeout
start = time.time()
@ -348,7 +348,7 @@ class TestLBaaSBasicOps(manager.NetworkScenarioTest):
port_id = vip['port_id']
floating_ip = self.create_floating_ip(vip, public_network_id,
port_id=port_id)
#?# self.floating_ips.setdefault(vip_id, [])
# ?# self.floating_ips.setdefault(vip_id, [])
self.floating_ips[vip_id].append(floating_ip)
# Check for floating ip status before you check load-balancer
self.check_floating_ip_status(floating_ip, "ACTIVE")
@ -409,7 +409,7 @@ class TestLBaaSBasicOps(manager.NetworkScenarioTest):
try:
server = None
# TODO(Puneet)
#server = urllib.request.urlopen(
# server = urllib.request.urlopen(
# "http://{0}/".format(vip_ip)).read()
counters[server] += 1
# HTTP exception means fail of server, so don't increase counter
@ -437,7 +437,7 @@ def copy_file_to_host(file_from, dest, host, username, pkey):
"-i %(pkey)s %(file1)s %(dest)s" % {'pkey': pkey,
'file1': file_from,
'dest': dest}
args = shlex.split(cmd.encode('utf-8'))
args = shlex.split(cmd)
subprocess_args = {'stdout': subprocess.PIPE,
'stderr': subprocess.STDOUT}
proc = subprocess.Popen(args, **subprocess_args)