Fix pytest warnings

azurelinuxagent/agent.py

@@ -328,7 +328,7 @@ def parse_args(sys_args):
         if arg == "":
             # Don't parse an empty parameter
             continue
-        m = re.match("^(?:[-/]*)configuration-path:([\w/\.\-_]+)", arg)  # pylint: disable=W1401
+        m = re.match(r"^(?:[-/]*)configuration-path:([\w/\.\-_]+)", arg)
         if not m is None:
             conf_file_path = m.group(1)
             if not os.path.exists(conf_file_path):

azurelinuxagent/common/osutil/default.py

@@ -69,7 +69,7 @@ def crypt(password, salt):
 if needed.
 """
 
-_IPTABLES_VERSION_PATTERN = re.compile("^[^\d\.]*([\d\.]+).*$")  # pylint: disable=W1401
+_IPTABLES_VERSION_PATTERN = re.compile(r"^[^\d\.]*([\d\.]+).*$")
 _IPTABLES_LOCKING_VERSION = FlexibleVersion('1.4.21')
 
 
@@ -117,7 +117,7 @@ def get_firewall_delete_conntrack_drop_command(wait, destination):
                       "--ctstate", "INVALID,NEW", "-j", "DROP"])
 
 
-PACKET_PATTERN = "^\s*(\d+)\s+(\d+)\s+DROP\s+.*{0}[^\d]*$"  # pylint: disable=W1401
+PACKET_PATTERN = r"^\s*(\d+)\s+(\d+)\s+DROP\s+.*{0}[^\d]*$"
 ALL_CPUS_REGEX = re.compile('^cpu .*')
 ALL_MEMS_REGEX = re.compile('^Mem.*')
 
@@ -134,7 +134,7 @@ def get_firewall_delete_conntrack_drop_command(wait, destination):
 IOCTL_SIOCGIFHWADDR = 0x8927
 IFNAMSIZ = 16
 
-IP_COMMAND_OUTPUT = re.compile('^\d+:\s+(\w+):\s+(.*)$')  # pylint: disable=W1401
+IP_COMMAND_OUTPUT = re.compile(r'^\d+:\s+(\w+):\s+(.*)$')
 
 STORAGE_DEVICE_PATH = '/sys/bus/vmbus/devices/'
 GEN2_DEVICE_ID = 'f8b3781a-1e82-4818-a1c3-63d806ec15bb'

azurelinuxagent/common/osutil/freebsd.py

@@ -551,7 +551,7 @@ def device_for_ide_port(self, port_id):
         err, output = shellutil.run_get_output(cmd_search_blkvsc)
         if err == 0:
             output = output.rstrip()
-            cmd_search_dev = "camcontrol devlist | grep {0} | awk -F \( '{{print $2}}'|sed -e 's/.*(//'| sed -e 's/).*//'".format(output)  # pylint: disable=W1401
+            cmd_search_dev = "camcontrol devlist | grep {0} | awk -F \\( '{{print $2}}'|sed -e 's/.*(//'| sed -e 's/).*//'".format(output)
             err, output = shellutil.run_get_output(cmd_search_dev)
             if err == 0:
                 for possible in output.rstrip().split(','):
@@ -562,7 +562,7 @@ def device_for_ide_port(self, port_id):
         err, output = shellutil.run_get_output(cmd_search_storvsc)
         if err == 0:
             output = output.rstrip()
-            cmd_search_dev = "camcontrol devlist | grep {0} | awk -F \( '{{print $2}}'|sed -e 's/.*(//'| sed -e 's/).*//'".format(output)  # pylint: disable=W1401
+            cmd_search_dev = "camcontrol devlist | grep {0} | awk -F \\( '{{print $2}}'|sed -e 's/.*(//'| sed -e 's/).*//'".format(output)
             err, output = shellutil.run_get_output(cmd_search_dev)
             if err == 0:
                 for possible in output.rstrip().split(','):

azurelinuxagent/common/utils/flexible_version.py

@@ -196,7 +196,7 @@ def _compile_pattern(self):
         if self.prerel_tags:
             tags = '|'.join(re.escape(tag) for tag in self.prerel_tags)
             self.prerel_tags_set = dict(zip(self.prerel_tags, range(len(self.prerel_tags))))
-            release_re = '(?:{prerel_sep}(?P<{tn}>{tags})(?P<{nn}>\d*))?'.format(  # pylint: disable=W1401
+            release_re = r'(?:{prerel_sep}(?P<{tn}>{tags})(?P<{nn}>\d*))?'.format(
                         prerel_sep=self._re_prerel_sep, 
                         tags=tags, 
                         tn=self._nn_prerel_tag, 

azurelinuxagent/common/version.py

@@ -62,8 +62,8 @@ def get_f5_platform():
     the version and product information is contained in the /VERSION file.
     """
     result = [None, None, None, None]
