x/vmware-nsx
Code Issues Proposed changes

Remove six package usage

Browse Source 
Since py2 is no longer supported, built in methods can replace the
six package usage, as been done in the neutron project

Change-Id: I922963fbbcc0ab263e1f6e56907b73b007015a75
This commit is contained in:
asarfaty 2020-07-19 11:51:34 +02:00
parent 0f42f01267
commit 00f43c2b63
37 changed files with 62 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pylintrc
View File

@ -135,7 +135,7 @@ deprecated-modules=
[TYPECHECK]
# List of module names for which member attributes should not be checked
ignored-modules=six.moves,_MovedItems
ignored-modules=_MovedItems
[REPORTS]
# Tells whether to display a full report or only the messages

1
lower-constraints.txt
View File

@ -39,7 +39,6 @@ pylint==1.7.1
python-openstackclient==5.3.0
reno==2.5.0
requests==2.14.2
six==1.11.0
SQLAlchemy==1.2.0
stestr==1.0.0
stevedore==1.20.0

1
requirements.txt
View File

@ -9,7 +9,6 @@ requests>=2.14.2 # Apache-2.0
netaddr>=0.7.18 # BSD
tenacity>=5.0.2 # Apache-2.0
SQLAlchemy!=1.1.5,!=1.1.6,!=1.1.7,!=1.1.8,>=1.2.0 # MIT
six>=1.11.0 # MIT
stevedore>=1.20.0 # Apache-2.0
neutron-lib>=2.0.0 # Apache-2.0
osc-lib>=1.14.0 # Apache-2.0

2
vmware_nsx/api_client/__init__.py
View File

@ -15,7 +15,7 @@
# under the License.
#
from six.moves import http_client as httplib
from http import client as httplib
from vmware_nsx._i18n import _

6
vmware_nsx/api_client/base.py
View File

@ -15,12 +15,11 @@
# under the License.
import abc
from http import client as httplib
import time
from oslo_config import cfg
from oslo_log import log as logging
import six
from six.moves import http_client as httplib
from vmware_nsx import api_client
@ -31,8 +30,7 @@ DEFAULT_CONCURRENT_CONNECTIONS = 3
DEFAULT_CONNECT_TIMEOUT = 5
@six.add_metaclass(abc.ABCMeta)
class ApiClientBase(object):
class ApiClientBase(object, metaclass=abc.ABCMeta):
"""An abstract baseclass for all API client implementations."""
def _create_connection(self, host, port, is_ssl):

3
vmware_nsx/api_client/client.py
View File

@ -14,10 +14,9 @@
# License for the specific language governing permissions and limitations
# under the License.
#
from http import client as httplib
from oslo_log import log as logging
from six.moves import http_client as httplib
from vmware_nsx.api_client import base
from vmware_nsx.api_client import eventlet_client

4
vmware_nsx/api_client/eventlet_request.py
View File

@ -14,12 +14,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from http import client as httplib
import urllib
import eventlet
from oslo_log import log as logging
from oslo_serialization import jsonutils
from six.moves import http_client as httplib
from six.moves import urllib
from vmware_nsx._i18n import _
from vmware_nsx.api_client import request

8
vmware_nsx/api_client/request.py
View File

