python-libnvme: drop 1.x backward-compatibility shims

Remove the deprecated method wrappers `discovery_ctrl_set()` and
`persistent_set()` on `ctrl`, and the module-level aliases
`NVME_LOG_LID_DISCOVER` and `root` that existed for compatibility
with libnvme 1.x code. Callers should use the property setters
(`ctrl.discovery_ctrl`, `ctrl.persistent`) and the canonical names
(`NVME_LOG_LID_DISCOVERY`, `global_ctx`) directly.

Add a new Python test module (test-objects.py) covering object
creation and property access, iterator correctness, boolean flag
get/set, and error handling for disconnected controllers, replacing
the removed deprecation-warning tests.

The new tests allowed finding and fixing a pre-existing bug in
nvme.i where a NULL returned by nvme_read_hostnqn() or
nvme_read_hostid() was not being handled and would cause a crash.

Signed-off-by: Martin Belanger <[email protected]>

cac

Author: Martin Belanger

libnvme/libnvme/meson.build

@@ -75,10 +75,11 @@ if want_python
     # Test section
     test('libnvme - python-import-libnvme', python3, args: ['-c', 'from libnvme import nvme'], env: test_env, depends: pynvme_clib)
 
-    py_tests = [ 
+    py_tests = [
         [ 'create-ctrl-object', [ files('tests/create-ctrl-obj.py'), ] ],
         [ 'sigsegv-during-gc', [ files('tests/gc.py'), ] ],
         [ 'read-nbft-file', [ files('tests/test-nbft.py'), '--filename', join_paths(meson.current_source_dir(), 'tests', 'NBFT') ] ],
+        [ 'object-properties-and-errors', [ files('tests/test-objects.py'), ] ],
     ]
     foreach test: py_tests
         description = test[0]

libnvme/libnvme/nvme.i

@@ -19,11 +19,11 @@
 
 %allowexception;
 
-%rename(global_ctx)      nvme_global_ctx;
-%rename(host)      nvme_host;
-%rename(ctrl)      nvme_ctrl;
-%rename(subsystem) nvme_subsystem;
-%rename(ns)        nvme_ns;
+%rename(global_ctx) nvme_global_ctx;
+%rename(host)       nvme_host;
+%rename(ctrl)       nvme_ctrl;
+%rename(subsystem)  nvme_subsystem;
+%rename(ns)         nvme_ns;
 
 %{
 	#include <ccan/list/list.h>
@@ -40,13 +40,13 @@
 	}
 	PyObject *read_hostnqn() {
 		char * val = nvme_read_hostnqn();
-		PyObject * obj = PyUnicode_FromString(val);
+		PyObject * obj = val ? PyUnicode_FromString(val) : Py_NewRef(Py_None);
 		free(val);
 		return obj;
 	}
 	PyObject *read_hostid() {
 		char * val = nvme_read_hostid();
-		PyObject * obj = PyUnicode_FromString(val);
+		PyObject * obj = val ? PyUnicode_FromString(val) : Py_NewRef(Py_None);
 		free(val);
 		return obj;
 	}
@@ -461,8 +461,9 @@ struct nvme_ctrl {
 		char *dhchap_ctrl_key;
 
 		/**
-		 * We are remapping the following members of the C code's
-		 * nvme_ctrl_t to different names in Python. Here's the mapping:
+		 * We are remapping the following member(s) of the C code's
+		 * nvme_ctrl_t to different name(s) in Python. Here's the
+		 * mapping:
 		 *
 		 * C code                 Python (SWIG)
 		 * =====================  =====================
@@ -703,14 +704,6 @@ struct nvme_ns {
 		return $self;
 	}
 
-	%pythoncode %{
-	def discovery_ctrl_set(self, discovery: bool):
-	    r"""DEPRECATED METHOD: Use property setter instead (e.g. ctrl.discovery_ctrl = True)"""
-	    import warnings
-	    warnings.warn("Use property setter instead (e.g. ctrl_obj.discovery_ctrl = True)", DeprecationWarning, stacklevel=2)
-	    return _nvme.ctrl_discovery_ctrl_set(self, discovery)
-	%}
-
 	bool init(struct nvme_host *h, int instance) {
 		return nvme_init_ctrl(h, $self, instance) == 0;
 	}
@@ -738,13 +731,6 @@ struct nvme_ns {
 	bool connected() {
 		return nvme_ctrl_get_name($self) != NULL;
 	}
-	%pythoncode %{
-	def persistent_set(self, persistent: bool):
-	    r"""DEPRECATED METHOD: Use property setter instead (e.g. ctrl.persistent = True)"""
-	    import warnings
-	    warnings.warn("Use property setter instead (e.g. ctrl_obj.persistent = True)", DeprecationWarning, stacklevel=2)
-	    return _nvme.ctrl_persistent_set(self, persistent)
-	%}
 	void rescan() {
 		nvme_rescan_ctrl($self);
 	}
@@ -1268,12 +1254,3 @@ PyObject *nbft_get(struct nvme_global_ctx *ctx, const char * filename);
 %rename($ignore, %$isvariable) "";  // ignore all variables
 
 %include "../src/nvme/types.h"
-
-
-%pythoncode %{
-# Definitions for backward compatibility between libnvme 1.x and 3.x (there is no 2.x)
-# Some terms (class/variable names) were changed and this allows running older python
-# code written for libnvme 1.x with libnvme 3.x (and possibly later).
-NVME_LOG_LID_DISCOVER = _nvme.NVME_LOG_LID_DISCOVERY
-root = global_ctx
-%}

libnvme/libnvme/tests/test-objects.py

@@ -0,0 +1,257 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: LGPL-2.1-or-later
+"""Unit tests for the libnvme Python bindings.
+
+These tests cover object creation, property access, and error handling.
+They do not require real NVMe hardware to run.
+"""
+import gc
+import unittest
+from libnvme import nvme
+
+
+class TestConstants(unittest.TestCase):
+    """Verify that well-known constants are accessible and have correct values."""
+
+    def test_disc_subsys_name_is_string(self):
+        self.assertIsInstance(nvme.NVME_DISC_SUBSYS_NAME, str)
+
+    def test_disc_subsys_name_value(self):
+        self.assertEqual(nvme.NVME_DISC_SUBSYS_NAME, 'nqn.2014-08.org.nvmexpress.discovery')
+
+    def test_log_lid_discovery_is_int(self):
+        self.assertIsInstance(nvme.NVME_LOG_LID_DISCOVERY, int)
+
+
+
+class TestGlobalCtx(unittest.TestCase):
+
+    def test_creation_no_args(self):
+        ctx = nvme.global_ctx()
+        self.assertIsNotNone(ctx)
+
+    def test_context_manager(self):
+        with nvme.global_ctx() as ctx:
+            self.assertIsNotNone(ctx)
+
+    def test_hosts_iterator_returns_list(self):
+        ctx = nvme.global_ctx()
+        hosts = list(ctx.hosts())
+        self.assertIsInstance(hosts, list)
+
+    def test_refresh_topology_does_not_raise(self):
+        ctx = nvme.global_ctx()
+        ctx.refresh_topology()
+
+    def test_log_level_all_valid_levels(self):
+        ctx = nvme.global_ctx()
+        for level in ('debug', 'info', 'notice', 'warning', 'err', 'crit', 'alert', 'emerg'):
+            with self.subTest(level=level):
+                ctx.log_level(level)
+
+
+class TestHost(unittest.TestCase):
+
+    def setUp(self):
+        self.ctx = nvme.global_ctx()
+
+    def tearDown(self):
+        self.ctx = None
+        gc.collect()
+
+    def test_creation_default(self):
+        host = nvme.host(self.ctx)
+        self.assertIsNotNone(host)
+
+    def test_creation_with_explicit_hostnqn(self):
+        hostnqn = 'nqn.2014-08.com.example:test-host-creation'
+        host = nvme.host(self.ctx, hostnqn=hostnqn)
+        self.assertIsNotNone(host)
+        self.assertEqual(host.hostnqn, hostnqn)
+
+    def test_creation_with_hostnqn_and_hostid(self):
+        hostnqn = 'nqn.2014-08.com.example:test-host-props'
+        hostid = '11111111-2222-3333-4444-555555555555'
+        host = nvme.host(self.ctx, hostnqn=hostnqn, hostid=hostid)
+        self.assertEqual(host.hostnqn, hostnqn)
+        self.assertEqual(host.hostid, hostid)
+
+    def test_creation_with_hostsymname(self):
+        hostnqn = 'nqn.2014-08.com.example:test-host-symname'
+        symname = 'my-storage-host'
+        host = nvme.host(self.ctx, hostnqn=hostnqn, hostsymname=symname)
+        self.assertEqual(host.hostsymname, symname)
+
+    def test_set_symname(self):
+        hostnqn = 'nqn.2014-08.com.example:test-host-set-symname'
+        host = nvme.host(self.ctx, hostnqn=hostnqn)
+        host.set_symname('updated-symname')
+        self.assertEqual(host.hostsymname, 'updated-symname')
+
+    def test_dhchap_host_key_is_none_by_default(self):
+        hostnqn = 'nqn.2014-08.com.example:test-host-dhchap'
+        host = nvme.host(self.ctx, hostnqn=hostnqn)
+        self.assertIsNone(host.dhchap_host_key)
+
+    def test_subsystems_iterator_returns_list(self):
+        host = nvme.host(self.ctx)
+        subsystems = list(host.subsystems())
+        self.assertIsInstance(subsystems, list)
+
+    def test_str_contains_class_name(self):
+        host = nvme.host(self.ctx)
+        self.assertIn('nvme.host', str(host))
+
+    def test_context_manager(self):
+        with nvme.host(self.ctx) as h:
+            self.assertIsNotNone(h)
+
+
+class TestCtrl(unittest.TestCase):
+
+    def setUp(self):
+        self.ctx = nvme.global_ctx()
+        self.subsysnqn = nvme.NVME_DISC_SUBSYS_NAME
+
+    def tearDown(self):
+        self.ctx = None
+        gc.collect()
+
+    def _make_loop_ctrl(self):
+        return nvme.ctrl(self.ctx, subsysnqn=self.subsysnqn, transport='loop')
+
+    def test_creation_loop_transport(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertIsNotNone(ctrl)
+
+    def test_creation_tcp_transport_with_traddr(self):
+        ctrl = nvme.ctrl(
+            self.ctx,
+            subsysnqn=self.subsysnqn,
+            transport='tcp',
+            traddr='192.168.1.1',
+            trsvcid='4420',
+        )
+        self.assertIsNotNone(ctrl)
+
+    def test_transport_property(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertEqual(ctrl.transport, 'loop')
+
+    def test_subsysnqn_property(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertEqual(ctrl.subsysnqn, self.subsysnqn)
+
+    def test_traddr_property(self):
+        ctrl = nvme.ctrl(
+            self.ctx,
+            subsysnqn=self.subsysnqn,
+            transport='tcp',
+            traddr='10.0.0.1',
+        )
+        self.assertEqual(ctrl.traddr, '10.0.0.1')
+
+    def test_trsvcid_property(self):
+        ctrl = nvme.ctrl(
+            self.ctx,
+            subsysnqn=self.subsysnqn,
+            transport='tcp',
+            traddr='10.0.0.1',
+            trsvcid='8009',
+        )
+        self.assertEqual(ctrl.trsvcid, '8009')
+
+    def test_connected_returns_false_before_connect(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertFalse(ctrl.connected())
+
+    def test_name_is_none_before_connect(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertIsNone(ctrl.name)
+
+    def test_str_contains_transport(self):
+        ctrl = self._make_loop_ctrl()
+        s = str(ctrl)
+        self.assertIn('loop', s)
+
+    def test_context_manager(self):
+        with nvme.ctrl(self.ctx, subsysnqn=self.subsysnqn, transport='loop') as c:
+            self.assertIsNotNone(c)
+
+    def test_namespaces_iterator_returns_list(self):
+        ctrl = self._make_loop_ctrl()
+        nss = list(ctrl.namespaces())
+        self.assertIsInstance(nss, list)
+
+    def test_discovery_ctrl_flag_default_false(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertFalse(ctrl.discovery_ctrl)
+
+    def test_discovery_ctrl_flag_set_and_clear(self):
+        ctrl = self._make_loop_ctrl()
+        ctrl.discovery_ctrl = True
+        self.assertTrue(ctrl.discovery_ctrl)
+        ctrl.discovery_ctrl = False
+        self.assertFalse(ctrl.discovery_ctrl)
+
+    def test_persistent_flag_default_false(self):
+        ctrl = self._make_loop_ctrl()
+        self.assertFalse(ctrl.persistent)
+
+    def test_persistent_flag_set(self):
+        ctrl = self._make_loop_ctrl()
+        ctrl.persistent = True
+        self.assertTrue(ctrl.persistent)
+
+    def test_unique_discovery_ctrl_flag(self):
+        ctrl = self._make_loop_ctrl()
+        ctrl.unique_discovery_ctrl = True
+        self.assertTrue(ctrl.unique_discovery_ctrl)
+
+    def test_multiple_ctrls_same_ctx(self):
+        """Multiple controllers can be created under the same context."""
+        ctrls = [self._make_loop_ctrl() for _ in range(5)]
+        self.assertEqual(len(ctrls), 5)
+        for c in ctrls:
+            self.assertFalse(c.connected())
+
+
+class TestCtrlErrorHandling(unittest.TestCase):
+    """Error paths that can be exercised without real hardware."""
+
+    def setUp(self):
+        self.ctx = nvme.global_ctx()
+        self.ctrl = nvme.ctrl(
+            self.ctx,
+            subsysnqn=nvme.NVME_DISC_SUBSYS_NAME,
+            transport='loop',
+        )
+
+    def tearDown(self):
+        self.ctrl = None
+        self.ctx = None
+        gc.collect()
+
+    def test_disconnect_unconnected_raises_attribute_error(self):
+        with self.assertRaises(AttributeError):
+            self.ctrl.disconnect()
+
+    def test_discover_unconnected_raises_attribute_error(self):
+        with self.assertRaises(AttributeError):
+            self.ctrl.discover()
+
+
+class TestHelperFunctions(unittest.TestCase):
+    """Module-level helper functions exposed by the bindings."""
+
+    def test_read_hostnqn_returns_string_or_none(self):
+        hostnqn = nvme.read_hostnqn()
+        self.assertIsInstance(hostnqn, (str, type(None)))
+
+    def test_read_hostid_returns_string_or_none(self):
+        hostid = nvme.read_hostid()
+        self.assertIsInstance(hostid, (str, type(None)))
+
+
+if __name__ == '__main__':
+    unittest.main()