From: Dennis Maisenbacher Date: Fri, 8 Nov 2024 07:10:47 +0000 (+0000) Subject: tests: Use json output for parsing X-Git-Url: https://www.infradead.org/git/?a=commitdiff_plain;h=6e6305eab7f13dcacae6e25abfaa7f8a0c9062b2;p=users%2Fsagi%2Fnvme-cli.git tests: Use json output for parsing To make the parsing of nvme cmd output easier use the json output when possible. Refactoring code snippets that needed to be touched when adjusing for json output. Signed-off-by: Dennis Maisenbacher --- diff --git a/tests/nvme_format_test.py b/tests/nvme_format_test.py index 1c889137..9faf3dc8 100644 --- a/tests/nvme_format_test.py +++ b/tests/nvme_format_test.py @@ -37,6 +37,7 @@ Namespace Format testcase :- - Delete Namespace. """ +import json import math import subprocess import time @@ -55,9 +56,7 @@ class TestNVMeFormatCmd(TestNVMe): - nsze : namespace size. - ncap : namespace capacity. - ctrl_id : controller id. - - lba_format_list : lis of supported format. - - ms_list : list of metadat size per format. - - lbads_list : list of LBA data size per format. + - lba_format_list : json list of supported format. - test_log_dir : directory for logs, temp files. """ @@ -74,8 +73,6 @@ class TestNVMeFormatCmd(TestNVMe): self.nsze = ncap self.ctrl_id = self.get_ctrl_id() self.lba_format_list = [] - self.ms_list = [] - self.lbads_list = [] self.test_log_dir = self.log_dir + "/" + self.__class__.__name__ self.setup_log_dir(self.__class__.__name__) self.delete_all_ns() @@ -106,30 +103,21 @@ class TestNVMeFormatCmd(TestNVMe): self.dps), 0) self.assertEqual(self.attach_ns(self.ctrl_id, self.default_nsid), 0) # read lbaf information - id_ns = f"{self.nvme_bin} id-ns {self.ctrl} " + \ - f"--namespace-id={self.default_nsid} " + \ - "| grep ^lbaf | awk '{print $2}' | tr -s \"\\n\" \" \"" - proc = subprocess.Popen(id_ns, shell=True, stdout=subprocess.PIPE, + id_ns_cmd = f"{self.nvme_bin} id-ns {self.ctrl} " + \ + f"--namespace-id={self.default_nsid} --output-format=json" + proc = subprocess.Popen(id_ns_cmd, + shell=True, + stdout=subprocess.PIPE, encoding='utf-8') - self.lba_format_list = proc.stdout.read().strip().split(" ") - if proc.wait() == 0: - # read lbads information - id_ns = f"{self.nvme_bin} id-ns {self.ctrl} " + \ - f"--namespace-id={self.default_nsid} " + \ - "| grep ^lbaf | awk '{print $5}' | cut -f 2 -d ':' | tr -s \"\\n\" \" \"" - proc = subprocess.Popen(id_ns, shell=True, stdout=subprocess.PIPE, - encoding='utf-8') - self.lbads_list = proc.stdout.read().strip().split(" ") - # read metadata information - id_ns = f"{self.nvme_bin} id-ns {self.ctrl} " + \ - f"--namespace-id={self.default_nsid} " + \ - "| grep ^lbaf | awk '{print $4}' | cut -f 2 -d ':' | tr -s \"\\n\" \" \"" - proc = subprocess.Popen(id_ns, shell=True, stdout=subprocess.PIPE, - encoding='utf-8') - self.ms_list = proc.stdout.read().strip().split(" ") - self.assertEqual(self.detach_ns(self.ctrl_id, self.default_nsid), 0) - self.assertEqual(self.delete_and_validate_ns(self.default_nsid), 0) - self.nvme_reset_ctrl() + err = proc.wait() + self.assertEqual(err, 0, "ERROR : nvme id-ns failed") + json_output = json.loads(proc.stdout.read()) + 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") + self.assertEqual(self.detach_ns(self.ctrl_id, self.default_nsid), 0) + self.assertEqual(self.delete_and_validate_ns(self.default_nsid), 0) + self.nvme_reset_ctrl() def test_format_ns(self): """ Testcase main """ @@ -137,20 +125,21 @@ class TestNVMeFormatCmd(TestNVMe): self.attach_detach_primary_ns() # iterate through all supported format - for i in range(0, len(self.lba_format_list)): - print("\nlba format " + str(self.lba_format_list[i]) + - " lbad " + str(self.lbads_list[i]) + - " ms " + str(self.ms_list[i])) - metadata_size = 1 if self.ms_list[i] == '8' else 0 + for flbas, lba_format in enumerate(self.lba_format_list): + ds = lba_format['ds'] + ms = lba_format['ms'] + print(f"\nlba format {str(flbas)}" + f"\nds {str(ds)}" + f"\nms {str(ms)}") + dps = 1 if str(ms) == '8' else 0 err = self.create_and_validate_ns(self.default_nsid, self.nsze, self.ncap, - self.lba_format_list[i], - metadata_size) + flbas, + dps) self.assertEqual(err, 0) self.assertEqual(self.attach_ns(self.ctrl_id, self.default_nsid), 0) - self.run_ns_io(self.default_nsid, self.lbads_list[i]) - time.sleep(5) + self.run_ns_io(self.default_nsid, int(ds)) self.assertEqual(self.detach_ns(self.ctrl_id, self.default_nsid), 0) self.assertEqual(self.delete_and_validate_ns(self.default_nsid), 0) self.nvme_reset_ctrl() diff --git a/tests/nvme_id_ns_test.py b/tests/nvme_id_ns_test.py index 48ad9247..247f7536 100644 --- a/tests/nvme_id_ns_test.py +++ b/tests/nvme_id_ns_test.py @@ -42,7 +42,7 @@ class TestNVMeIdentifyNamespace(TestNVMe): """ Pre Section for TestNVMeIdentifyNamespace. """ super().setUp() self.setup_log_dir(self.__class__.__name__) - self.ns_list = self.get_ns_list() + self.nsid_list = self.get_nsid_list() def tearDown(self): """ @@ -81,7 +81,7 @@ class TestNVMeIdentifyNamespace(TestNVMe): - 0 on success, error code on failure. """ err = 0 - for namespace in self.ns_list: + for namespace in self.nsid_list: err = self.get_id_ns(str(namespace)) return err diff --git a/tests/nvme_smart_log_test.py b/tests/nvme_smart_log_test.py index 196998b8..ebc076c1 100644 --- a/tests/nvme_smart_log_test.py +++ b/tests/nvme_smart_log_test.py @@ -76,14 +76,14 @@ class TestNVMeSmartLogCmd(TestNVMe): - Returns: - 0 on success, error code on failure. """ - ns_list = self.get_ns_list() - for nsid in range(0, len(ns_list)): - self.get_smart_log_ns(ns_list[nsid]) + nsid_list = self.get_nsid_list() + for nsid in nsid_list: + self.get_smart_log_ns(nsid) return 0 def test_smart_log(self): """ Testcase main """ self.assertEqual(self.get_smart_log_ctrl(), 0) - smlp = self.supp_check_id_ctrl("lpa") + smlp = int(self.get_id_ctrl_field_value("lpa"), 16) if smlp & 0x1: self.assertEqual(self.get_smart_log_all_ns(), 0) diff --git a/tests/nvme_test.py b/tests/nvme_test.py index 7c910164..0b2c325d 100644 --- a/tests/nvme_test.py +++ b/tests/nvme_test.py @@ -194,7 +194,7 @@ class TestNVMe(unittest.TestCase): "ERROR : nvme list-ctrl could not find ctrl") return str(json_output['ctrl_list'][0]['ctrl_id']) - def get_ns_list(self): + def get_nsid_list(self): """ Wrapper for extracting the namespace list. - Args: - None @@ -202,14 +202,17 @@ class TestNVMe(unittest.TestCase): - List of the namespaces. """ ns_list = [] - ns_list_cmd = f"{self.nvme_bin} list-ns {self.ctrl}" + ns_list_cmd = f"{self.nvme_bin} list-ns {self.ctrl} " + \ + "--output-format=json" proc = subprocess.Popen(ns_list_cmd, shell=True, stdout=subprocess.PIPE, encoding='utf-8') self.assertEqual(proc.wait(), 0, "ERROR : nvme list namespace failed") - for line in proc.stdout: - ns_list.append(line.split('x')[-1]) + json_output = json.loads(proc.stdout.read()) + + for ns in json_output['nsid_list']: + ns_list.append(ns['nsid']) return ns_list @@ -220,22 +223,16 @@ class TestNVMe(unittest.TestCase): - Returns: - maximum number of namespaces supported. """ - pattern = re.compile("^nn[ ]+: [0-9]", re.IGNORECASE) - max_ns = -1 - max_ns_cmd = f"{self.nvme_bin} id-ctrl {self.ctrl}" + max_ns_cmd = f"{self.nvme_bin} id-ctrl {self.ctrl} " + \ + "--output-format=json" proc = subprocess.Popen(max_ns_cmd, shell=True, stdout=subprocess.PIPE, encoding='utf-8') err = proc.wait() self.assertEqual(err, 0, "ERROR : reading maximum namespace count failed") - - for line in proc.stdout: - if pattern.match(line): - max_ns = line.split(":")[1].strip() - break - print(max_ns) - return int(max_ns) + json_output = json.loads(proc.stdout.read()) + return int(json_output['nn']) def get_lba_status_supported(self): """ Check if 'Get LBA Status' command is supported by the device @@ -280,24 +277,9 @@ class TestNVMe(unittest.TestCase): - Args: - None - Returns: - - maximum number of namespaces supported. + - Total NVM capacity. """ - pattern = re.compile("^tnvmcap[ ]+: [0-9]", re.IGNORECASE) - ncap = -1 - ncap_cmd = f"{self.nvme_bin} id-ctrl {self.ctrl}" - proc = subprocess.Popen(ncap_cmd, - shell=True, - stdout=subprocess.PIPE, - encoding='utf-8') - err = proc.wait() - self.assertEqual(err, 0, "ERROR : reading nvm capacity failed") - - for line in proc.stdout: - if pattern.match(line): - ncap = line.split(":")[1].strip() - break - print(ncap) - return int(ncap) + return int(self.get_id_ctrl_field_value("tnvmcap")) def get_id_ctrl_field_value(self, field): """ Wrapper for extracting id-ctrl field values @@ -346,10 +328,11 @@ class TestNVMe(unittest.TestCase): err = proc.wait() self.assertEqual(err, 0, "ERROR : reading nvm capacity failed") + # Not using json output here because parsing flbas makes this less + # readable as the format index is split into lower and upper bits for line in proc.stdout: if "in use" in line: nvm_format = 2 ** int(line.split(":")[3].split()[0]) - print(nvm_format) return int(nvm_format) def delete_all_ns(self): @@ -362,13 +345,16 @@ class TestNVMe(unittest.TestCase): delete_ns_cmd = f"{self.nvme_bin} delete-ns {self.ctrl} " + \ "--namespace-id=0xFFFFFFFF" self.assertEqual(self.exec_cmd(delete_ns_cmd), 0) - list_ns_cmd = f"{self.nvme_bin} list-ns {self.ctrl} --all | wc -l" + list_ns_cmd = f"{self.nvme_bin} list-ns {self.ctrl} --all " + \ + "--output-format=json" proc = subprocess.Popen(list_ns_cmd, shell=True, stdout=subprocess.PIPE, encoding='utf-8') - output = proc.stdout.read().strip() - self.assertEqual(output, '0', "ERROR : deleting all namespace failed") + self.assertEqual(proc.wait(), 0, "ERROR : nvme list-ns failed") + json_output = json.loads(proc.stdout.read()) + self.assertEqual(len(json_output['nsid_list']), 0, + "ERROR : deleting all namespace failed") def create_ns(self, nsze, ncap, flbas, dps): """ Wrapper for creating a namespace. @@ -518,6 +504,7 @@ class TestNVMe(unittest.TestCase): encoding='utf-8') err = proc.wait() self.assertEqual(err, 0, "ERROR : nvme error log failed") + # This sanity checkes the 'normal' output line = proc.stdout.readline() err_log_entry_count = int(line.split(" ")[5].strip().split(":")[1]) entry_count = 0 @@ -550,26 +537,3 @@ class TestNVMe(unittest.TestCase): encoding='utf-8') run_io_result = run_io.communicate()[1] self.assertEqual(run_io_result, None) - - def supp_check_id_ctrl(self, key): - """ Wrapper for support check. - - Args: - - key : search key. - - Returns: - - value for key requested. - """ - id_ctrl = f"{self.nvme_bin} id-ctrl {self.ctrl}" - print("\n" + id_ctrl) - proc = subprocess.Popen(id_ctrl, - shell=True, - stdout=subprocess.PIPE, - encoding='utf-8') - err = proc.wait() - self.assertEqual(err, 0, "ERROR : nvme Identify controller Data \ - structure failed") - for line in proc.stdout: - if key in line: - key = line.replace(",", "", 1) - print(key) - val = (key.split(':'))[1].strip() - return int(val, 16)