-    f5_version = re.compile("^Version: (\d+\.\d+\.\d+)")  # pylint: disable=W1401
-    f5_product = re.compile("^Product: ([\w-]+)")  # pylint: disable=W1401
+    f5_version = re.compile(r"^Version: (\d+\.\d+\.\d+)")
+    f5_product = re.compile(r"^Product: ([\w-]+)")
 
     with open('/VERSION', 'r') as fh:
         content = fh.readlines()
@@ -105,15 +105,15 @@ def get_checkpoint_platform():
 def get_distro():
 
     if 'FreeBSD' in platform.system():
-        release = re.sub('\-.*\Z', '', ustr(platform.release()))  # pylint: disable=W1401
+        release = re.sub(r'\-.*\Z', '', ustr(platform.release()))
         osinfo = ['freebsd', release, '', 'freebsd']
     elif 'OpenBSD' in platform.system():
-        release = re.sub('\-.*\Z', '', ustr(platform.release()))  # pylint: disable=W1401
+        release = re.sub(r'\-.*\Z', '', ustr(platform.release()))
         osinfo = ['openbsd', release, '', 'openbsd']
     elif 'Linux' in platform.system():
         osinfo = get_linux_distribution(0, 'alpine')
     elif 'NS-BSD' in platform.system():
-        release = re.sub('\-.*\Z', '', ustr(platform.release()))  # pylint: disable=W1401
+        release = re.sub(r'\-.*\Z', '', ustr(platform.release()))
         osinfo = ['nsbsd', release, '', 'nsbsd']
     else:
         try:
@@ -222,13 +222,13 @@ def has_logrotate():
 
 AGENT_PATTERN = "{0}-(.*)".format(AGENT_NAME)
 AGENT_NAME_PATTERN = re.compile(AGENT_PATTERN)
-AGENT_PKG_PATTERN = re.compile(AGENT_PATTERN+"\.zip")  # pylint: disable=W1401
+AGENT_PKG_PATTERN = re.compile(AGENT_PATTERN+r"\.zip")
 AGENT_DIR_PATTERN = re.compile(".*/{0}".format(AGENT_PATTERN))
 
 # The execution mode of the VM - IAAS or PAAS. Linux VMs are only executed in IAAS mode.
 AGENT_EXECUTION_MODE = "IAAS"
 
-EXT_HANDLER_PATTERN = b".*/WALinuxAgent-(\d+.\d+.\d+[.\d+]*).*-run-exthandlers"  # pylint: disable=W1401
+EXT_HANDLER_PATTERN = br".*/WALinuxAgent-(\d+.\d+.\d+[.\d+]*).*-run-exthandlers"
 EXT_HANDLER_REGEX = re.compile(EXT_HANDLER_PATTERN)
 
 __distro__ = get_distro()

azurelinuxagent/ga/env.py

