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,9 +16,6 @@
 
 """
 
-import logging
-import subprocess
-
 from nvme_test import TestNVMe
 
 logger = logging.getLogger(__name__)
@@ -44,14 +41,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 +54,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 +70,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_fw_log_test.py

@@ -56,7 +56,7 @@ def get_fw_log(self):
                 - 0 on success, error code on failure.
         """
         fw_log_cmd = f"{self.nvme_bin} fw-log {self.ctrl}"
-        return self.exec_cmd(fw_log_cmd)
+        return self.run_cmd(fw_log_cmd).returncode
 
     def test_fw_log(self):
         """ Testcase main """

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
@@ -90,13 +79,13 @@ def get_mandatory_features(self, feature_id):
                 get_feat_cmd = f"{self.nvme_bin} get-feature {self.ctrl} " + \
                     f"--feature-id={str(feature_id)} " + \
                     f"--cdw11={str(vector)} --human-readable"
-                self.assertEqual(self.exec_cmd(get_feat_cmd), 0)
+                self.assertEqual(self.run_cmd(get_feat_cmd).returncode, 0)
         else:
             get_feat_cmd = f"{self.nvme_bin} get-feature {self.ctrl} " + \
                 f"--feature-id={str(feature_id)} --human-readable"
             if str(feature_id) == "0x05":
                 get_feat_cmd += f" --namespace-id={self.default_nsid}"
-            self.assertEqual(self.exec_cmd(get_feat_cmd), 0)
+            self.assertEqual(self.run_cmd(get_feat_cmd).returncode, 0)
 
     def test_get_mandatory_features(self):
         """ Testcase main """

tests/nvme_get_lba_status_test.py

@@ -54,7 +54,7 @@ def get_lba_status(self):
             f"--max-dw={str(self.max_dw)} " + \
             f"--action={str(self.action)} " + \
             f"--range-len={str(self.range_len)}"
-        return self.exec_cmd(get_lba_status_cmd)
+        return self.run_cmd(get_lba_status_cmd).returncode
 
     def test_get_lba_status(self):
         """ Testcase main """

tests/nvme_id_ns_test.py

@@ -60,7 +60,7 @@ def get_id_ns(self, nsid):
         """
         id_ns_cmd = f"{self.nvme_bin} id-ns {self.ctrl} " + \
             f"--namespace-id={str(nsid)}"
-        return self.exec_cmd(id_ns_cmd)
+        return self.run_cmd(id_ns_cmd).returncode
 
     def get_id_ns_all(self):
         """

tests/nvme_lba_status_log_test.py

@@ -44,7 +44,7 @@ def get_lba_stat_log(self):
                 - 0 on success, error code on failure.
         """
         lba_stat_log_cmd = f"{self.nvme_bin} lba-status-log {self.ctrl}"
-        return self.exec_cmd(lba_stat_log_cmd)
+        return self.run_cmd(lba_stat_log_cmd).returncode
 
     def test_lba_stat_log(self):
         """ Testcase main """