CP-41972: update ACK to python3 and fixed as reviewer's suggestions

Signed-off-by: alexhimmel <alex.ac@qq.com>

Author: alexhimmel

acktools/__init__.py

@@ -1,13 +1,13 @@
-# /usr/bin/env python
+# /usr/bin/env python3
 
 import subprocess
 
 
 def make_local_call(call):
     """Function wrapper for making a simple call to shell"""
-    process = subprocess.Popen(call, stdout=subprocess.PIPE)    # NOSONAR
+    process = subprocess.Popen(call, stdout=subprocess.PIPE, universal_newlines=True)    # NOSONAR
     stdout, stderr = process.communicate()
     if process.returncode == 0:
-        return str(stdout).strip()
+        return stdout.strip()
     else:
         raise Exception("Error: '%s' '%s'" % (stdout, stderr))

acktools/log.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 # Copyright (c) 2005-2022 Citrix Systems Inc.
 # Copyright (c) 2022-12-01 Cloud Software Group Holdings, Inc.
@@ -48,7 +48,7 @@ def configure_log(name, path, to_stdout=True):
             '%%(asctime)-8s %s: %%(levelname)-8s %%(filename)s:%%(lineno)-10d %%(message)s' % name)
         fileh.setFormatter(formatter)
         log.addHandler(fileh)
-    except IOError, e:
+    except IOError as e:
         print("Error writing to file handler. Ignoring.")
         print(str(e))
 
@@ -57,7 +57,7 @@ def configure_log(name, path, to_stdout=True):
             sth = logging.StreamHandler(sys.__stdout__)
             sth.setLevel(logging.DEBUG)
             log.addHandler(sth)
-        except IOError, e:
+        except IOError as e:
             print("Error writing to stdout handler. Ignoring.")
             print(str(e))
 

acktools/net/route.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 # Copyright (c) 2005-2022 Citrix Systems Inc.
 # Copyright (c) 2022-12-01 Cloud Software Group Holdings, Inc.
@@ -31,6 +31,8 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 # SUCH DAMAGE.
 
+import sys
+sys.path.append("/opt/xensource/packages/files/auto-cert-kit/pypackages")
 import acktools
 from acktools import utils
 

acktools/net/tests/route_tests.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 
 import unittest
 import mock

acktools/tests/acktools_tests.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 
 import unittest
 import acktools

acktools/utils.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 # Copyright (c) 2005-2022 Citrix Systems Inc.
 # Copyright (c) 2022-12-01 Cloud Software Group Holdings, Inc.
@@ -44,7 +44,7 @@ def cli_table_to_recs(table_string):
     route_recs = []
     for line in lines:
         vals = line.split()
-        rec = dict(zip(header_line, vals))
+        rec = dict(list(zip(header_line, vals)))
         route_recs.append(rec)
 
     return route_recs

autocertkit/ack_cli.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 # Copyright (c) 2005-2022 Citrix Systems Inc.
 # Copyright (c) 2022-12-01 Cloud Software Group Holdings, Inc.
@@ -38,7 +38,7 @@
 import utils
 import sys
 import os
-import ConfigParser
+import configparser
 
 import testbase
 import inspect
@@ -51,7 +51,7 @@
 import test_report
 import test_runner
 from optparse import OptionParser
-from exceptions import *
+import builtins as exceptions
 
 MIN_VLAN = 0
 MAX_VLAN = 4096
@@ -140,7 +140,7 @@ def parse_cmd_args():
 
     if options.optionstr:
         kvp_rec = kvp_string_to_rec(options.optionstr)
-        for k, v in kvp_rec.iteritems():
+        for k, v in kvp_rec.items():
             config[k] = v
 
     check_files(config)
@@ -155,7 +155,7 @@ def check_files(config):
         if opt in config.keys():
             assert_file_exists(os.path.join(INSTALL_DIR, config[opt]), label)
 
-    for key, value in config['netconf'].iteritems():
+    for key, value in config['netconf'].items():
         if key.startswith('eth'):
             vf_driver_pkg = value['vf_driver_pkg']
             if vf_driver_pkg:
