tests: modernize subprocess calls

tests/nvme_copy_test.py

@@ -16,13 +16,8 @@
 
 """
 
-import logging
-import subprocess
-
 from nvme_test import TestNVMe
 
-logger = logging.getLogger(__name__)
-
 
 class TestNVMeCopy(TestNVMe):
 
@@ -44,14 +39,10 @@ def setUp(self):
             # get host behavior support data
             get_features_cmd = f"{self.nvme_bin} get-feature {self.ctrl} " + \
                 "--feature-id=0x16 --data-len=512 --raw-binary"
-            logger.debug(get_features_cmd)
-            proc = subprocess.Popen(get_features_cmd,
-                                    shell=True,
-                                    stdout=subprocess.PIPE,
-                                    encoding='utf-8')
-            err = proc.wait()
+            result = self.run_cmd(get_features_cmd)
+            err = result.returncode
             self.assertEqual(err, 0, "ERROR : nvme get-feature failed")
-            self.host_behavior_data = proc.stdout.read()
+            self.host_behavior_data = result.stdout
             # enable cross-namespace copy formats
             if self.host_behavior_data[4] & cross_namespace_copy:
                 # skip if already enabled
@@ -61,23 +52,13 @@ def setUp(self):
                 data = self.host_behavior_data[:4] + cross_namespace_copy.to_bytes(2, 'little') + self.host_behavior_data[6:]
                 set_features_cmd = f"{self.nvme_bin} set-feature " + \
                     f"{self.ctrl} --feature-id=0x16 --data-len=512"
-                proc = subprocess.Popen(set_features_cmd,
-                                        shell=True,
-                                        stdout=subprocess.PIPE,
-                                        stdin=subprocess.PIPE,
-                                        encoding='utf-8')
-                proc.communicate(input=data)
-                self.assertEqual(proc.returncode, 0, "Failed to enable cross-namespace copy formats")
+                result = self.run_cmd(set_features_cmd, stdin_data=data)
+                self.assertEqual(result.returncode, 0, "Failed to enable cross-namespace copy formats")
         get_ns_id_cmd = f"{self.nvme_bin} get-ns-id {self.ns1}"
-        logger.debug(get_ns_id_cmd)
-        proc = subprocess.Popen(get_ns_id_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        err = proc.wait()
+        result = self.run_cmd(get_ns_id_cmd)
+        err = result.returncode
         self.assertEqual(err, 0, "ERROR : nvme get-ns-id failed")
-        output = proc.stdout.read()
-        logger.debug(output)
+        output = result.stdout
         self.ns1_nsid = int(output.strip().split(':')[-1])
         self.setup_log_dir(self.__class__.__name__)
 
@@ -87,13 +68,7 @@ def tearDown(self):
             # restore saved host behavior support data
             set_features_cmd = f"{self.nvme_bin} set-feature {self.ctrl} " + \
                 "--feature-id=0x16 --data-len=512"
-            logger.debug(set_features_cmd)
-            proc = subprocess.Popen(set_features_cmd,
-                                    shell=True,
-                                    stdout=subprocess.PIPE,
-                                    stdin=subprocess.PIPE,
-                                    encoding='utf-8')
-            proc.communicate(input=self.host_behavior_data)
+            self.run_cmd(set_features_cmd, stdin_data=self.host_behavior_data)
         super().tearDown()
 
     def copy(self, sdlba, blocks, slbs, **kwargs):

tests/nvme_format_test.py

@@ -40,7 +40,6 @@
 import json
 import logging
 import math
-import subprocess
 
 from nvme_test import TestNVMe
 
@@ -110,16 +109,9 @@ def attach_detach_primary_ns(self):
         # read lbaf information
         id_ns_cmd = f"{self.nvme_bin} id-ns {self.ctrl} " + \
             f"--namespace-id={self.default_nsid} --output-format=json"
-        logger.debug(id_ns_cmd)
-        proc = subprocess.Popen(id_ns_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        err = proc.wait()
-        self.assertEqual(err, 0, "ERROR : nvme id-ns failed")
-        output = proc.stdout.read()
-        logger.debug(output)
-        json_output = json.loads(output)
+        result = self.run_cmd(id_ns_cmd)
+        self.assertEqual(result.returncode, 0, "ERROR : nvme id-ns failed")
+        json_output = json.loads(result.stdout)
         self.lba_format_list = json_output['lbafs']
         self.assertTrue(len(self.lba_format_list) > 0,
                         "ERROR : nvme id-ns could not find any lba formats")

tests/nvme_get_features_test.py

@@ -34,13 +34,8 @@
     9. 0Bh M Asynchronous Event Configuration.
 """
 
-import logging
-import subprocess
-
 from nvme_test import TestNVMe
 
-logger = logging.getLogger(__name__)
-
 
 class TestNVMeGetMandatoryFeatures(TestNVMe):
 
@@ -62,14 +57,8 @@ def setUp(self):
         device = self.ctrl.split('/')[-1]
         get_vector_list_cmd = "grep " + device + "q /proc/interrupts |" \
                               " cut -d : -f 1 | tr -d ' ' | tr '\n' ' '"
-        logger.debug(get_vector_list_cmd)
-        proc = subprocess.Popen(get_vector_list_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        output = proc.stdout.read()
-        logger.debug(output)
-        self.vector_list_len = len(output.strip().split(" "))
+        result = self.run_cmd(get_vector_list_cmd)
+        self.vector_list_len = len(result.stdout.strip().split(" "))
 
     def tearDown(self):
         """ Post Section for TestNVMeGetMandatoryFeatures