import unittest import os import re import shutil import overrides_hack from utils import create_sparse_tempfile, create_nvmet_device, delete_nvmet_device, setup_nvme_target, teardown_nvme_target, find_nvme_ctrl_devs_for_subnqn, find_nvme_ns_devs_for_subnqn, get_nvme_hostnqn, run_command, TestTags, tag_test, read_file, write_file, required_plugins import gi gi.require_version('GLib', '2.0') gi.require_version('BlockDev', '3.0') from gi.repository import GLib, BlockDev @required_plugins(("nvme", "loop")) class NVMeTest(unittest.TestCase): requested_plugins = BlockDev.plugin_specs_from_names(("nvme", "loop")) # kernel nvme target needs disk images on a real filesystem (not tmpfs) TMPDIR = "/var/tmp" @classmethod def setUpClass(cls): if not shutil.which("nvme"): raise unittest.SkipTest("nvme executable (nvme-cli package) not found in $PATH, skipping.") if not shutil.which("nvmetcli"): raise unittest.SkipTest("nvmetcli executable not found in $PATH, skipping.") ret, out, err = run_command("modprobe nvme-fabrics") if ret != 0: raise unittest.SkipTest("nvme-fabrics kernel module unavailable, skipping.") if not BlockDev.is_initialized(): BlockDev.init(cls.requested_plugins, None) else: BlockDev.reinit(cls.requested_plugins, True, None) def _safe_unlink(self, f): try: os.unlink(f) except FileNotFoundError: pass class NVMePluginVersionTestCase(NVMeTest): @tag_test(TestTags.NOSTORAGE) def test_plugin_version(self): self.assertEqual(BlockDev.get_plugin_soname(BlockDev.Plugin.NVME), "libbd_nvme.so.3") @tag_test(TestTags.NOSTORAGE) def test_availability(self): avail = BlockDev.nvme_is_tech_avail(BlockDev.NVMETech.NVME, 0) self.assertTrue(avail) avail = BlockDev.nvme_is_tech_avail(BlockDev.NVMETech.FABRICS, 0) self.assertTrue(avail) class NVMeTestCase(NVMeTest): def setUp(self): self.dev_file = None self.nvme_dev = None self.nvme_ns_dev = None self.addCleanup(self._clean_up) self.dev_file = create_sparse_tempfile("nvme_test", 1024**3, dir=self.TMPDIR) self.nvme_dev = create_nvmet_device(self.dev_file) self.nvme_ns_dev = self.nvme_dev + "n1" os.unlink(self.dev_file) def _clean_up(self): if self.nvme_dev: try: delete_nvmet_device(self.nvme_dev) except RuntimeError: # just move on, we can do no better here pass if self.dev_file: self._safe_unlink(self.dev_file) @tag_test(TestTags.CORE) def test_ns_info(self): """Test namespace info""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_get_namespace_info("/dev/nonexistent") with self.assertRaisesRegex(GLib.GError, r"Error getting Namespace Identifier \(NSID\): Inappropriate ioctl for device"): BlockDev.nvme_get_namespace_info(self.nvme_dev) # not much information can be gathered from loop-based NVMe target devices... info = BlockDev.nvme_get_namespace_info(self.nvme_ns_dev) self.assertFalse(info.features & BlockDev.NVMENamespaceFeature.THIN) self.assertTrue (info.features & BlockDev.NVMENamespaceFeature.MULTIPATH_SHARED) self.assertFalse(info.features & BlockDev.NVMENamespaceFeature.FORMAT_PROGRESS) self.assertFalse(info.features & BlockDev.NVMENamespaceFeature.ROTATIONAL) self.assertIsNone(info.eui64) self.assertEqual(info.format_progress_remaining, 0) self.assertEqual(len(info.lba_formats), 1) self.assertGreater(len(info.nguid), 0) self.assertEqual(info.nsid, 1) self.assertEqual(info.ncap, 262144) self.assertEqual(info.nsize, 262144) self.assertEqual(info.nuse, 262144) self.assertGreater(len(info.uuid), 0) self.assertFalse(info.write_protected) self.assertEqual(info.current_lba_format.data_size, 4096) self.assertEqual(info.current_lba_format.metadata_size, 0) self.assertEqual(info.current_lba_format.relative_performance, BlockDev.NVMELBAFormatRelativePerformance.BEST) @tag_test(TestTags.CORE) def test_ctrl_info(self): """Test controller info""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_get_controller_info("/dev/nonexistent") info = BlockDev.nvme_get_controller_info(self.nvme_dev) self.assertEqual(info.ctrl_id, 1) self.assertEqual(info.pci_vendor_id, 0) self.assertEqual(info.pci_subsys_vendor_id, 0) self.assertIsNone(info.fguid) self.assertIn("Linux", info.model_number) self.assertGreater(len(info.serial_number), 0) self.assertGreater(len(info.firmware_ver), 0) self.assertGreater(len(info.nvme_ver), 0) self.assertTrue (info.features & BlockDev.NVMEControllerFeature.MULTIPORT) self.assertTrue (info.features & BlockDev.NVMEControllerFeature.MULTICTRL) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SRIOV) self.assertTrue (info.features & BlockDev.NVMEControllerFeature.ANA_REPORTING) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.FORMAT) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.FORMAT_ALL_NS) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.NS_MGMT) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SELFTEST) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SELFTEST_SINGLE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SANITIZE_CRYPTO) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SANITIZE_BLOCK) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SANITIZE_OVERWRITE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SECURE_ERASE_ALL_NS) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SECURE_ERASE_CRYPTO) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.STORAGE_DEVICE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.ENCLOSURE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.MGMT_PCIE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.MGMT_SMBUS) self.assertEqual(info.controller_type, BlockDev.NVMEControllerType.IO) self.assertEqual(info.selftest_ext_time, 0) self.assertEqual(info.hmb_pref_size, 0) self.assertEqual(info.hmb_min_size, 0) self.assertEqual(info.size_total, 0) self.assertEqual(info.size_unalloc, 0) self.assertEqual(info.num_namespaces, 1024) self.assertEqual(info.subsysnqn, "libblockdev_subnqn") @tag_test(TestTags.CORE) def test_smart_log(self): """Test SMART health log""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_get_smart_log("/dev/nonexistent") log = BlockDev.nvme_get_smart_log(self.nvme_dev) self.assertEqual(log.avail_spare, 0) self.assertEqual(log.cctemp, 0) self.assertEqual(log.critical_temp_time, 0) self.assertEqual(log.ctrl_busy_time, 0) self.assertEqual(log.media_errors, 0) self.assertEqual(log.percent_used, 0) self.assertEqual(log.power_cycles, 0) self.assertEqual(log.power_on_hours, 0) self.assertEqual(log.spare_thresh, 0) self.assertEqual(log.temp_sensors, [0, 0, 0, 0, 0, 0, 0, 0]) self.assertEqual(log.temperature, 0) self.assertEqual(log.unsafe_shutdowns, 0) self.assertEqual(log.warning_temp_time, 0) self.assertEqual(log.wctemp, 0) self.assertFalse(log.critical_warning & BlockDev.NVMESmartCriticalWarning.SPARE) self.assertFalse(log.critical_warning & BlockDev.NVMESmartCriticalWarning.TEMPERATURE) self.assertFalse(log.critical_warning & BlockDev.NVMESmartCriticalWarning.DEGRADED) self.assertFalse(log.critical_warning & BlockDev.NVMESmartCriticalWarning.READONLY) self.assertFalse(log.critical_warning & BlockDev.NVMESmartCriticalWarning.VOLATILE_MEM) self.assertFalse(log.critical_warning & BlockDev.NVMESmartCriticalWarning.PMR_READONLY) @tag_test(TestTags.CORE) def test_error_log(self): """Test error log retrieval""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_get_error_log_entries("/dev/nonexistent") log = BlockDev.nvme_get_error_log_entries(self.nvme_dev) self.assertIsNotNone(log) # TODO: find a way to spoof drive errors @tag_test(TestTags.CORE) def test_self_test_log(self): """Test self-test log retrieval""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_get_self_test_log("/dev/nonexistent") message = r"NVMe Get Log Page - Device Self-test Log command error: Invalid Field in Command: A reserved coded value or an unsupported value in a defined field|NVMe Get Log Page - Device Self-test Log command error: unrecognized" with self.assertRaisesRegex(GLib.GError, message): # Cannot retrieve self-test log on a nvme target loop devices BlockDev.nvme_get_self_test_log(self.nvme_dev) self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.NO_ERROR), "success") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.ABORTED), "aborted") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.CTRL_RESET), "ctrl_reset") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.NS_REMOVED), "ns_removed") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.ABORTED_FORMAT), "aborted_format") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.FATAL_ERROR), "fatal_error") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.UNKNOWN_SEG_FAIL), "unknown_seg_fail") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.KNOWN_SEG_FAIL), "known_seg_fail") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.ABORTED_UNKNOWN), "aborted_unknown") self.assertEqual(BlockDev.nvme_self_test_result_to_string(BlockDev.NVMESelfTestResult.ABORTED_SANITIZE), "aborted_sanitize") @tag_test(TestTags.CORE) def test_self_test(self): """Test issuing the self-test command""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_device_self_test("/dev/nonexistent", BlockDev.NVMESelfTestAction.SHORT) message = r"Invalid value specified for the self-test action" with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_dev, BlockDev.NVMESelfTestAction.NOT_RUNNING) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_ns_dev, BlockDev.NVMESelfTestAction.NOT_RUNNING) message = r"NVMe Device Self-test command error: Invalid Command Opcode: A reserved coded value or an unsupported value in the command opcode field|NVMe Device Self-test command error: Invalid Queue Identifier: The creation of the I/O Completion Queue failed due to an invalid queue identifier specified as part of the command" with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_dev, BlockDev.NVMESelfTestAction.SHORT) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_ns_dev, BlockDev.NVMESelfTestAction.SHORT) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_dev, BlockDev.NVMESelfTestAction.EXTENDED) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_ns_dev, BlockDev.NVMESelfTestAction.EXTENDED) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_dev, BlockDev.NVMESelfTestAction.VENDOR_SPECIFIC) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_ns_dev, BlockDev.NVMESelfTestAction.VENDOR_SPECIFIC) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_dev, BlockDev.NVMESelfTestAction.ABORT) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_device_self_test(self.nvme_ns_dev, BlockDev.NVMESelfTestAction.ABORT) @tag_test(TestTags.CORE) def test_format(self): """Test issuing the format command""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_format("/dev/nonexistent", 0, 0, BlockDev.NVMEFormatSecureErase.NONE) message = r"Couldn't match desired LBA data block size in a device supported LBA format data sizes" with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_format(self.nvme_ns_dev, 123, 0, BlockDev.NVMEFormatSecureErase.NONE) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_format(self.nvme_dev, 123, 0, BlockDev.NVMEFormatSecureErase.NONE) # format doesn't really work on the kernel loop target message = r"Format NVM command error: Invalid Command Opcode: A reserved coded value or an unsupported value in the command opcode field|Format NVM command error: Invalid Queue Identifier: The creation of the I/O Completion Queue failed due to an invalid queue identifier specified as part of the command" with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_format(self.nvme_ns_dev, 0, 0, BlockDev.NVMEFormatSecureErase.NONE) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_format(self.nvme_dev, 0, 0, BlockDev.NVMEFormatSecureErase.NONE) @tag_test(TestTags.CORE) def test_sanitize_log(self): """Test sanitize log retrieval""" with self.assertRaisesRegex(GLib.GError, r".*Failed to open device .*': No such file or directory"): BlockDev.nvme_get_sanitize_log("/dev/nonexistent") message = r"NVMe Get Log Page - Sanitize Status Log command error: Invalid Field in Command: A reserved coded value or an unsupported value in a defined field|NVMe Get Log Page - Sanitize Status Log command error: unrecognized" with self.assertRaisesRegex(GLib.GError, message): # Cannot retrieve sanitize log on a nvme target loop devices BlockDev.nvme_get_sanitize_log(self.nvme_dev) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_get_sanitize_log(self.nvme_ns_dev) @tag_test(TestTags.CORE) def test_sanitize(self): """Test issuing the sanitize command""" message = r".*Failed to open device .*': No such file or directory" with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_sanitize("/dev/nonexistent", BlockDev.NVMESanitizeAction.BLOCK_ERASE, False, 0, 0, False) message = r"Sanitize command error: Invalid Command Opcode: A reserved coded value or an unsupported value in the command opcode field|Sanitize command error: Invalid Queue Identifier: The creation of the I/O Completion Queue failed due to an invalid queue identifier specified as part of the command" for i in [BlockDev.NVMESanitizeAction.BLOCK_ERASE, BlockDev.NVMESanitizeAction.CRYPTO_ERASE, BlockDev.NVMESanitizeAction.OVERWRITE, BlockDev.NVMESanitizeAction.EXIT_FAILURE]: with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_sanitize(self.nvme_dev, i, False, 0, 0, False) with self.assertRaisesRegex(GLib.GError, message): BlockDev.nvme_sanitize(self.nvme_ns_dev, i, False, 0, 0, False) class NVMeFabricsTestCase(NVMeTest): SUBNQN = 'libblockdev_nvme' DISCOVERY_NQN = 'nqn.2014-08.org.nvmexpress.discovery' def setUp(self): self.dev_files = [] self.hostnqn = get_nvme_hostnqn() self.have_stable_nqn = os.path.exists('/sys/class/dmi/id/product_uuid') def _setup_target(self, num_devices): self.addCleanup(self._clean_up) for i in range(num_devices): self.dev_files += [create_sparse_tempfile("nvmeof_test%d" % i, 1024**3, dir=self.TMPDIR)] setup_nvme_target(self.dev_files, self.SUBNQN) for d in self.dev_files: os.unlink(d) def _clean_up(self): teardown_nvme_target() for d in self.dev_files: self._safe_unlink(d) def _nvme_disconnect(self, subnqn, ignore_errors=False): ret, out, err = run_command("nvme disconnect --nqn=%s" % subnqn) if not ignore_errors and (ret != 0 or 'disconnected 0 ' in out): raise RuntimeError("Error disconnecting the '%s' subsystem NQN: '%s %s'" % (subnqn, out, err)) def _get_sysconf_dir(self): try: makefile = read_file(os.path.join(os.environ['LIBBLOCKDEV_PROJ_DIR'], 'Makefile')) r = re.search(r'sysconfdir = (.*)', makefile) return r.group(1) except: return None @tag_test(TestTags.CORE) def test_connect_single_ns(self): """Test simple connect and disconnect""" # test that no device node exists for given subsystem nqn ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) # nothing to disconnect with self.assertRaisesRegex(GLib.GError, r"No subsystems matching '.*' NQN found."): BlockDev.nvme_disconnect(self.SUBNQN) # nothing to connect to msg = r'Error connecting the controller: ' with self.assertRaisesRegex(GLib.GError, msg): BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, self.hostnqn, None) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.nvme_connect(self.SUBNQN, 'loop', '127.0.0.1', None, None, None, self.hostnqn, None) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, None, None) self._setup_target(1) # make a connection ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, None, None) self.addCleanup(self._nvme_disconnect, self.SUBNQN, ignore_errors=True) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 1) for c in ctrls: self.assertTrue(re.match(r'/dev/nvme[0-9]+', c)) self.assertTrue(os.path.exists(c)) namespaces = find_nvme_ns_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(namespaces), 1) for ns in namespaces: self.assertTrue(re.match(r'/dev/nvme[0-9]+n[0-9]+', ns)) self.assertTrue(os.path.exists(ns)) # make a duplicate connection ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, None, None) self.assertTrue(ret) # should see two controllers now ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 2) for c in ctrls: self.assertTrue(re.match(r'/dev/nvme[0-9]+', c)) self.assertTrue(os.path.exists(c)) # disconnect with self.assertRaisesRegex(GLib.GError, r"No subsystems matching '.*' NQN found."): BlockDev.nvme_disconnect(self.SUBNQN + "xx") with self.assertRaisesRegex(GLib.GError, r"No controllers matching the /dev/nvme.*xx device name found."): BlockDev.nvme_disconnect_by_path(ctrls[0] + "xx") # should disconnect both connections as long the SUBNQN matches BlockDev.nvme_disconnect(self.SUBNQN) for c in ctrls: self.assertFalse(os.path.exists(c)) for ns in namespaces: self.assertFalse(os.path.exists(ns)) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) namespaces = find_nvme_ns_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(namespaces), 0) @tag_test(TestTags.CORE) def test_connect_multiple_ns(self): """Test connect and disconnect multiple namespaces""" NUM_NS = 3 # test that no device node exists for given subsystem nqn ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) self._setup_target(NUM_NS) # make a connection ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, self.hostnqn, None) self.addCleanup(self._nvme_disconnect, self.SUBNQN, ignore_errors=True) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 1) for c in ctrls: self.assertTrue(re.match(r'/dev/nvme[0-9]+', c)) self.assertTrue(os.path.exists(c)) namespaces = find_nvme_ns_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(namespaces), NUM_NS) for ns in namespaces: self.assertTrue(re.match(r'/dev/nvme[0-9]+n[0-9]+', ns)) self.assertTrue(os.path.exists(ns)) # verify the sysfs paths ret, ns_sysfs_path, err = run_command("udevadm info --query=path %s" % ns) if ret != 0: raise RuntimeError("Error getting udev info for %s: '%s'" % (ns, err)) self.assertIsNotNone(ns_sysfs_path) self.assertGreater(len(ns_sysfs_path), 0) ns_sysfs_path = "/sys" + ns_sysfs_path ret, ctrl_sysfs_path, err = run_command("udevadm info --query=path %s" % ctrls[0]) if ret != 0: raise RuntimeError("Error getting udev info for %s: '%s'" % (ctrls[0], err)) self.assertIsNotNone(ctrl_sysfs_path) self.assertGreater(len(ctrl_sysfs_path), 0) ctrl_sysfs_path = "/sys" + ctrl_sysfs_path ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, None, None, None) self.assertIsNotNone(ctrl_sysfs_paths) self.assertEqual(len(ctrl_sysfs_paths), 1) self.assertEqual(ctrl_sysfs_path, ctrl_sysfs_paths[0]) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path + "xxx", None, None, None) self.assertEqual(len(ctrl_sysfs_paths), 0) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, self.SUBNQN, None, None) self.assertEqual(len(ctrl_sysfs_paths), 1) self.assertEqual(ctrl_sysfs_path, ctrl_sysfs_paths[0]) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, self.SUBNQN, self.hostnqn, None) self.assertEqual(len(ctrl_sysfs_paths), 1) self.assertEqual(ctrl_sysfs_path, ctrl_sysfs_paths[0]) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, "unknownsubsysnqn", None, None) self.assertEqual(len(ctrl_sysfs_paths), 0) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, None, "unknownhostnqn", None) self.assertEqual(len(ctrl_sysfs_paths), 0) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, self.SUBNQN, "unknownhostnqn", None) self.assertEqual(len(ctrl_sysfs_paths), 0) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, None, None, "unknownhostid") self.assertEqual(len(ctrl_sysfs_paths), 0) ctrl_sysfs_paths = BlockDev.nvme_find_ctrls_for_ns(ns_sysfs_path, self.SUBNQN, self.hostnqn, "unknownhostid") self.assertEqual(len(ctrl_sysfs_paths), 0) # disconnect BlockDev.nvme_disconnect_by_path(ctrls[0]) for c in ctrls: self.assertFalse(os.path.exists(c)) for ns in namespaces: self.assertFalse(os.path.exists(ns)) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) namespaces = find_nvme_ns_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(namespaces), 0) @tag_test(TestTags.CORE) def test_host_nqn(self): """Test Host NQN/ID manipulation and a simple connect""" HOSTNQN_PATH = '/etc/nvme/hostnqn' HOSTID_PATH = '/etc/nvme/hostid' FAKE_HOSTNQN1 = 'nqn.2014-08.org.nvmexpress:uuid:ffffffff-ffff-ffff-ffff-ffffffffffff' FAKE_HOSTNQN2 = 'nqn.2014-08.org.nvmexpress:uuid:beefbeef-beef-beef-beef-beefdeadbeef' FAKE_HOSTID1 = 'aaaaaaaa-ffff-ffff-ffff-ffffffffffff' FAKE_HOSTID2 = 'beeeeeef-beef-beef-beef-beefdeadbeef' # libnvme might have been configured with a different prefix than libblockdev sysconf_dir = self._get_sysconf_dir() if sysconf_dir != '/etc': self.skipTest("libblockdev was not configured with standard prefix (/usr)") # test that no device node exists for given subsystem nqn ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) # save hostnqn and hostid files try: saved_hostnqn = read_file(HOSTNQN_PATH) self.addCleanup(write_file, HOSTNQN_PATH, saved_hostnqn) except: self.addCleanup(self._safe_unlink, HOSTNQN_PATH) pass try: saved_hostid = read_file(HOSTID_PATH) self.addCleanup(write_file, HOSTID_PATH, saved_hostid) except: self.addCleanup(self._safe_unlink, HOSTID_PATH) pass self._safe_unlink(HOSTNQN_PATH) self._safe_unlink(HOSTID_PATH) hostnqn = BlockDev.nvme_get_host_nqn() self.assertEqual(len(hostnqn), 0) hostnqn = BlockDev.nvme_generate_host_nqn() self.assertTrue(hostnqn.startswith('nqn.2014-08.org.nvmexpress:uuid:')) hostid = BlockDev.nvme_get_host_id() self.assertEqual(len(hostid), 0) # connection without hostnqn set self._setup_target(1) ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, None, None) self.addCleanup(self._nvme_disconnect, self.SUBNQN, ignore_errors=True) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 1) if self.have_stable_nqn: sysfs_hostnqn = read_file('/sys/class/nvme/%s/hostnqn' % os.path.basename(ctrls[0])) self.assertEqual(sysfs_hostnqn.strip(), BlockDev.nvme_generate_host_nqn()) BlockDev.nvme_disconnect(self.SUBNQN) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, FAKE_HOSTNQN1, FAKE_HOSTID1) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 1) sysfs_hostnqn = read_file('/sys/class/nvme/%s/hostnqn' % os.path.basename(ctrls[0])) sysfs_hostid = read_file('/sys/class/nvme/%s/hostid' % os.path.basename(ctrls[0])) self.assertEqual(sysfs_hostnqn.strip(), FAKE_HOSTNQN1) self.assertEqual(sysfs_hostid.strip(), FAKE_HOSTID1) BlockDev.nvme_disconnect(self.SUBNQN) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) # fill with custom IDs ret = BlockDev.nvme_set_host_nqn(FAKE_HOSTNQN1) self.assertTrue(ret) ret = BlockDev.nvme_set_host_id(FAKE_HOSTID1) self.assertTrue(ret) hostnqn = BlockDev.nvme_get_host_nqn() self.assertEqual(hostnqn, FAKE_HOSTNQN1) hostid = BlockDev.nvme_get_host_id() self.assertEqual(hostid, FAKE_HOSTID1) ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, None, None) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 1) sysfs_hostnqn = read_file('/sys/class/nvme/%s/hostnqn' % os.path.basename(ctrls[0])) sysfs_hostid = read_file('/sys/class/nvme/%s/hostid' % os.path.basename(ctrls[0])) self.assertEqual(sysfs_hostnqn.strip(), FAKE_HOSTNQN1) self.assertEqual(sysfs_hostid.strip(), FAKE_HOSTID1) BlockDev.nvme_disconnect(self.SUBNQN) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) ret = BlockDev.nvme_connect(self.SUBNQN, 'loop', None, None, None, None, FAKE_HOSTNQN2, FAKE_HOSTID2) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 1) sysfs_hostnqn = read_file('/sys/class/nvme/%s/hostnqn' % os.path.basename(ctrls[0])) sysfs_hostid = read_file('/sys/class/nvme/%s/hostid' % os.path.basename(ctrls[0])) self.assertEqual(sysfs_hostnqn.strip(), FAKE_HOSTNQN2) self.assertEqual(sysfs_hostid.strip(), FAKE_HOSTID2) BlockDev.nvme_disconnect(self.SUBNQN) ctrls = find_nvme_ctrl_devs_for_subnqn(self.SUBNQN) self.assertEqual(len(ctrls), 0) self._safe_unlink(HOSTNQN_PATH) self._safe_unlink(HOSTID_PATH) @tag_test(TestTags.CORE) def test_persistent_dc(self): """Test connecting a persistent Discovery Controller""" # test that no device node exists for given subsystem nqn ctrls = find_nvme_ctrl_devs_for_subnqn(self.DISCOVERY_NQN) self.assertEqual(len(ctrls), 0) # nothing to disconnect with self.assertRaisesRegex(GLib.GError, r"No subsystems matching '.*' NQN found."): BlockDev.nvme_disconnect(self.DISCOVERY_NQN) # nothing to connect to msg = r'Error connecting the controller: ' with self.assertRaisesRegex(GLib.GError, msg): BlockDev.nvme_connect(self.DISCOVERY_NQN, 'loop', None, None, None, None, self.hostnqn, None) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.nvme_connect(self.DISCOVERY_NQN, 'loop', '127.0.0.1', None, None, None, self.hostnqn, None) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.nvme_connect(self.DISCOVERY_NQN, 'loop', None, None, None, None, None, None) self._setup_target(1) # make a connection ret = BlockDev.nvme_connect(self.DISCOVERY_NQN, 'loop', None, None, None, None, None, None) # TODO: there are some extra steps to make the DC actually persistent, a simple # connect like this appears to be working fine for our needs though. self.addCleanup(self._nvme_disconnect, self.DISCOVERY_NQN, ignore_errors=True) self.assertTrue(ret) ctrls = find_nvme_ctrl_devs_for_subnqn(self.DISCOVERY_NQN) self.assertEqual(len(ctrls), 1) for c in ctrls: self.assertTrue(re.match(r'/dev/nvme[0-9]+', c)) self.assertTrue(os.path.exists(c)) namespaces = find_nvme_ns_devs_for_subnqn(self.DISCOVERY_NQN) self.assertEqual(len(namespaces), 0) # issue an IDENTIFY_CTRL command info = BlockDev.nvme_get_controller_info(ctrls[0]) self.assertEqual(info.ctrl_id, 1) self.assertEqual(info.pci_vendor_id, 0) self.assertEqual(info.pci_subsys_vendor_id, 0) self.assertIsNone(info.fguid) self.assertIn("Linux", info.model_number) self.assertGreater(len(info.serial_number), 0) self.assertGreater(len(info.firmware_ver), 0) self.assertGreater(len(info.nvme_ver), 0) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.MULTIPORT) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.MULTICTRL) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SRIOV) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.ANA_REPORTING) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.FORMAT) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.FORMAT_ALL_NS) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.NS_MGMT) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SELFTEST) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SELFTEST_SINGLE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SANITIZE_CRYPTO) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SANITIZE_BLOCK) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SANITIZE_OVERWRITE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SECURE_ERASE_ALL_NS) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.SECURE_ERASE_CRYPTO) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.STORAGE_DEVICE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.ENCLOSURE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.MGMT_PCIE) self.assertFalse(info.features & BlockDev.NVMEControllerFeature.MGMT_SMBUS) self.assertEqual(info.controller_type, BlockDev.NVMEControllerType.DISCOVERY) self.assertEqual(info.selftest_ext_time, 0) self.assertEqual(info.hmb_pref_size, 0) self.assertEqual(info.hmb_min_size, 0) self.assertEqual(info.size_total, 0) self.assertEqual(info.size_unalloc, 0) self.assertEqual(info.num_namespaces, 0) self.assertEqual(info.subsysnqn, self.DISCOVERY_NQN) # disconnect with self.assertRaisesRegex(GLib.GError, r"No subsystems matching '.*' NQN found."): BlockDev.nvme_disconnect(self.DISCOVERY_NQN + "xx") with self.assertRaisesRegex(GLib.GError, r"No controllers matching the /dev/nvme.*xx device name found."): BlockDev.nvme_disconnect_by_path(ctrls[0] + "xx") BlockDev.nvme_disconnect(self.DISCOVERY_NQN) for c in ctrls: self.assertFalse(os.path.exists(c)) for ns in namespaces: self.assertFalse(os.path.exists(ns)) ctrls = find_nvme_ctrl_devs_for_subnqn(self.DISCOVERY_NQN) self.assertEqual(len(ctrls), 0) namespaces = find_nvme_ns_devs_for_subnqn(self.DISCOVERY_NQN) self.assertEqual(len(namespaces), 0)