CI: Convert Python2 testing scripts to Python3

test/apps/test_fuzzy_match.py

@@ -1,22 +1,22 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 
 #
 # Copyright (C) 2022 NVIDIA CORPORATION & AFFILIATES. ALL RIGHTS RESERVED.
 #
 # See file LICENSE for terms.
 
 import os
-import commands
 import re
 import argparse
+import subprocess
 import sys
 import logging
 
 
 class Environment(object):
     '''Handles environment variables setup and cleanup'''
     def __init__(self, env_vars):
-        logging.info('Using env vars: %s' % env_vars)
+        logging.info(f'Using env vars: {env_vars}')
         self.env_vars = env_vars;
 
     def __enter__(self):
@@ -46,17 +46,17 @@ def run(self, test_case):
             matches = self.get_fuzzy_matches()
 
             if matches != test_case:
-                raise Exception('Wrong fuzzy list: got: %s, expected: %s' % (matches, test_case))
+                raise Exception(f'Wrong fuzzy list: got: {matches}, expected: {test_case}')
 
-            logging.info('found all expected matches: %s' % test_case)
+            logging.info(f'found all expected matches: {test_case}')
 
     def exec_ucx_info(self):
-        cmd = self.ucx_info + ' -u m -w'
-        logging.info('running cmd: %s' % cmd)
+        cmd = f'{self.ucx_info} -u m -w'
+        logging.info(f'running cmd: {cmd}')
 
-        status, output = commands.getstatusoutput(cmd)
+        status, output = subprocess.getstatusoutput(cmd)
         if status != 0:
-            raise Exception('Received unexpected exit code from ucx_info: ' + str(status))
+            raise Exception(f'Received unexpected exit code from ucx_info: {status}')
 
         logging.info(output)
         return output
@@ -76,12 +76,12 @@ def get_fuzzy_matches(self):
         output_vars = warn_match.group(1).split(';')
         matches = [re.match(r'(\w+)(?: \(maybe: (.*)\?\))?', var.strip()) for var in output_vars]
         if None in matches:
-            raise Exception('Unexpected warning message format: %s' % warn_msg)
+            raise Exception(f'Unexpected warning message format: {warn_msg}')
 
         return {m.group(1) : [x.strip() for x in m.group(2).split(',')] if m.group(2) else [] for m in matches}
 
 def has_ib():
-    status, output = commands.getstatusoutput('ibv_devinfo')
+    status, output = subprocess.getstatusoutput('ibv_devinfo')
     if status != 0:
         return False
 
@@ -110,4 +110,3 @@ def has_ib():
     except Exception as e:
         logging.error(str(e))
         sys.exit(1)
-

test/apps/test_ucx_tls.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 #
 # Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2017. ALL RIGHTS RESERVED.
 #
@@ -9,11 +9,10 @@
 import subprocess
 import os
 import re
-import commands
 import itertools
 import contextlib
-from distutils.version import LooseVersion
 from optparse import OptionParser
+from pkg_resources import parse_version
 
 
 #expected AM transport selections per given number of eps
@@ -115,12 +114,12 @@ def _override_env(var_name, value):
 
 def exec_cmd(cmd):
     if options.verbose:
-        print cmd
+        print(cmd)
 
-    status, output = commands.getstatusoutput(cmd)
+    status, output = subprocess.getstatusoutput(cmd)
     if options.verbose:
-        print "return code " + str(status)
-        print output
+        print(f"return code {status}")
+        print(output)
 
     return status, output
 
@@ -131,7 +130,7 @@ def find_am_transport(dev, neps=1, override=0, tls="ib"):
     with _override_env("UCX_TLS", tls), \
          _override_env("UCX_NET_DEVICES", dev):
 
-        status, output = exec_cmd(ucx_info + ucx_info_args + str(neps) + " | grep am")
+        status, output = exec_cmd(f"{ucx_info}{ucx_info_args}{neps} | grep am")
 
     match = re.search(r'\d+:(\S+)/\S+', output)
     if match:
@@ -148,55 +147,55 @@ def test_fallback_from_rc(dev, neps) :
     os.putenv("UCX_TLS", "ib")
     os.putenv("UCX_NET_DEVICES", dev)
 
-    status,output = exec_cmd(ucx_info + ucx_info_args + str(neps) + " | grep rc")
+    status, output = exec_cmd(f"{ucx_info}{ucx_info_args}{neps} | grep rc")
 
     os.unsetenv("UCX_TLS")
     os.unsetenv("UCX_NET_DEVICES")
 
     if output != "":
-        print "RC transport must not be used when estimated number of EPs = " + str(neps)
+        print(f"RC transport must not be used when estimated number of EPs = {neps}")
         sys.exit(1)
 
     os.putenv("UCX_TLS", "rc,ud,tcp")
 
-    status,output_rc = exec_cmd(ucx_info + ucx_info_args + str(neps) + " | grep rc")
+    status, output_rc = exec_cmd(f"{ucx_info}{ucx_info_args}{neps} | grep rc")
 
-    status,output_tcp = exec_cmd(ucx_info + ucx_info_args + str(neps) + " | grep tcp")
+    status, output_tcp = exec_cmd(f"{ucx_info}{ucx_info_args}{neps} | grep tcp")
 
     if output_rc != "" or output_tcp != "":