@@ -35,9 +35,9 @@
 from azurelinuxagent.ga.periodic_operation import PeriodicOperation
 
 CACHE_PATTERNS = [
-    re.compile("^(.*)\.(\d+)\.(agentsManifest)$", re.IGNORECASE),  # pylint: disable=W1401
-    re.compile("^(.*)\.(\d+)\.(manifest\.xml)$", re.IGNORECASE),  # pylint: disable=W1401
-    re.compile("^(.*)\.(\d+)\.(xml)$", re.IGNORECASE)  # pylint: disable=W1401
+    re.compile(r"^(.*)\.(\d+)\.(agentsManifest)$", re.IGNORECASE),
+    re.compile(r"^(.*)\.(\d+)\.(manifest\.xml)$", re.IGNORECASE),
+    re.compile(r"^(.*)\.(\d+)\.(xml)$", re.IGNORECASE)
 ]
 
 MAXIMUM_CACHED_FILES = 50

azurelinuxagent/ga/update.py

@@ -896,7 +896,7 @@ def _get_pid_parts(self):
         pid_file = conf.get_agent_pid_file_path()
         pid_dir = os.path.dirname(pid_file)
         pid_name = os.path.basename(pid_file)
-        pid_re = re.compile("(\d+)_{0}".format(re.escape(pid_name)))  # pylint: disable=W1401
+        pid_re = re.compile(r"(\d+)_{0}".format(re.escape(pid_name)))
         return pid_dir, pid_name, pid_re
 
     def _get_pid_files(self):

azurelinuxagent/pa/rdma/centos.py

@@ -82,7 +82,7 @@ def is_rdma_package_up_to_date(self, pkg, fw_version):
         # Example match (pkg name, -, followed by 3 segments, fw_version and -):
         # - pkg=microsoft-hyper-v-rdma-4.1.0.142-20160323.x86_64
         # - fw_version=142
-        pattern = '{0}-(\d+\.){{3,}}({1})-'.format(self.rdma_user_mode_package_name, fw_version)  # pylint: disable=W1401
+        pattern = r'{0}-(\d+\.){{3,}}({1})-'.format(self.rdma_user_mode_package_name, fw_version)
         return re.match(pattern, pkg)
 
     @staticmethod
@@ -155,7 +155,7 @@ def install_rdma_drivers(self, fw_version):
 
         # Install kernel mode driver (kmod-microsoft-hyper-v-rdma-*)
         kmod_pkg = self.get_file_by_pattern(
-            pkgs, "%s-(\d+\.){3,}(%s)-\d{8}\.x86_64.rpm" % (self.rdma_kernel_mode_package_name, fw_version))  # pylint: disable=W1401
+            pkgs, r"%s-(\d+\.){3,}(%s)-\d{8}\.x86_64.rpm" % (self.rdma_kernel_mode_package_name, fw_version))
         if not kmod_pkg:
             raise Exception("RDMA kernel mode package not found")
         kmod_pkg_path = os.path.join(pkg_dir, kmod_pkg)
@@ -164,7 +164,7 @@ def install_rdma_drivers(self, fw_version):
 
         # Install user mode driver (microsoft-hyper-v-rdma-*)
         umod_pkg = self.get_file_by_pattern(
-            pkgs, "%s-(\d+\.){3,}(%s)-\d{8}\.x86_64.rpm" % (self.rdma_user_mode_package_name, fw_version))  # pylint: disable=W1401
+            pkgs, r"%s-(\d+\.){3,}(%s)-\d{8}\.x86_64.rpm" % (self.rdma_user_mode_package_name, fw_version))
         if not umod_pkg:
             raise Exception("RDMA user mode package not found")
         umod_pkg_path = os.path.join(pkg_dir, umod_pkg)

azurelinuxagent/pa/rdma/rdma.py

@@ -246,7 +246,7 @@ def provision_network_direct_rdma(self):
             return
         retcode, out = shellutil.run_get_output("modinfo %s" % module_name)
         if retcode == 0:
-            version = re.search("version:\s+(\d+)\.(\d+)\.(\d+)\D", out, re.IGNORECASE)  # pylint: disable=W1401
+            version = re.search(r"version:\s+(\d+)\.(\d+)\.(\d+)\D", out, re.IGNORECASE)
             if version:
                 v1 = int(version.groups(0)[0])
                 v2 = int(version.groups(0)[1])
