tests: replace open-coded Popen+wait patterns with self.exec_cmd()

Agent-Logs-Url: https://github.com/igaw/nvme-cli/sessions/c94edfc8-a4da-40d2-a47c-892dd4a6e765

Co-authored-by: igaw <1050803+igaw@users.noreply.github.com>

Author: Copilot

tests/nvme_fw_log_test.py

@@ -26,8 +26,6 @@
     1. Execute fw-log on a device.
 """
 
-import subprocess
-
 from nvme_test import TestNVMe
 
 
@@ -58,11 +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}"
-        proc = subprocess.Popen(fw_log_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        return proc.wait()
+        return self.exec_cmd(fw_log_cmd)
 
     def test_fw_log(self):
         """ Testcase main """

tests/nvme_get_features_test.py

@@ -84,21 +84,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"
-                proc = subprocess.Popen(get_feat_cmd,
-                                        shell=True,
-                                        stdout=subprocess.PIPE,
-                                        encoding='utf-8')
-                self.assertEqual(proc.wait(), 0)
+                self.assertEqual(self.exec_cmd(get_feat_cmd), 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}"
-            proc = subprocess.Popen(get_feat_cmd,
-                                    shell=True,
-                                    stdout=subprocess.PIPE,
-                                    encoding='utf-8')
-            self.assertEqual(proc.wait(), 0)
+            self.assertEqual(self.exec_cmd(get_feat_cmd), 0)
 
     def test_get_mandatory_features(self):
         """ Testcase main """

tests/nvme_get_lba_status_test.py

@@ -12,8 +12,6 @@
     1. Execute get-lba-status on a device.
 """
 
-import subprocess
-
 from nvme_test import TestNVMe
 
 
@@ -56,11 +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)}"
-        proc = subprocess.Popen(get_lba_status_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        return proc.wait()
+        return self.exec_cmd(get_lba_status_cmd)
 
     def test_get_lba_status(self):
         """ Testcase main """

tests/nvme_id_ns_test.py

@@ -27,8 +27,6 @@
     2. Execute id-ns on all namespaces
 """
 
-import subprocess
-
 from nvme_test import TestNVMe
 
 
@@ -62,11 +60,7 @@ def get_id_ns(self, nsid):
         """
         id_ns_cmd = f"{self.nvme_bin} id-ns {self.ctrl} " + \
             f"--namespace-id={str(nsid)}"
-        proc = subprocess.Popen(id_ns_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        return proc.wait()
+        return self.exec_cmd(id_ns_cmd)
 
     def get_id_ns_all(self):
         """

tests/nvme_lba_status_log_test.py

@@ -12,8 +12,6 @@
     1. Execute lba-status-log on a device.
 """
 
-import subprocess
-
 from nvme_test import TestNVMe
 
 
@@ -46,11 +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}"
-        proc = subprocess.Popen(lba_stat_log_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        return proc.wait()
+        return self.exec_cmd(lba_stat_log_cmd)
 
     def test_lba_stat_log(self):
         """ Testcase main """

tests/nvme_test.py

@@ -132,7 +132,7 @@ def validate_pci_device(self):
         """
         x1, x2, dev = self.ctrl.split('/')
         cmd = "find /sys/devices -name \\*" + dev + " | grep -i pci"
-        err = self.exec_cmd(cmd)
+        err = subprocess.call(cmd, shell=True, stdout=subprocess.DEVNULL)
         self.assertEqual(err, 0, "ERROR : Only NVMe PCI subsystem is supported")
 
     def load_config(self):
@@ -188,10 +188,17 @@ def nvme_reset_ctrl(self):
                 - None
         """
         nvme_reset_cmd = f"{self.nvme_bin} reset {self.ctrl}"
-        err = self.exec_cmd(nvme_reset_cmd)
+        err = subprocess.call(nvme_reset_cmd,
+                              shell=True,
+                              stdout=subprocess.DEVNULL)
         self.assertEqual(err, 0, "ERROR : nvme reset failed")
         rescan_cmd = "echo 1 > /sys/bus/pci/rescan"
-        self.assertEqual(self.exec_cmd(rescan_cmd), 0, "ERROR : pci rescan failed")
+        proc = subprocess.Popen(rescan_cmd,
+                                shell=True,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE,
+                                encoding='utf-8')
+        self.assertEqual(proc.wait(), 0, "ERROR : pci rescan failed")
 
     def get_ctrl_id(self):
         """ Wrapper for extracting the first controller id.
@@ -409,7 +416,9 @@ def create_and_validate_ns(self, nsid, nsze, ncap, flbas, dps):
                              "ERROR : create namespace failed")
             id_ns_cmd = f"{self.nvme_bin} id-ns {self.ctrl} " + \
                 f"--namespace-id={str(nsid)}"
-            err = self.exec_cmd(id_ns_cmd)
+            err = subprocess.call(id_ns_cmd,
+                                  shell=True,
+                                  stdout=subprocess.DEVNULL)
         return err
 
     def attach_ns(self, ctrl_id, nsid):
@@ -422,7 +431,9 @@ def attach_ns(self, ctrl_id, nsid):
         """
         attach_ns_cmd = f"{self.nvme_bin} attach-ns {self.ctrl} " + \
             f"--namespace-id={str(nsid)} --controllers={ctrl_id} --verbose"
-        err = self.exec_cmd(attach_ns_cmd)
+        err = subprocess.call(attach_ns_cmd,
+                              shell=True,
+                              stdout=subprocess.DEVNULL)
         if err != 0:
             return err
 
@@ -446,7 +457,9 @@ def detach_ns(self, ctrl_id, nsid):
         """
         detach_ns_cmd = f"{self.nvme_bin} detach-ns {self.ctrl} " + \
             f"--namespace-id={str(nsid)} --controllers={ctrl_id} --verbose"
-        return self.exec_cmd(detach_ns_cmd)
+        return subprocess.call(detach_ns_cmd,
+                               shell=True,
+                               stdout=subprocess.DEVNULL)
 
     def delete_and_validate_ns(self, nsid):
         """ Wrapper for deleting and validating that namespace is deleted.
@@ -458,7 +471,9 @@ def delete_and_validate_ns(self, nsid):
         # delete the namespace
         delete_ns_cmd = f"{self.nvme_bin} delete-ns {self.ctrl} " + \
             f"--namespace-id={str(nsid)} --verbose"
-        err = self.exec_cmd(delete_ns_cmd)
+        err = subprocess.call(delete_ns_cmd,
+                              shell=True,
+                              stdout=subprocess.DEVNULL)
         self.assertEqual(err, 0, "ERROR : delete namespace failed")
         return err
 
@@ -471,11 +486,7 @@ def get_smart_log(self, nsid):
         """
         smart_log_cmd = f"{self.nvme_bin} smart-log {self.ctrl} " + \
             f"--namespace-id={str(nsid)}"
-        proc = subprocess.Popen(smart_log_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        err = proc.wait()
+        err = self.exec_cmd(smart_log_cmd)
         self.assertEqual(err, 0, "ERROR : nvme smart log failed")
         return err
 
@@ -491,11 +502,7 @@ def get_id_ctrl(self, vendor=False):
         else:
             id_ctrl_cmd = f"{self.nvme_bin} id-ctrl " +\
                 f"--vendor-specific {self.ctrl}"
-        proc = subprocess.Popen(id_ctrl_cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                encoding='utf-8')
-        err = proc.wait()
+        err = self.exec_cmd(id_ctrl_cmd)
         self.assertEqual(err, 0, "ERROR : nvme id controller failed")
         return err