-        print "RC/TCP transports must not be used when estimated number of EPs = " + str(neps)
+        print(f"RC/TCP transports must not be used when estimated number of EPs = {neps}")
         sys.exit(1)
 
     os.unsetenv("UCX_TLS")
 
 def test_ucx_tls_positive(tls):
     # Use TLS list in "allow" mode and verify that the found tl is in the list
     found_tl = find_am_transport(None, tls=tls)
-    print "Using UCX_TLS=" + tls + ", found TL: " + str(found_tl)
+    print(f"Using UCX_TLS={tls}, found TL: {found_tl}")
     if tls == 'all':
         return
     if not found_tl:
         sys.exit(1)
     tls = tls.split(',')
-    if found_tl in tls or "\\" + found_tl in tls:
+    if found_tl in tls or f"\\{found_tl}" in tls:
         return
     for tl in tls:
         if tl in tl_aliases and found_tl in tl_aliases[tl]:
             return
-    print "Found TL doesn't belong to the allowed UCX_TLS"
+    print("Found TL doesn't belong to the allowed UCX_TLS")
     sys.exit(1)
 
 def test_ucx_tls_negative(tls):
     # Use TLS list in "negate" mode and verify that the found tl is not in the list
     found_tl = find_am_transport(None, tls="^"+tls)
-    print "Using UCX_TLS=^" + tls + ", found TL: " + str(found_tl)
+    print(f"Using UCX_TLS={tls}, found TL: {found_tl}")
     tls = tls.split(',')
     if not found_tl or found_tl in tls:
-        print "No available TL found"
+        print("No available TL found")
         sys.exit(1)
     for tl in tls:
         if tl in tl_aliases and found_tl in tl_aliases[tl]:
-            print "Found TL belongs to the forbidden UCX_TLS"
+            print("Found TL belongs to the forbidden UCX_TLS")
             sys.exit(1)
 
 def _powerset(iterable, with_empty_set=True):
@@ -238,7 +237,7 @@ def test_tls_allow_list(ucx_info):
     bin_prefix = options.prefix + "/bin"
 
 if not (os.path.isdir(bin_prefix)):
-    print "directory \"" + bin_prefix + "\" does not exist"
+    print(f"directory \"{bin_prefix}\" does not exist")
     parser.print_help()
     exit(1)
 
@@ -261,46 +260,45 @@ def test_tls_allow_list(ucx_info):
     if dev_attrs.find("PORT_ACTIVE") == -1:
         continue
 
-    if not os.path.exists("/sys/class/infiniband/%s/ports/%s/gids/0" % (dev, port)):
-        print "Skipping dummy device: ", dev
+    if not os.path.exists(f"/sys/class/infiniband/{dev}/ports/{port}/gids/0"):
+        print("Skipping dummy device: ", dev)
         continue
 
-    driver_name = os.path.basename(os.readlink("/sys/class/infiniband/%s/device/driver" % dev))
+    driver_name = os.path.basename(os.readlink(f"/sys/class/infiniband/{dev}/device/driver"))
     dev_name    = driver_name.split("_")[0] # should be mlx4 or mlx5
     if not dev_name in ['mlx4', 'mlx5']:
-        print "Skipping unknown device: ", dev_name
+        print("Skipping unknown device: ", dev_name)
         continue
 
     if dev_attrs.find("Ethernet") == -1:
         dev_tl_map = am_tls[dev_name]
         dev_tl_override_map = am_tls[dev_name + "_override"]
         override = 1
     else:
-        fw_ver = open("/sys/class/infiniband/%s/fw_ver" % dev).read()
-        if LooseVersion(fw_ver) >= LooseVersion("16.23.0"):
+        fw_ver = open(f"/sys/class/infiniband/{dev}/fw_ver").read()
+        if parse_version(fw_ver) >= parse_version("16.23.0"):
             dev_tl_map = am_tls[dev_name+"_roce_dc"]
         else:
             dev_tl_map = am_tls[dev_name+"_roce_no_dc"]
         override = 0
 
     for n_eps in sorted(dev_tl_map):
         tl = find_am_transport(dev + ':' + port, n_eps)
-        print dev+':' + port + "               eps: ", n_eps, " expected am tl: " + \
-              dev_tl_map[n_eps] + " selected: " + str(tl)
+        print(f"{dev}:{port} eps: {n_eps} expected am tl: {dev_tl_map[n_eps]} selected: {tl}")
 
         if dev_tl_map[n_eps] != tl:
             sys.exit(1)
 
         if override:
-            tl = find_am_transport(dev + ':' + port, n_eps, 1)
-            print dev+':' + port + " UCX_NUM_EPS=2 eps: ", n_eps, " expected am tl: " + \
-                  dev_tl_override_map[n_eps] + " selected: " + str(tl)
+            tl = find_am_transport(f"{dev}:{port}", n_eps, 1)
+            print(f"{dev}:{port} UCX_NUM_EPS=2 eps: {n_eps} expected am tl: \
+                  {dev_tl_override_map[n_eps]} selected: {tl}")
 
             if dev_tl_override_map[n_eps] != tl:
                 sys.exit(1)
 
         if n_eps >= (rc_max_num_eps * 2):
-            test_fallback_from_rc(dev + ':' + port, n_eps)
+            test_fallback_from_rc(f"{dev}:{port}", n_eps)
 
 # Test UCX_TLS configuration (TL choice according to "allow" and "negate" lists)
 test_tls_allow_list(ucx_info)