tests: modernize subprocess calls

Python recommends to use the subprocess.run interface these days. There are
several different ways how calls are done which also makes the logging sprinling
throught the code base. Switch over to the run API and use one central place for
it.

Signed-off-by: Daniel Wagner <[email protected]>

Author: Copilot

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