openstack/tripleo-validations
Code Issues Proposed changes

Removing six from modules

Browse Source 
Imports and uses of six module were removed from the project.
Logic dependent on Py2/3 usage was removed and calls to functions were replaced
with closest functional equivalent.

After merge of this change, the project will no longer be Python 2 compatible
in any way.

Note: Do not backport past train branch

Signed-off-by: Jiri Podivin <jpodivin@redhat.com>
Change-Id: Iebdaa06b6fb173d2be9921ee87a2d5db5fe84240
This commit is contained in:
Jiri Podivin 2022-10-04 12:48:20 +02:00
parent 9960993bae
commit 0d3612e2a1
8 changed files with 32 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
doc/requirements.txt
View File

@ -5,4 +5,3 @@ reno>=3.1.0 # Apache-2.0
doc8>=0.8.0 # Apache-2.0
bashate>=0.6.0 # Apache-2.0
ruamel.yaml>=0.15.5 # MIT
six>=1.11.0 # MIT

3
doc/source/_exts/generate_validations_doc.py
View File

@ -14,7 +14,6 @@
from glob import glob
import os
import six
import yaml
DEFAULT_METADATA = {
@ -169,7 +168,7 @@ def get_groups():
groups = set()
contents = parse_groups_file()
for group_name in six.iterkeys(contents):
for group_name in contents.keys():
groups.add(group_name)
return groups, contents

4
library/ceph_pools_pg_protection.py
View File

@ -17,8 +17,6 @@
Used by the ceph-pg validation.
"""
import six
from yaml import safe_load as yaml_safe_load
from ansible.module_utils.basic import AnsibleModule
@ -162,7 +160,7 @@ def check_pg_num(pool, pg_num, size, num_osds=0, max_pgs_per_osd=200, pools={}):
projected = 0
if len(pool) < 0:
projected = projected + (pg_num * size)
for pool_name, pool_sizes in six.iteritems(pools):
for pool_name, pool_sizes in pools.items():
if pool_name == pool:
projected = projected + (pg_num * size)
else:

40
library/network_environment.py
View File

@ -23,8 +23,6 @@ import itertools
import netaddr
import os.path
import six
from ansible.module_utils.basic import AnsibleModule
# from os_net_config import validator
@ -89,7 +87,7 @@ def open_network_environment_files(netenv_path, template_files):
.format(netenv_path, e)])
nic_configs = []
resource_registry = network_data.get('resource_registry', {})
for nic_name, relative_path in six.iteritems(resource_registry):
for nic_name, relative_path in iter(resource_registry.items()):
if nic_name.endswith("Net::SoftwareConfig"):
nic_config_path = os.path.normpath(
os.path.join(os.path.dirname(netenv_path), relative_path))
@ -119,9 +117,9 @@ def validate_network_environment(network_data, nic_configs):
poolsinfo = {}
vlaninfo = {}
staticipinfo = {}
parameter_defaults = network_data.get('parameter_defaults', {})
for item, data in six.iteritems(network_data.get('parameter_defaults',
{})):
for item, data in parameter_defaults.items():
if item.endswith('NetCidr'):
cidrinfo[item] = data
elif item.endswith('AllocationPools'):
@ -158,7 +156,7 @@ def check_nic_configs(path, nic_data):
if not isinstance(resources, collections.abc.Mapping):
return ["The nic_data must contain the 'resources' key and it must be "
"a dictionary."]
for name, resource in six.iteritems(resources):
for name, resource in iter(resources.items()):
try:
nested_path = [
('properties', collections.abc.Mapping, 'dictionary'),
@ -249,7 +247,7 @@ def check_allocation_pools_pairing(filedata, pools):
if not isinstance(pools, collections.abc.Mapping):
return ["The `pools` argument must be a dictionary."]
errors = []
for poolitem, pooldata in six.iteritems(pools):
for poolitem, pooldata in iter(pools.items()):
pool_objs = []
if not isinstance(pooldata, collections.abc.Iterable):
errors.append('The IP ranges in {} must form a list.'
@ -322,7 +320,7 @@ def check_static_ip_pool_collision(static_ips, pools):
return ["The Pools input must be a dictionary."]
errors = []
pool_ranges = []
for pool_name, ranges in six.iteritems(pools):
for pool_name, ranges in iter(pools.items()):
if not isinstance(ranges, collections.abc.Iterable):
errors.append("The IP ranges in {} must form a list."
.format(pool_name))
@ -337,11 +335,11 @@ def check_static_ip_pool_collision(static_ips, pools):
continue
pool_ranges.append((pool_name, ip_range))
for role, services in six.iteritems(static_ips):
for role, services in iter(static_ips.items()):
if not isinstance(services, collections.abc.Mapping):
errors.append("The {} must be a dictionary.".format(role))
continue
for service, ips in six.iteritems(services):
for service, ips in iter(services.items()):
if not isinstance(ips, collections.abc.Iterable):
errors.append("The {}->{} must be an array."
.format(role, service))
@ -381,7 +379,7 @@ def check_vlan_ids(vlans):
return ["The vlans parameter must be a dictionary."]
errors = []
invertdict = {}
for k, v in six.iteritems(vlans):
for k, v in vlans.items():
if v not in invertdict:
invertdict[v] = k
else:
@ -403,17 +401,17 @@ def check_static_ip_in_cidr(networks, static_ips):
# TODO(shadower): Refactor this so networks are always valid and already
# converted to `netaddr.IPNetwork` here. Will be useful in the other
# checks.
for name, cidr in six.iteritems(networks):
for name, cidr in iter(networks.items()):
try:
network_ranges[name] = netaddr.IPNetwork(cidr)
except (netaddr.AddrFormatError, ValueError):
errors.append("Network '{}' has an invalid CIDR: '{}'"
.format(name, cidr))
for role, services in six.iteritems(static_ips):
for role, services in iter(static_ips.items()):
if not isinstance(services, collections.abc.Mapping):
errors.append("The {} must be a dictionary.".format(role))
continue
for service, ips in six.iteritems(services):
for service, ips in iter(services.items()):
range_name = service.title().replace('_', '') + 'NetCidr'
if range_name in network_ranges:
if not isinstance(ips, collections.abc.Iterable):
@ -439,18 +437,18 @@ def duplicate_static_ips(static_ips):
ipset = collections.defaultdict(list)
# TODO(shadower): we're doing this netsted loop multiple times. Turn it
# into a generator or something.
for role, services in six.iteritems(static_ips):
for role, services in iter(static_ips.items()):
if not isinstance(services, collections.abc.Mapping):
errors.append("The {} must be a dictionary.".format(role))
continue
for service, ips in six.iteritems(services):
for service, ips in iter(services.items()):
if not isinstance(ips, collections.abc.Iterable):
errors.append("The {}->{} must be a list."
.format(role, service))
continue
for ip in ips:
ipset[ip].append((role, service))
for ip, sources in six.iteritems(ipset):
for ip, sources in ipset.items():
if len(sources) > 1:
msg = "The {} IP address was entered multiple times: {}."
formatted_sources = ("{}[{}]"
@ -467,7 +465,7 @@ def validate_node_pool_size(plan_env_path, ip_pools_path, template_files):
param_defaults = plan_env.get('parameter_defaults')
node_counts = {
param.replace('Count', ''): count
for param, count in six.iteritems(param_defaults)
for param, count in param_defaults.items()
if param.endswith('Count') and count > 0
}
@ -481,11 +479,11 @@ def validate_node_pool_size(plan_env_path, ip_pools_path, template_files):
param_defaults = ip_pools.get('parameter_defaults')
role_pools = {
param.replace('IPs', ''): pool
for param, pool in six.iteritems(param_defaults)
for param, pool in param_defaults.items()
if param.endswith('IPs') and param.replace('IPs', '') in node_counts
}
for role, node_count in six.iteritems(node_counts):
for role, node_count in iter(node_counts.items()):
try:
pools = role_pools[role]
except KeyError:
@ -494,7 +492,7 @@ def validate_node_pool_size(plan_env_path, ip_pools_path, template_files):
"pools defined.".format(node_count, role)
)
continue
for pool_name, pool_ips in six.iteritems(pools):
for pool_name, pool_ips in pools.items():
if len(pool_ips) < node_count:
warnings.append(
"Insufficient number of IPs in '{}' pool for '{}' role: "

12
library/switch_vlans.py
View File

@ -20,8 +20,6 @@ import collections.abc as collectionsAbc
import os.path
import six
from ansible.module_utils.basic import AnsibleModule # noqa
from tripleo_validations import utils
from yaml import safe_load as yaml_safe_load
@ -78,7 +76,7 @@ def open_network_environment_files(netenv_path, template_files):
.format(netenv_path, e)])
nic_configs = []
resource_registry = network_data.get('resource_registry', {})
for nic_name, relative_path in six.iteritems(resource_registry):
for nic_name, relative_path in iter(resource_registry.items()):
if nic_name.endswith("Net::SoftwareConfig"):
nic_config_path = os.path.normpath(
os.path.join(os.path.dirname(netenv_path), relative_path))
@ -111,10 +109,10 @@ def validate_switch_vlans(netenv_path, template_files, introspection_data):
# Store VLAN IDs from network-environment.yaml.
vlaninfo = {}
for item, data in six.iteritems(network_data.get('parameter_defaults',
{})):
parameter_defaults = network_data.get('parameter_defaults', {})
for item in parameter_defaults.keys():
if item.endswith('NetworkVlanID'):
vlaninfo[item] = data
vlaninfo[item] = parameter_defaults[item]
# Get the VLANs which are actually used in nic configs
for nic_config_name, nic_config_path, nic_config in nic_configs:
@ -125,7 +123,7 @@ def validate_switch_vlans(netenv_path, template_files, introspection_data):
if not isinstance(resources, collectionsAbc.Mapping):
return [], ["The nic_data must contain the 'resources' key "
"and it must be a dictionary."]
for name, resource in six.iteritems(resources):
for name, resource in iter(resources.items()):
try:
nested_path = [
('properties', collectionsAbc.Mapping, 'dictionary'),

1
requirements.txt
View File

@ -12,7 +12,6 @@ python-ironicclient>=2.7.0 # Apache-2.0
python-ironic-inspector-client>=3.1.1 # Apache-2.0
os-net-config>=7.1.0 # Apache-2.0
oslo.utils>=3.40.2 # Apache-2.0
six>=1.11.0 # MIT
tripleo-common>=7.1.0 # Apache-2.0
setuptools>=50.3.0 # MIT
tripleo-ansible>=6.0.0 # Apache-2.0

22
roles/dhcp_validations/files/rogue_dhcp.py
View File

@ -14,7 +14,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import fcntl
import six
import socket
import struct
import sys
@ -155,10 +154,7 @@ class DHCPDiscover(object):
def _checksum(self, msg):
s = 0
for i in range(0, len(msg), 2):
if six.PY3:
w = msg[i] + (msg[i + 1] << 8)
else:
w = ord(msg[i]) + (ord(msg[i + 1]) << 8)
w = msg[i] + (msg[i + 1] << 8)
s = s + w
s = (s >> 16) + (s & 0xffff)
s = s + (s >> 16)
@ -178,26 +174,18 @@ def get_hw_addresses(interfaces):
def inspect_frame(data):
eth_type = struct.unpack('!H', data[12:14])[0]
protocol = data[23] if six.PY3 else ord(data[23])
protocol = data[23]
src_port = struct.unpack('!H', data[34:36])[0]
dst_port = struct.unpack('!H', data[36:38])[0]
msg_type = data[42] if six.PY3 else ord(data[42])
msg_type = data[42]
# Make sure we got a DHCP Offer
if eth_type == ETH_P_IP \
and protocol == socket.IPPROTO_UDP \
and src_port == 67 \
and dst_port == 68 \
and msg_type == 2: # DHCP Boot Reply
if six.PY3:
server_ip_address = '.'.join(["%s" % m for m in
data[26:30]])
server_hw_address = ":".join(["%02x" % m for m in
data[6:12]])
else:
server_ip_address = '.'.join(["%s" % ord(m) for m in
data[26:30]])
server_hw_address = ":".join(["%02x" % ord(m) for m in
data[6:12]])
server_ip_address = '.'.join(["%s" % m for m in data[26:30]])
server_hw_address = ":".join(["%02x" % m for m in data[6:12]])
dhcp_servers.append([server_ip_address, server_hw_address])

3
tripleo_validations/utils.py
View File

@ -14,7 +14,6 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from six import string_types
import collections.abc as collectionsAbc
from glanceclient import client as glance_client
@ -83,7 +82,7 @@ def get_ironic_client(auth_variables):
def filtered(obj):
"""Only return properties of obj whose value can be properly serialized."""
return {k: v for k, v in obj.__dict__.items()
if isinstance(v, (string_types, int, list, dict, type(None)))}
if isinstance(v, (str, int, list, dict, type(None)))}
def get_nested(data, name, path):