@@ -473,7 +473,7 @@ def update_dat_conf(paths, ipv4_addr):
 
     @staticmethod
     def replace_dat_conf_contents(cfg, ipv4_addr):
-        old = "ofa-v2-ib0 u2.0 nonthreadsafe default libdaplofa.so.2 dapl.2.0 \"\S+ 0\""  # pylint: disable=W1401
+        old = r"ofa-v2-ib0 u2.0 nonthreadsafe default libdaplofa.so.2 dapl.2.0 \"\S+ 0\""
         new = "ofa-v2-ib0 u2.0 nonthreadsafe default libdaplofa.so.2 dapl.2.0 \"{0} 0\"".format(
             ipv4_addr)
         return re.sub(old, new, cfg)

azurelinuxagent/pa/rdma/ubuntu.py

@@ -37,7 +37,7 @@ def install_driver(self):
             logger.error("RDMA: Could not determine firmware version. No driver will be installed")
             return
         #replace . with _, we are looking for number like 144_0
-        nd_version = re.sub('\.', '_', nd_version)  # pylint: disable=W1401
+        nd_version = re.sub(r'\.', '_', nd_version)
 
         #Check to see if we need to reconfigure driver
         status,module_name = shellutil.run_get_output('modprobe -R hv_network_direct', chk_err=False)
@@ -79,13 +79,13 @@ def install_driver(self):
         status,output = shellutil.run_get_output('apt-cache show --no-all-versions linux-azure')
         if status != 0:
             return
-        r = re.search('Version: (\S+)', output)  # pylint: disable=W1401
+        r = re.search(r'Version: (\S+)', output)
         if not r:
             logger.error("RDMA: version not found in package linux-azure.")
             return
         package_version = r.groups()[0]
         #Remove the ending .<upload number> after <ABI number>
-        package_version = re.sub("\.\d+$", "", package_version)  # pylint: disable=W1401
+        package_version = re.sub(r"\.\d+$", "", package_version)
 
         logger.info('RDMA: kernel_version=%s package_version=%s' % (kernel_version, package_version))
         kernel_version_array = [ int(x) for x in kernel_version.split('.') ]
@@ -111,9 +111,9 @@ def update_modprobed_conf(self, nd_version):
             with open(modprobed_file, 'r') as f:
                 lines = f.read()
 
-        r = re.search('alias hv_network_direct hv_network_direct_\S+', lines)  # pylint: disable=W1401
+        r = re.search(r'alias hv_network_direct hv_network_direct_\S+', lines)
         if r:
-            lines = re.sub('alias hv_network_direct hv_network_direct_\S+', 'alias hv_network_direct hv_network_direct_%s' % nd_version, lines)  # pylint: disable=W1401
+            lines = re.sub(r'alias hv_network_direct hv_network_direct_\S+', 'alias hv_network_direct hv_network_direct_%s' % nd_version, lines)
         else:
             lines += '\nalias hv_network_direct hv_network_direct_%s\n' % nd_version
         with open('/etc/modprobe.d/vmbus-rdma.conf', 'w') as f:

ci/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+filterwarnings =
+    ignore:distro.linux_distribution\(\) is deprecated

ci/pytest.sh

@@ -7,13 +7,13 @@ EXIT_CODE=0
 echo "========================================="
 echo "****   pytest *** non-sudo tests     ****"
 echo "========================================="
-pytest --ignore-glob '*/test_cgroupconfigurator_sudo.py' --verbose tests || EXIT_CODE=$(($EXIT_CODE || $?))
+pytest --verbose --config-file ci/pytest.ini --ignore-glob '*/test_cgroupconfigurator_sudo.py' tests || EXIT_CODE=$(($EXIT_CODE || $?))
 echo EXIT_CODE pytests non-sudo = $EXIT_CODE
 
 echo "========================================="
 echo "****     pytest *** sudo tests       ****"
 echo "========================================="