@@ -196,7 +196,7 @@ def parse_netconf_file(filename):
     """
     utils.log.debug("Parse network config file: %s" % filename)
 
-    cp = ConfigParser.ConfigParser()
+    cp = configparser.ConfigParser()
     cp.read(filename)
     rec = {}
     for section in cp.sections():
@@ -282,7 +282,7 @@ def parse_section_static_net(cp, rec, section):
                                     'should be in format of "static_<network_id>_<vlan_id>"')
     net = arr[1]
     vlan = arr[2]
-    if not unicode(net.strip()).isdecimal() or not unicode(vlan.strip()).isdecimal():
+    if not str(net.strip()).isdecimal() or not str(vlan.strip()).isdecimal():
         raise utils.InvalidArgument('static addressing section', section,
                                     'should be valid network and/or vlan to determine')
     rec[section] = parse_static_config(cp, section)
@@ -461,7 +461,7 @@ def pre_flight_checks(session, config):
     # Check that we have at least two network adaptors, on the same network
     recs = config['netconf']
     ifaces = {}
-    for k, v in recs.iteritems():
+    for k, v in recs.items():
         if k.startswith('eth'):
             ifaces[k] = v['network_id']
 
@@ -479,8 +479,8 @@ def pre_flight_checks(session, config):
             len(ifaces.keys()), ifaces.keys()))
 
     # If less than 2 interfaces share the same network, raise an exception
-    for k, v in ifaces.iteritems():
-        if ifaces.values().count(v) < 2:
+    for k, v in ifaces.items():
+        if list(ifaces.values()).count(v) < 2:
             raise Exception("Error: ethernet device %s on network %s is defined in the network config but does not have a matching partner. \
                 Please review the nework configuration and minumum requirements of this kit." % (k, v))
 

autocertkit/models.py

@@ -89,7 +89,7 @@ def create_xml_node(self, dom):
         xml_node = dom.createElement(str(self.get_name()))
 
         # Set the elements XML attributes
-        for k, v in self.attr.iteritems():
+        for k, v in self.attr.items():
             xml_node.setAttribute(str(k), str(v))
 
         # Set the elements value
@@ -175,7 +175,7 @@ def is_done(self):
     def create_xml_node(self, dom):
         """Write this test method out to an xml node"""
         xml_node = dom.createElement('test_method')
-        for k, v in self.attr.iteritems():
+        for k, v in self.attr.items():
             xml_node.setAttribute(str(k), str(v))
 
         for elem in self.elems:
@@ -208,7 +208,7 @@ def update_elem(self, k, v):
 
     def update(self, rec):
         """Update elements in this method object from a provided record"""
-        for k, v in rec.iteritems():
+        for k, v in rec.items():
             self.update_elem(k, v)
 
 
@@ -416,7 +416,7 @@ def get_id(self):
             if self.tag == "OP":
                 xs_id = "XenServer %s" % self.config['product_version']
                 return xs_id
-        except Exception, e:
+        except Exception as e:
             log.error("Exception occurred getting ID: '%s'" % str(e))
         return "Unknown ID"
 
@@ -443,7 +443,7 @@ def get_description(self):
             if self.tag == "OP":
                 build_id = "build %s" % self.config['build_number']
                 return build_id
-        except Exception, e:
+        except Exception as e:
             log.error("Exception occurred getting Description: '%s'" % str(e))
         return "Unknown Device"
 
@@ -557,7 +557,7 @@ def print_report(self, stream):
                 stream.write("%s\n" % method.get_name())
         else:
             stream.write("Capabilities:\n")
-            for k, v in self.get_caps().iteritems():
+            for k, v in self.get_caps().items():
                 if v:
                     reqval = "Supported"
                 else:

autocertkit/network_tests.py

@@ -219,7 +219,7 @@ def run(self):
                 # Capture interface statistcs post test run
                 bytes_transferred = int(iperf_data['transfer'])
                 self.validate_stats(bytes_transferred)
-            except Exception, e:
+            except Exception as e:
                 traceb = traceback.format_exc()
                 log.warning(traceb)
                 fail_data["failed_attempt_%d" % (attempt_count)] = str(e)
@@ -684,7 +684,7 @@ def _set_offload_params(self, session, pif):
         log.debug(self.OFFLOAD_CONFIG)
         device = session.xenapi.PIF.get_device(pif)
         log.debug("Device: %s" % device)
-        for k, v in self.OFFLOAD_CONFIG.iteritems():
+        for k, v in self.OFFLOAD_CONFIG.items():
             set_hw_offload(session, device, k, v)
 
     def _verify_ethtool_offloads(self, session, device):
@@ -693,7 +693,7 @@ def _verify_ethtool_offloads(self, session, device):
 
         hw_offloads = get_hw_offloads(session, device)
         log.debug("verify offloads...%s" % hw_offloads)
-        for k, v in self.OFFLOAD_CONFIG.iteritems():
+        for k, v in self.OFFLOAD_CONFIG.items():
             if k not in hw_offloads:
                 raise Exception("Cannot determine status of %s." % k)
             log.debug("Device: %s (%s offload: %s)" %

autocertkit/ssh.py

@@ -60,7 +60,7 @@ def __init__(self,
             self.trans = None
             try:
                 self.connect(ip, username, password, timeout)
-            except Exception, e:
+            except Exception as e:
                 log.error(traceback.format_exc())
                 desc = str(e)
                 log.error("SSH retry %d exception %s" % (retry, desc))
@@ -110,7 +110,7 @@ def connect(self, ip, username, password, timeout):
             try:
                 self.trans.start_client()
                 goes = 0
-            except Exception, e:
+            except Exception as e:
                 goes = goes - 1
                 if goes > 0:
                     self.log.debug("Retrying SSHSession connection %d" % goes)
@@ -185,7 +185,7 @@ def __init__(self, ip, command, username, password, opt):
             self.client.shutdown(1)
             self.hStdout = self.client.makefile()
             self.hStderr = None if combine_stderr else self.client.makefile_stderr()
-        except Exception, e:
+        except Exception as e:
             self.reply = "SSH command executed failed: %s" % str(e),
             self.toreply = 1
             self.close()

autocertkit/status.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 # Copyright (c) 2005-2022 Citrix Systems Inc.
 # Copyright (c) 2022-12-01 Cloud Software Group Holdings, Inc.
@@ -47,7 +47,7 @@
 
 def get_process_strings():
     ps = subprocess.Popen(
-        ['ps', 'aux'], stdout=subprocess.PIPE).communicate()[0]
+        ['ps', 'aux'], stdout=subprocess.PIPE, universal_newlines=True).communicate()[0]
     process_strings = []
     for line in ps.split('\n'):
         if 'ack_cli.py' in line or 'test_runner.py' in line:

autocertkit/test_generators.py

@@ -141,7 +141,7 @@ def append_xml_config(self, doc, xml_node):
 
         device_config = self.get_device_config()
         if device_config:
-            for k, v in device_config.iteritems():
+            for k, v in device_config.items():
                 device_node.setAttribute(k, v)
 
         # Set the unique device id

autocertkit/test_report.py

@@ -43,7 +43,7 @@ def count_by_result(class_recs, fn):
 
     test_matches = []
 
-    for class_name, (method_list, test_class_caps) in class_recs.iteritems():
+    for class_name, (method_list, test_class_caps) in class_recs.items():
         for method in method_list:
             if fn(method['result']):
                 test_matches.append("%s.%s" %

autocertkit/test_runner.py

@@ -136,13 +136,13 @@ def remove_child_nodes(parent_node):
 
 
 def recurse_add_records_to_node(topnode, record):
-    for k, v in record.iteritems():
+    for k, v in record.items():
         node = dom.createElement(k)
         topnode.appendChild(node)
 
         if type(v) == dict:
             # Set attributes for element
-            for key, value in v.iteritems():
+            for key, value in v.items():
                 log.debug("Value = %s Type=%s" %
                           (str(value), str(type(value))))
                 if type(value) == dict:

autocertkit/testbase.py

@@ -103,7 +103,7 @@ def run(self, debug=False, test_name=None):
                 continue
 
             # This assumes that we do not keep IPs across individual tests
-            for vlan, sm in self.static_managers.iteritems():
+            for vlan, sm in self.static_managers.items():
                 sm.release_all()
 
             # Release Alarm signal to prevent handled signal from previous test
@@ -169,7 +169,7 @@ def run_test(self, test, debug, rec):
             self.copy_field(rec, res, 'reason', False)
             self.copy_field(rec, res, 'warning', False)
 
-        except Exception, e:
+        except Exception as e:
             traceb = traceback.format_exc()
             rec['status'] = 'done'
             rec['result'] = 'fail'
@@ -298,7 +298,7 @@ def generate_static_net_conf_common(self, netid_rec, res):
 
         # Iterate through the network config structure to
         # see if we have any static managers to initialise.
-        for k, v in self.get_netconf().iteritems():
+        for k, v in self.get_netconf().items():
             # We only care about vlans on the physical network ID this test is
             # running on
 
@@ -328,7 +328,7 @@ def generate_static_net_conf(self):
         netconf = self.get_netconf()
         log.debug("Netconf: %s" % netconf)
         netid_rec = {}
-        for iface, rec in netconf.iteritems():
+        for iface, rec in netconf.items():
             if iface.startswith('eth'):
                 log.debug("iface: %s Rec: %s" % (iface, rec))
                 nid = rec['network_id']
@@ -412,7 +412,7 @@ def get_pifs_to_use(self):
         equiv_devs = self.get_equivalent_devices()
         try:
             return filter_pif_devices(self.session, equiv_devs)
-        except Exception, e:
+        except Exception as e:
             log.error(
                 "Caught Exception - may be OK if running in single NIC mode.")
             log.error("Exception Occurred: %s" % str(e))
@@ -497,7 +497,7 @@ def get_bondable_ifaces(self, iface):
         # Construct a list of interface names who have the same physical ID
         # as the provided interface.
 
-        blist = intersection([k for k, v in netconf.iteritems() if k.startswith('eth') and
+        blist = intersection([k for k, v in netconf.items() if k.startswith('eth') and
                               v['network_id'] == phy_id],
                              netconf.keys())
 

autocertkit/tests/frontend-test.py

@@ -1,11 +1,11 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
 import unittest
 import sys
 import os
 import os.path
 import random
-import exceptions
+import builtins as exceptions
 import tempfile
 import shutil