Remove Pylint warning: consider using with - CP: #2231

azurelinuxagent/agent.py

@@ -371,11 +371,12 @@ def start(conf_file_path=None):
     Start agent daemon in a background process and set stdout/stderr to
     /dev/null
     """
-    devnull = open(os.devnull, 'w')
     args = [sys.argv[0], '-daemon']
     if conf_file_path is not None:
         args.append('-configuration-path:{0}'.format(conf_file_path))
-    subprocess.Popen(args, stdout=devnull, stderr=devnull)
+
+    with open(os.devnull, 'w') as devnull:
+        subprocess.Popen(args, stdout=devnull, stderr=devnull)
 
 
 if __name__ == '__main__' :

azurelinuxagent/common/utils/archive.py

@@ -192,12 +192,16 @@ def archive(self):
         fn_tmp = "{0}.zip.tmp".format(self._path)
         filename = "{0}.zip".format(self._path)
 
-        ziph = zipfile.ZipFile(fn_tmp, 'w')
-        for current_file in os.listdir(self._path):
-            full_path = os.path.join(self._path, current_file)
-            ziph.write(full_path, current_file, zipfile.ZIP_DEFLATED)
-
-        ziph.close()
+        ziph = None
+        try:
+            # contextmanager for zipfile.ZipFile doesn't exist for py2.6, manually closing it
+            ziph = zipfile.ZipFile(fn_tmp, 'w')
+            for current_file in os.listdir(self._path):
+                full_path = os.path.join(self._path, current_file)
+                ziph.write(full_path, current_file, zipfile.ZIP_DEFLATED)
+        finally:
+            if ziph is not None:
+                ziph.close()
 
         os.rename(fn_tmp, filename)
         shutil.rmtree(self._path)

azurelinuxagent/daemon/scvmm.py

@@ -63,9 +63,9 @@ def start_scvmm_agent(self, mount_point=None):
         if mount_point is None:
             mount_point = conf.get_dvd_mount_point()
         startup_script = os.path.join(mount_point, VMM_STARTUP_SCRIPT_NAME)
-        devnull = open(os.devnull, 'w')
-        subprocess.Popen(["/bin/bash", startup_script, "-p " + mount_point],
-                         stdout=devnull, stderr=devnull)
+        with open(os.devnull, 'w') as devnull:
+            subprocess.Popen(["/bin/bash", startup_script, "-p " + mount_point],
+                             stdout=devnull, stderr=devnull)
 
     def run(self):
         if self.detect_scvmm_env():

azurelinuxagent/pa/rdma/ubuntu.py

@@ -108,15 +108,15 @@ def update_modprobed_conf(self, nd_version):
         if not os.path.isfile(modprobed_file):
             logger.info("RDMA: %s not found, it will be created" % modprobed_file)
         else:
-            f = open(modprobed_file, 'r')
-            lines = f.read()
-            f.close()
+            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
         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
         else:
             lines += '\nalias hv_network_direct hv_network_direct_%s\n' % nd_version
-        f = open('/etc/modprobe.d/vmbus-rdma.conf', 'w')
-        f.write(lines)
-        f.close()
+        with open('/etc/modprobe.d/vmbus-rdma.conf', 'w') as f:
+            f.write(lines)
+
         logger.info("RDMA: hv_network_direct alias updated to ND %s" % nd_version)

ci/2.7.pylintrc

@@ -7,6 +7,7 @@ disable=C, # (C) convention, for programming standard violation
     consider-using-dict-comprehension, # (R1717): *Consider using a dictionary comprehension*
     consider-using-in, # (R1714): *Consider merging these comparisons with "in" to %r*
     consider-using-set-comprehension, # (R1718): *Consider using a set comprehension*
+    consider-using-with, # (R1732): *Emitted if a resource-allocating assignment or call may be replaced by a 'with' block*
     duplicate-code, # (R0801): *Similar lines in %s files*
     no-init, # (W0232): Class has no __init__ method
     no-else-break, # (R1723): *Unnecessary "%s" after "break"*

ci/3.6.pylintrc

@@ -6,6 +6,7 @@ disable=C, # (C) convention, for programming standard violation
     consider-using-dict-comprehension, # (R1717): *Consider using a dictionary comprehension*
     consider-using-in, # (R1714): *Consider merging these comparisons with "in" to %r*
     consider-using-set-comprehension, # (R1718): *Consider using a set comprehension*
+    consider-using-with, # (R1732): *Emitted if a resource-allocating assignment or call may be replaced by a 'with' block*
     duplicate-code, # (R0801): *Similar lines in %s files*
     no-init, # (W0232): Class has no __init__ method
     no-else-break, # (R1723): *Unnecessary "%s" after "break"*

tests/ga/test_update.py

@@ -174,13 +174,14 @@ def agent_versions(self):
         v.sort(reverse=True)
         return v
 
+    @contextlib.contextmanager
     def get_error_file(self, error_data=None):
         if error_data is None:
             error_data = NO_ERROR
-        fp = tempfile.NamedTemporaryFile(mode="w")
-        json.dump(error_data if error_data is not None else NO_ERROR, fp)
-        fp.seek(0)
-        return fp
+        with tempfile.NamedTemporaryFile(mode="w") as fp:
+            json.dump(error_data if error_data is not None else NO_ERROR, fp)
+            fp.seek(0)
+            yield fp
 
     def create_error(self, error_data=None):
         if error_data is None:

tests/utils/test_archive.py

@@ -237,9 +237,13 @@ def assert_datetime_close_to(self, time1, time2, within):
             self.fail("the timestamps are outside of the tolerance of by {0} seconds".format(secs))
 
     def assert_zip_contains(self, zip_filename, files):
-        ziph = zipfile.ZipFile(zip_filename, 'r')
-        zip_files = [x.filename for x in ziph.filelist]
-        for current_file in files:
-            self.assertTrue(current_file in zip_files, "'{0}' was not found in {1}".format(current_file, zip_filename))
-
-        ziph.close()
+        ziph = None
+        try:
+            # contextmanager for zipfile.ZipFile doesn't exist for py2.6, manually closing it
+            ziph = zipfile.ZipFile(zip_filename, 'r')
+            zip_files = [x.filename for x in ziph.filelist]
+            for current_file in files:
+                self.assertTrue(current_file in zip_files, "'{0}' was not found in {1}".format(current_file, zip_filename))
+        finally:
+            if ziph is not None:
+                ziph.close()

tests/utils/test_extension_process_util.py

@@ -37,6 +37,8 @@ def setUp(self):
         self.stderr.write("The five boxing wizards jump quickly.".encode("utf-8"))
 
     def tearDown(self):
+        self.stderr.close()
+        self.stdout.close()
         if self.tmp_dir is not None:
             shutil.rmtree(self.tmp_dir)