Harden Python NVMe tests to fail gracefully on malformed JSON

meson_options.txt

@@ -77,6 +77,12 @@ option(
   value : false,
   description: 'Run tests against real hardware'
 )
+option(
+  'nvme-bin',
+  type : 'string',
+  value : '',
+  description: 'Path to nvme binary used for end-to-end tests (default: use the binary from the build tree)'
+)
 option(
   'tests',
   type : 'boolean',

tests/README

@@ -49,3 +49,21 @@ Running testcases with framework
 
     2. Running all the testcases (in the build root directory) :-
        $ meson test -C .build
+
+Selecting the nvme binary
+--------------------------
+    By default the tests use the nvme binary built in the current build tree.
+    To test against a different version (e.g. a distribution-provided binary)
+    two mechanisms are available:
+
+    a. Environment variable (highest priority, overrides config.json):
+       $ NVME_BIN=/usr/bin/nvme python3 tests/tap_runner.py --start-dir tests nvme_id_ctrl_test
+
+    b. Meson option (for meson-driven test runs):
+       $ meson setup .build -Dnvme-bin=/usr/bin/nvme
+       $ meson test -C .build
+
+    c. config.json entry (lower priority than NVME_BIN):
+       Add "nvme_bin": "/usr/bin/nvme" to tests/config.json.
+
+    Priority order: NVME_BIN env var > config.json nvme_bin > build-tree binary

tests/meson.build

@@ -41,6 +41,18 @@ python_module = import('python')
 
 python = python_module.find_installation('python3')
 
+# Determine which nvme binary to expose to the end-to-end tests.
+# The -Dnvme-bin option lets callers point at a distribution-provided (or
+# any other) binary without touching config.json:
+#   meson setup .build -Dnvme-bin=/usr/bin/nvme
+# When the option is not set the binary produced by the current build is used.
+nvme_bin_opt = get_option('nvme-bin')
+if nvme_bin_opt != ''
+    nvme_bin_for_test = nvme_bin_opt
+else
+    nvme_bin_for_test = meson.project_build_root() / 'nvme'
+endif
+
 foreach t : tests
     t_name = t.split('.')[0]
     test(
@@ -51,7 +63,10 @@ foreach t : tests
             '--start-dir', meson.current_source_dir(),
             t_name,
         ],
-        env: ['PATH=' + meson.project_build_root() + ':/usr/bin:/usr/sbin'],
+        env: [
+            'PATH=' + meson.project_build_root() + ':/usr/bin:/usr/sbin',
+            'NVME_BIN=' + nvme_bin_for_test,
+        ],
         timeout: 500,
         protocol: 'tap',
         is_parallel: false,

tests/nvme_copy_test.py

@@ -22,8 +22,6 @@
 
 """
 
-import json
-
 from nvme_test import TestNVMe, to_decimal
 
 
@@ -101,15 +99,21 @@ def _get_current_ns_pif(self):
         result = self.run_cmd(id_ns_cmd)
         if result.returncode != 0:
             return 0
-        flbas = int(json.loads(result.stdout).get("flbas", 0))
+        id_ns_data = self.parse_json_output(result.stdout, "nvme id-ns")
+        flbas = int(id_ns_data.get("flbas", 0))
         lbaf_idx = (flbas & 0xF) | (((flbas >> 5) & 0x3) << 4)
 
         nvm_id_ns_cmd = f"{self.nvme_bin} nvm-id-ns {self.ns1} --output-format=json"
         result = self.run_cmd(nvm_id_ns_cmd)
         if result.returncode != 0:
             return 0
-        elbafs = json.loads(result.stdout).get("elbafs", [])
+        nvm_id_ns_data = self.parse_json_output(result.stdout, "nvme nvm-id-ns")
+        elbafs = nvm_id_ns_data.get("elbafs", [])
+        self.assertIsInstance(elbafs, list,
+                              f"ERROR : nvm-id-ns returned invalid elbafs type: {type(elbafs).__name__}")
         if lbaf_idx < len(elbafs):
+            self.assertIsInstance(elbafs[lbaf_idx], dict,
+                                  f"ERROR : invalid elbaf entry: {elbafs[lbaf_idx]!r}")
             return elbafs[lbaf_idx].get("pif", 0)
         return 0
 
@@ -144,8 +148,13 @@ def _find_64b_guard_lbaf_index(self):
         result = self.run_cmd(nvm_id_ns_cmd)
         if result.returncode != 0:
             return None
-        elbafs = json.loads(result.stdout).get("elbafs", [])
+        nvm_id_ns_data = self.parse_json_output(result.stdout, "nvme nvm-id-ns")
+        elbafs = nvm_id_ns_data.get("elbafs", [])
+        self.assertIsInstance(elbafs, list,
+                              f"ERROR : nvm-id-ns returned invalid elbafs type: {type(elbafs).__name__}")
         for i, elbaf in enumerate(elbafs):
+            self.assertIsInstance(elbaf, dict,
+                                  f"ERROR : invalid elbaf entry: {elbaf!r}")
             if elbaf.get("pif", 0) == 2:  # NVME_NVM_PIF_64B_GUARD = 2
                 return i
         return None
@@ -215,7 +224,7 @@ def _enable_cdfe_for_format(self, desc_format):
         result = self.run_cmd(get_features_cmd)
         self.assertEqual(result.returncode, 0,
                          "ERROR : nvme feat host-behavior-support failed")
-        data = json.loads(result.stdout)
+        data = self.parse_json_output(result.stdout, "nvme feat host-behavior-support")
         fields = data.get("Feature: 0x16", [{}])[0]
         current_cdfe = (
             (0x4 if fields.get("Copy Descriptor Format 2h Enable (CDF2E)") == "True" else 0) |

tests/nvme_format_test.py

@@ -37,7 +37,6 @@
            - Delete Namespace.
 """
 
-import json
 import logging
 import math
 
@@ -111,8 +110,10 @@ def attach_detach_primary_ns(self):
             f"--namespace-id={self.default_nsid} --output-format=json"
         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']
+        json_output = self.parse_json_output(result.stdout, "nvme id-ns")
+        self.lba_format_list = json_output.get('lbafs', [])
+        self.assertIsInstance(self.lba_format_list, list,
+                              f"ERROR : lbafs must be a list, got {type(self.lba_format_list).__name__}")
         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)
@@ -127,6 +128,12 @@ def test_format_ns(self):
         print("##### Testing lba formats:")
         # iterate through all supported format
         for flbas, lba_format in enumerate(self.lba_format_list):
+            self.assertIsInstance(lba_format, dict,
+                                  f"ERROR : lba format entry must be dict, got {type(lba_format).__name__}: {lba_format!r}")
+            self.assertIn('ds', lba_format,
+                          f"ERROR : lba format entry missing ds: {lba_format!r}")
+            self.assertIn('ms', lba_format,
+                          f"ERROR : lba format entry missing ms: {lba_format!r}")
             ds = lba_format['ds']
             ms = lba_format['ms']
             print(f"\nlba format {str(flbas)}"