-sudo env "PATH=$PATH" pytest --verbose tests/ga/test_cgroupconfigurator_sudo.py || EXIT_CODE=$(($EXIT_CODE || $?))
+sudo env "PATH=$PATH" pytest --verbose --config-file ci/pytest.ini tests/ga/test_cgroupconfigurator_sudo.py || EXIT_CODE=$(($EXIT_CODE || $?))
 echo EXIT_CODE pytests sudo = $EXIT_CODE
 
 exit "$EXIT_CODE"

tests/common/test_singletonperthread.py

@@ -6,7 +6,7 @@
 from tests.lib.tools import AgentTestCase, clear_singleton_instances
 
 
-class TestClassToTestSingletonPerThread(SingletonPerThread):
+class Singleton(SingletonPerThread):
     """
     Since these tests deal with testing in a multithreaded environment,
     we employ the use of multiprocessing.Queue() to ensure that the data is consistent.
@@ -47,7 +47,7 @@ def setUp(self):
         # In a multi-threaded environment, exceptions thrown in the child thread will not be propagated to the parent
         # thread. In order to achieve that, adding all exceptions to a Queue and then checking that in parent thread.
         self.errors = Queue()
-        clear_singleton_instances(TestClassToTestSingletonPerThread)
+        clear_singleton_instances(Singleton)
 
     def _setup_multithread_and_execute(self, func1, args1, func2, args2, t1_name=None, t2_name=None):
 
@@ -69,7 +69,7 @@ def _setup_multithread_and_execute(self, func1, args1, func2, args2, t1_name=Non
     @staticmethod
     def _get_test_class_instance(q, err):
         try:
-            obj = TestClassToTestSingletonPerThread()
+            obj = Singleton()
             q.put(obj)
         except Exception as e:
             err.put(str(e))
@@ -91,8 +91,8 @@ def check_obj(name):
         return t1_object, t2_object
 
     def test_it_should_have_only_one_instance_for_same_thread(self):
-        obj1 = TestClassToTestSingletonPerThread()
-        obj2 = TestClassToTestSingletonPerThread()
+        obj1 = Singleton()
+        obj2 = Singleton()
 
         self.assertEqual(obj1.uuid, obj2.uuid)
 
@@ -137,7 +137,7 @@ def test_singleton_object_should_match_thread_name(self):
         t1_name = str(uuid.uuid4())
         t2_name = str(uuid.uuid4())
 
-        test_class_obj_name = lambda t_name: "%s__%s" % (TestClassToTestSingletonPerThread.__name__, t_name)
+        test_class_obj_name = lambda t_name: "%s__%s" % (Singleton.__name__, t_name)
 
         self._setup_multithread_and_execute(func1=self._get_test_class_instance,
                                             args1=(instances, self.errors),
@@ -146,7 +146,7 @@ def test_singleton_object_should_match_thread_name(self):
                                             t1_name=t1_name,
                                             t2_name=t2_name)
 
-        singleton_instances = TestClassToTestSingletonPerThread._instances  # pylint: disable=no-member
+        singleton_instances = Singleton._instances  # pylint: disable=no-member
 
         # Assert instance names are consistent with the thread names
         self.assertIn(test_class_obj_name(t1_name), singleton_instances)

tests/lib/tools.py

@@ -156,8 +156,6 @@ def setUpClass(cls):
             cls.assertIsNone = cls.emulate_assertIsNone
         if not hasattr(cls, "assertIsNotNone"):
             cls.assertIsNotNone = cls.emulate_assertIsNotNone
-        if hasattr(cls, "assertRaisesRegexp"):
-            cls.assertRaisesRegex = cls.assertRaisesRegexp
         if not hasattr(cls, "assertRaisesRegex"):
             cls.assertRaisesRegex = cls.emulate_raises_regex
         if not hasattr(cls, "assertListEqual"):