@ -17,15 +17,14 @@
import abc
import copy
from http import client as httplib
import socket
import time
import urllib
import eventlet
from oslo_log import log as logging
from oslo_utils import excutils
import six
from six.moves import http_client as httplib
from six.moves import urllib
from vmware_nsx._i18n import _
from vmware_nsx import api_client
@ -40,8 +39,7 @@ DEFAULT_MAXIMUM_REQUEST_ID = 4294967295
DOWNLOAD_TIMEOUT = 180
@six.add_metaclass(abc.ABCMeta)
class ApiRequest(object):
class ApiRequest(object, metaclass=abc.ABCMeta):
'''An abstract baseclass for all ApiRequest implementations.
This defines the interface and property structure for both eventlet and

4
vmware_nsx/api_replay/client.py
View File

@ -14,8 +14,6 @@ import copy
import logging
import socket
import six
from keystoneauth1 import identity
from keystoneauth1 import session
from neutronclient.common import exceptions as n_exc
@ -394,7 +392,7 @@ class ApiReplayClient(utils.PrepareObjectForMigration):
total_num = len(routers_routes)
LOG.info("Migrating %s routers routes", total_num)
for count, (router_id, routes) in enumerate(
six.iteritems(routers_routes), 1):
routers_routes.items(), 1):
try:
self.dest_neutron.update_router(router_id,
{'router': {'routes': routes}})

5
vmware_nsx/common/driver_api.py
View File

@ -15,11 +15,8 @@
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class ExtensionDriver(object):
class ExtensionDriver(object, metaclass=abc.ABCMeta):
"""Define stable abstract interface for extension drivers.
An extension driver extends the core resources implemented by the
plugin with additional attributes. Methods that process create

3
vmware_nsx/common/utils.py
View File

@ -22,7 +22,6 @@ import hashlib
import xml.etree.ElementTree as et
import eventlet
import six
import tenacity
from tenacity import _utils as tenacity_utils
@ -135,7 +134,7 @@ def is_nsxv_dhcp_binding_supported(nsx_version):
def get_tags(**kwargs):
tags = ([dict(tag=value, scope=key)
for key, value in six.iteritems(kwargs)])
for key, value in kwargs.items()])
tags.append({"tag": NEUTRON_VERSION, "scope": "quantum"})
return sorted(tags, key=lambda x: x['tag'])

5
vmware_nsx/db/db.py
View File

@ -14,7 +14,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from sqlalchemy.orm import exc
from neutron_lib.db import api as db_api
@ -31,12 +30,12 @@ LOG = logging.getLogger(__name__)
def _apply_filters_to_query(query, model, filters, like_filters=None):
if filters:
for key, value in six.iteritems(filters):
for key, value in filters.items():
column = getattr(model, key, None)
if column:
query = query.filter(column.in_(value))
if like_filters:
for key, search_term in six.iteritems(like_filters):
for key, search_term in like_filters.items():
column = getattr(model, key, None)
if column:
query = query.filter(column.like(search_term))

3
vmware_nsx/db/nsxv_db.py
View File

@ -21,7 +21,6 @@ from neutron_lib.db import api as db_api
from oslo_db import exception as db_exc
from oslo_log import log as logging
from oslo_utils import excutils
import six
from sqlalchemy import func
from sqlalchemy.orm import exc
from sqlalchemy.sql import expression as expr
@ -113,7 +112,7 @@ def update_nsxv_router_binding(session, router_id, **kwargs):
with session.begin(subtransactions=True):
binding = (session.query(nsxv_models.NsxvRouterBinding).
filter_by(router_id=router_id).one())
for key, value in six.iteritems(kwargs):
for key, value in kwargs.items():
binding[key] = value
return binding

3
vmware_nsx/nsxlib/mh/__init__.py
View File

@ -17,7 +17,6 @@ from neutron import version
from neutron_lib import exceptions as exception
from oslo_log import log
from oslo_serialization import jsonutils
import six
from vmware_nsx._i18n import _
from vmware_nsx.api_client import exception as api_exc
@ -84,7 +83,7 @@ def format_exception(etype, e, exception_locals):
"""
msg = [_("Error. %(type)s exception: %(exc)s.") %
{'type': etype, 'exc': e}]
lcls = dict((k, v) for k, v in six.iteritems(exception_locals)
lcls = dict((k, v) for k, v in exception_locals.items()
if k != 'request')
msg.append(_("locals=[%s]") % str(lcls))
return ' '.join(msg)

3
vmware_nsx/nsxlib/mh/lsn.py
View File

@ -16,7 +16,6 @@
from neutron_lib import exceptions as exception
from oslo_log import log
from oslo_serialization import jsonutils
import six
from vmware_nsx._i18n import _
from vmware_nsx.api_client import exception as api_exc
@ -216,7 +215,7 @@ def _get_opts(name, value):
def lsn_port_dhcp_configure(
cluster, lsn_id, lsn_port_id, is_enabled=True, dhcp_options=None):
dhcp_options = dhcp_options or {}
opts = [_get_opts(key, val) for key, val in six.iteritems(dhcp_options)]
opts = [_get_opts(key, val) for key, val in dhcp_options.items()]
dhcp_obj = {'options': opts}
_lsn_port_configure_action(
cluster, lsn_id, lsn_port_id, 'dhcp', is_enabled, dhcp_obj)

4
vmware_nsx/osc/v2/project_plugin_map.py
View File

@ -13,8 +13,6 @@
"""Project Plugin mapping action implementations"""
import six
from openstack import exceptions as os_exceptions
from openstack import resource
from openstackclient.i18n import _
@ -89,7 +87,7 @@ class CreateProjectPluginMap(command.ShowOne):
except os_exceptions.HttpException as exc:
msg = _("Error while executing command: %s") % exc.message
if exc.details:
msg += ", " + six.text_type(exc.details)
msg += ", " + str(exc.details)
raise osc_exceptions.CommandError(msg)
display_columns, columns = _get_columns(obj)
data = utils.get_item_properties(obj, columns, formatters={})

4
vmware_nsx/plugins/common/housekeeper/base_job.py
View File

@ -17,13 +17,11 @@ import abc
from neutron_lib.plugins import directory
from oslo_log import log
import six
LOG = log.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class BaseJob(object):
class BaseJob(object, metaclass=abc.ABCMeta):
_core_plugin = None

6
vmware_nsx/plugins/common_v3/plugin.py
View File

@ -26,8 +26,6 @@ from oslo_utils import excutils
from sqlalchemy import exc as sql_exc
import webob.exc
from six import moves
from neutron.db import agents_db
from neutron.db import agentschedulers_db
from neutron.db import allowedaddresspairs_db as addr_pair_db
@ -969,11 +967,11 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
if vlan_ranges:
vlan_ids = set()
for vlan_min, vlan_max in vlan_ranges:
vlan_ids |= set(moves.range(vlan_min, vlan_max + 1))
vlan_ids |= set(range(vlan_min, vlan_max + 1))
else:
vlan_min = constants.MIN_VLAN_TAG
vlan_max = constants.MAX_VLAN_TAG
vlan_ids = set(moves.range(vlan_min, vlan_max + 1))
vlan_ids = set(range(vlan_min, vlan_max + 1))
used_ids_in_range = [binding.vlan_id for binding in bindings
if binding.vlan_id in vlan_ids]
not_allowed_in_range = set(used_ids_in_range + restricted_vlans)

3
vmware_nsx/plugins/common_v3/utils.py
View File

@ -15,7 +15,6 @@
import os
import random
import re
from six import string_types
from oslo_config import cfg
from oslo_context import context as context_utils
@ -234,7 +233,7 @@ def get_network_dns_domain(az, network):
dns_domain = None
if network.get('dns_domain'):
net_dns = network['dns_domain']
if isinstance(net_dns, string_types):
if isinstance(net_dns, str):
dns_domain = net_dns
elif hasattr(net_dns, "dns_domain"):
dns_domain = net_dns.dns_domain

5
vmware_nsx/plugins/nsx_v/drivers/abstract_router_driver.py
View File

@ -14,8 +14,6 @@
import abc
import six
from neutron.db import l3_db
from neutron.db import models_v2
from neutron_lib.callbacks import events
@ -28,8 +26,7 @@ from vmware_nsx.plugins.nsx_v import availability_zones as nsx_az
from vmware_nsx.plugins.nsx_v.vshield import edge_utils
@six.add_metaclass(abc.ABCMeta)
class RouterAbstractDriver(object):
class RouterAbstractDriver(object, metaclass=abc.ABCMeta):
"""Abstract router driver that expose API for nsxv plugin."""
@abc.abstractmethod

23
vmware_nsx/plugins/nsx_v/plugin.py
View File

@ -69,8 +69,6 @@ from oslo_serialization import jsonutils
from oslo_utils import excutils
from oslo_utils import netutils
from oslo_utils import uuidutils
import six
from six import moves
from sqlalchemy.orm import exc as sa_exc
from neutron.api import extensions as neutron_extensions
@ -1200,11 +1198,11 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
if vlan_ranges:
vlan_ids = set()
for vlan_min, vlan_max in vlan_ranges:
vlan_ids |= set(moves.range(vlan_min, vlan_max + 1))
vlan_ids |= set(range(vlan_min, vlan_max + 1))
else:
vlan_min = constants.MIN_VLAN_TAG
vlan_max = constants.MAX_VLAN_TAG
vlan_ids = set(moves.range(vlan_min, vlan_max + 1))
vlan_ids = set(range(vlan_min, vlan_max + 1))
used_ids_in_range = set([binding.vlan_id for binding in bindings
if binding.vlan_id in vlan_ids])
free_ids = list(vlan_ids ^ used_ids_in_range)
@ -1328,8 +1326,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
# Delete VLAN networks on other DVSes if it
# fails to be created on one DVS and reraise
# the original exception.
for dvsmoref, netmoref in six.iteritems(
dvs_pg_mappings):
for dvsmoref, netmoref in dvs_pg_mappings.items():
self._delete_backend_network(
netmoref, dvsmoref)
dvs_pg_mappings[dvs_id] = net_moref
@ -1347,10 +1344,8 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
# Delete VLAN networks on other DVSes if it
# fails to be created on one DVS and reraise
# the original exception.
for dvsmoref, netmoref in six.iteritems(
dvs_pg_mappings):
self._delete_backend_network(
netmoref, dvsmoref)
for dvsm, netm in dvs_pg_mappings.items():
self._delete_backend_network(netm, dvsm)
try:
net_data[psec.PORTSECURITY] = net_data.get(psec.PORTSECURITY, True)
if not cfg.CONF.nsxv.spoofguard_enabled:
@ -1435,7 +1430,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
network_type == c_utils.NsxVNetworkTypes.FLAT):
# Save netmoref to dvs id mappings for VLAN network
# type for future access.
for dvs_id, netmoref in six.iteritems(dvs_pg_mappings):
for dvs_id, netmoref in dvs_pg_mappings.items():
nsx_db.add_neutron_nsx_network_mapping(
session=context.session,
neutron_id=new_net['id'],
@ -1464,7 +1459,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
self._delete_backend_network(net_moref)
elif (network_type and
network_type != c_utils.NsxVNetworkTypes.PORTGROUP):
for dvsmrf, netmrf in six.iteritems(dvs_pg_mappings):
for dvsmrf, netmrf in dvs_pg_mappings.items():
self._delete_backend_network(netmrf, dvsmrf)
LOG.exception('Failed to create network')
@ -1734,7 +1729,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
# Delete VLAN networks on other DVSes if it
# fails to be created on one DVS and reraise
# the original exception.
for dvsmoref, netmoref in six.iteritems(dvs_pg_mappings):
for dvsmoref, netmoref in dvs_pg_mappings.items():
self._delete_backend_network(netmoref, dvsmoref)
dvs_pg_mappings[dvs_id] = net_moref
@ -1903,7 +1898,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
# Save netmoref to dvs id mappings for VLAN network
# type for future access.
dvs_ids = []
for dvs_id, netmoref in six.iteritems(new_dvs_pg_mappings):
for dvs_id, netmoref in new_dvs_pg_mappings.items():
nsx_db.add_neutron_nsx_network_mapping(
session=context.session,
neutron_id=id,

8
vmware_nsx/plugins/nsx_v/vshield/common/VcnsApiClient.py
View File

@ -19,7 +19,6 @@ import xml.etree.ElementTree as et
from oslo_context import context as context_utils
from oslo_serialization import jsonutils
import requests
import six
from vmware_nsx.plugins.nsx_v.vshield.common import exceptions
@ -38,7 +37,7 @@ def _xmldump(obj):
config = ""
attr = ""
if isinstance(obj, dict):
for key, value in six.iteritems(obj):
for key, value in obj.items():
if key.startswith('__'):
# Skip the key and evaluate it's value.
a, x = _xmldump(value)
@ -88,9 +87,8 @@ class VcnsApiHelper(object):
def __init__(self, address, user, password, format='json', ca_file=None,
insecure=True, timeout=None):
# pylint: disable=deprecated-method
encode_fn = base64.encodestring if six.PY2 else base64.encodebytes
self.authToken = encode_fn(six.b("%s:%s" % (user, password)))
self.authToken = base64.encodebytes(
bytes("%s:%s" % (user, password), 'utf-8'))
self.user = user
self.passwd = password
self.address = address

12
vmware_nsx/plugins/nsx_v/vshield/edge_utils.py
View File

@ -33,8 +33,6 @@ from oslo_serialization import jsonutils
from oslo_utils import excutils
from oslo_utils import timeutils
from oslo_utils import uuidutils
import six
from six import moves
from sqlalchemy import exc as db_base_exc
from sqlalchemy.orm import exc as sa_exc
@ -254,7 +252,7 @@ class EdgeManager(object):
availability_zone=None):
router_ids = [(vcns_const.BACKUP_ROUTER_PREFIX +
_uuid())[:vcns_const.EDGE_NAME_LEN]
for i in moves.range(num)]
for i in range(num)]
for router_id in router_ids:
nsxv_db.add_nsxv_router_binding(
@ -1061,12 +1059,12 @@ class EdgeManager(object):
works at NSXv version 6.2.3 or higher.
"""
for binding in static_bindings:
if 'dhcpOptions' not in six.iterkeys(binding):
if 'dhcpOptions' not in iter(binding):
binding['dhcpOptions'] = {}
if 'option121' not in six.iterkeys(binding['dhcpOptions']):
if 'option121' not in iter(binding['dhcpOptions']):
binding['dhcpOptions']['option121'] = {'staticRoutes': []}
binding_opt121 = binding['dhcpOptions']['option121']
if 'staticRoutes' not in six.iterkeys(binding_opt121):
if 'staticRoutes' not in iter(binding_opt121):
binding_opt121['staticRoutes'] = []
binding_opt121['staticRoutes'].append({
'destinationSubnet': dest_cidr,
@ -1079,7 +1077,7 @@ class EdgeManager(object):
We can add the MTU via dhcp option26.
This func can only works at NSXv version 6.2.3 or higher.
"""
if 'dhcpOptions' not in six.iterkeys(static_binding):
if 'dhcpOptions' not in iter(static_binding):
static_binding['dhcpOptions'] = {}
static_binding['dhcpOptions']['option26'] = mtu
return static_binding

4
vmware_nsx/plugins/nsx_v/vshield/nsxv_edge_cfg_obj.py
View File

@ -16,13 +16,11 @@
import abc
from oslo_serialization import jsonutils
import six
from vmware_nsx.plugins.nsx_v.vshield import vcns
@six.add_metaclass(abc.ABCMeta)
class NsxvEdgeCfgObj(object):
class NsxvEdgeCfgObj(object, metaclass=abc.ABCMeta):
def __init__(self):
return

5
vmware_nsx/plugins/nsx_v/vshield/tasks/tasks.py
View File

@ -22,7 +22,6 @@ from eventlet import greenthread
from neutron_lib import exceptions
from oslo_log import log as logging
from oslo_service import loopingcall
import six
from vmware_nsx._i18n import _
from vmware_nsx.plugins.nsx_v.vshield.tasks import constants
@ -353,7 +352,7 @@ class TaskManager(object):
def show_pending_tasks(self):
for task in self._tasks_queue:
LOG.info(str(task))
for resource, tasks in six.iteritems(self._tasks):
for resource, tasks in self._tasks.items():
for task in tasks:
LOG.info(str(task))
if self._main_thread_exec_task:
@ -361,7 +360,7 @@ class TaskManager(object):
def count(self):
count = 0
for resource_id, tasks in six.iteritems(self._tasks):
for resource_id, tasks in self._tasks.items():
count += len(tasks)
return count

5
vmware_nsx/plugins/nsx_v/vshield/vcns.py
View File

@ -19,7 +19,6 @@ from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import strutils
import six
from vmware_nsx.common import nsxv_constants
from vmware_nsx.common import utils
@ -1007,7 +1006,7 @@ class Vcns(object):
config = self.get_tuning_configuration()
LOG.debug("Tuning configuration: %s", config)
tuning = et.Element('tuningConfiguration')
for opt, val in six.iteritems(config):
for opt, val in config.items():
child = et.Element(opt)
if opt == 'aggregatePublishing':
child.text = 'true'
@ -1021,7 +1020,7 @@ class Vcns(object):
uri = "/api/4.0/edgePublish/tuningConfiguration"
config = self.get_tuning_configuration()
tuning = et.Element('tuningConfiguration')
for opt, val in six.iteritems(config):
for opt, val in config.items():
child = et.Element(opt)
if (opt == 'edgeVCpuReservationPercentage' or
opt == 'edgeMemoryReservationPercentage'):

5
vmware_nsx/services/ipam/common/driver.py
View File

@ -16,8 +16,6 @@
import abc
import six
from oslo_log import log as logging
from neutron.ipam import driver as ipam_base
@ -33,8 +31,7 @@ from vmware_nsx.extensions import projectpluginmap
LOG = logging.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class NsxIpamBase(object):
class NsxIpamBase(object, metaclass=abc.ABCMeta):
@classmethod
def get_core_plugin(cls):
return directory.get_plugin()

3
vmware_nsx/shell/admin/plugins/common/utils.py
View File

@ -16,7 +16,6 @@ import sys
from unittest import mock
from neutron import quota
import six
from vmware_nsx._i18n import _
from vmware_nsx.db import db
from vmware_nsx.shell import resources as nsxadmin
@ -83,7 +82,7 @@ def query_yes_no(question, default="yes"):
while True:
sys.stdout.write(question + prompt)
choice = six.moves.input().lower()
choice = input().lower()
if default is not None and choice == '':
return valid[default]
elif choice in valid:

3
vmware_nsx/tests/unit/nsx_v/test_plugin.py
View File

@ -63,7 +63,6 @@ from neutron_lib.services.qos import constants as qos_consts
from neutron_lib.utils import helpers
from oslo_config import cfg
from oslo_utils import uuidutils
import six
from testtools import matchers
import webob.exc
@ -4414,7 +4413,7 @@ class TestVdrTestCase(L3NatTest, L3NatTestCaseBase,
data = {'tenant_id': 'whatever'}
data['name'] = 'router1'
data['distributed'] = dist_input
for k, v in six.iteritems(kwargs):
for k, v in kwargs.items():
data[k] = v
router_req = self.new_create_request(
'routers', {'router': data}, self.fmt)

3
vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py
View File

@ -18,7 +18,6 @@ import xml.etree.ElementTree as ET
import netaddr
from oslo_serialization import jsonutils
from oslo_utils import uuidutils
import six
from vmware_nsx._i18n import _
from vmware_nsx.plugins.nsx_v.vshield.common import constants
@ -79,7 +78,7 @@ class FakeVcns(object):
self._fake_nsx_api = fake_nsx_api
def _validate_edge_name(self, name):
for edge_id, edge in six.iteritems(self._edges):
for edge_id, edge in self._edges.items():
if edge['name'] == name:
return False
return True

3
vmware_nsx/tests/unit/nsx_v/vshield/test_edge_utils.py
View File

@ -20,7 +20,6 @@ from neutron_lib import context
from neutron_lib import exceptions as n_exc
from oslo_config import cfg
from oslo_utils import uuidutils
from six import moves
from vmware_nsx.common import config as conf
from vmware_nsx.common import exceptions as nsx_exc
@ -388,7 +387,7 @@ class EdgeManagerTestCase(EdgeUtilsTestCaseMixin):
'appliance_size': size,
'edge_type': edge_type,
'availability_zone': availability_zone.name}
for i in moves.range(num)]
for i in range(num)]
def _create_available_router_bindings(
self, num, size=nsxv_constants.LARGE,

3
vmware_nsx/tests/unit/nsx_v/vshield/test_vcns_driver.py
View File

@ -19,7 +19,6 @@ from eventlet import greenthread
from neutron.tests import base
from neutron_lib import context as neutron_context
from oslo_config import cfg
import six
from vmware_nsx.common import exceptions as nsxv_exc
from vmware_nsx.plugins.nsx_v import availability_zones as nsx_az
@ -247,7 +246,7 @@ class VcnsDriverTaskManagerTestCase(base.BaseTestCase):
# if _thread is None it means it was killed in stop()
self.assertIsNone(manager._thread)
for res, tasks in six.iteritems(alltasks):
for res, tasks in alltasks.items():
for task in tasks:
self.assertEqual(ts_const.TaskStatus.ABORT, task.status)

3
vmware_nsx/tests/unit/nsxlib/mh/nsx_cluster.py
View File

@ -15,7 +15,6 @@
from oslo_config import cfg
from oslo_log import log as logging
import six
from vmware_nsx._i18n import _
from vmware_nsx.common import exceptions
@ -47,7 +46,7 @@ class NSXCluster(object):
self._deprecated_attributes = {}
self._sanity_check(kwargs)
for opt, val in six.iteritems(self._deprecated_attributes):
for opt, val in self._deprecated_attributes.items():
LOG.deprecated(_("Attribute '%s' has been deprecated or moved "
"to a new section. See new configuration file "
"for details."), opt)

4
vmware_nsx/tests/unit/nsxlib/mh/test_lsn.py
View File

@ -14,8 +14,6 @@
# limitations under the License.
from unittest import mock
import six
from neutron.tests import base
from neutron_lib import exceptions
from oslo_serialization import jsonutils
@ -253,7 +251,7 @@ class LSNTestCase(base.BaseTestCase):
self.cluster, lsn_id, lsn_port_id, is_enabled, opts)
opt_array = [
{"name": key, "value": val}
for key, val in six.iteritems(opts)
for key, val in opts.items()
]
self.mock_request.assert_has_calls([
mock.call("PUT", "/ws.v1/lservices-node/%s/dhcp" % lsn_id,

3
vmware_nsx/tests/unit/services/lbaas/lb_data_models.py
View File

@ -28,7 +28,6 @@ and also converting to dictionaries.
from neutron.db.models import servicetype as servicetype_db
from neutron.db import models_v2
from neutron_lib.db import model_base
import six
from sqlalchemy.ext import orderinglist
from sqlalchemy.orm import collections
@ -60,8 +59,6 @@ class BaseDataModel(object):
ret[attr] = item
elif isinstance(getattr(self, attr), BaseDataModel):
ret[attr] = value.to_dict()
elif six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
return ret

37
vmware_nsx/tests/unit/services/lbaas/lb_db_models.py
View File

@ -13,8 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from neutron.db.models import servicetype as st_db
from neutron.db import models_v2
from neutron_lib.db import constants as db_const
@ -101,10 +99,7 @@ class MemberV2(model_base.BASEV2, model_base.HasId, model_base.HasProject):
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [
@ -146,10 +141,7 @@ class HealthMonitorV2(model_base.BASEV2, model_base.HasId,
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [
@ -221,10 +213,7 @@ class LoadBalancer(model_base.BASEV2, model_base.HasId, model_base.HasProject):
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [
@ -302,10 +291,7 @@ class PoolV2(model_base.BASEV2, model_base.HasId, model_base.HasProject):
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [
@ -388,10 +374,7 @@ class L7Rule(model_base.BASEV2, model_base.HasId, model_base.HasProject):
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [
@ -448,10 +431,7 @@ class L7Policy(model_base.BASEV2, model_base.HasId, model_base.HasProject):
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [
@ -530,10 +510,7 @@ class Listener(model_base.BASEV2, model_base.HasId, model_base.HasProject):
ret = {}
for attr in attributes:
value = getattr(sa_model, attr)
if six.PY2 and isinstance(value, six.text_type):
ret[attr.encode('utf8')] = value.encode('utf8')
else:
ret[attr] = value
ret[attr] = value
return ret
ret_dict = to_dict(self, [

4
vmware_nsx/tests/unit/shell/test_admin_utils.py
View File

@ -21,7 +21,6 @@ from oslo_config import cfg
from oslo_log import _options
from oslo_log import log as logging
from oslo_utils import uuidutils
import six
from neutron.common import config as neutron_config
from neutron.db import servicetype_db # noqa
@ -52,8 +51,7 @@ NSX_INI_PATH = vmware.get_fake_conf('nsx.ini.test')
BASE_CONF_PATH = vmware.get_fake_conf('neutron.conf.test')
@six.add_metaclass(abc.ABCMeta)
class AbstractTestAdminUtils(base.BaseTestCase):
class AbstractTestAdminUtils(base.BaseTestCase, metaclass=abc.ABCMeta):
def setUp(self):
cfg.CONF.unregister_opts(_options.common_cli_opts)