Adding upstream version 0.8.upstream/0.8upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 484 insertions, 0 deletions

diff --git a/nvmet/test_nvmet.py b/nvmet/test_nvmet.py
new file mode 100644
index 0000000..13e2f6c
--- /dev/null
+++ b/nvmet/test_nvmet.py
@@ -0,0 +1,484 @@
+
+import os
+import random
+import stat
+import string
+import unittest
+import nvmet.nvme as nvme
+
+# Default test devices are ram disks, but allow user to specify different
+# block devices or files.
+NVMET_TEST_DEVICES = os.getenv("NVMET_TEST_DEVICES",
+                               "/dev/ram0,/dev/ram1").split(',')
+
+
+def test_devices_present():
+    return len([x for x in NVMET_TEST_DEVICES
+                if os.path.exists(x) and
+                (stat.S_ISBLK(os.stat(x).st_mode) or os.path.isfile(x))]) >= 2
+
+
+class TestNvmet(unittest.TestCase):
+    def test_subsystem(self):
+        root = nvme.Root()
+        root.clear_existing()
+        for s in root.subsystems:
+            self.assertTrue(False, 'Found Subsystem after clear')
+
+        # create mode
+        s1 = nvme.Subsystem(nqn='testnqn1', mode='create')
+        self.assertIsNotNone(s1)
+        self.assertEqual(len(list(root.subsystems)), 1)
+
+        # any mode, should create
+        s2 = nvme.Subsystem(nqn='testnqn2', mode='any')
+        self.assertIsNotNone(s2)
+        self.assertEqual(len(list(root.subsystems)), 2)
+
+        # random name
+        s3 = nvme.Subsystem(mode='create')
+        self.assertIsNotNone(s3)
+        self.assertEqual(len(list(root.subsystems)), 3)
+
+        # duplicate
+        self.assertRaises(nvme.CFSError, nvme.Subsystem,
+                          nqn='testnqn1', mode='create')
+        self.assertEqual(len(list(root.subsystems)), 3)
+
+        # lookup using any, should not create
+        s = nvme.Subsystem(nqn='testnqn1', mode='any')
+        self.assertEqual(s1, s)
+        self.assertEqual(len(list(root.subsystems)), 3)
+
+        # lookup only
+        s = nvme.Subsystem(nqn='testnqn2', mode='lookup')
+        self.assertEqual(s2, s)
+        self.assertEqual(len(list(root.subsystems)), 3)
+
+        # lookup without nqn
+        self.assertRaises(nvme.CFSError, nvme.Subsystem, mode='lookup')
+
+        # and delete them all
+        for s in root.subsystems:
+            s.delete()
+        self.assertEqual(len(list(root.subsystems)), 0)
+
+    def test_namespace(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        s = nvme.Subsystem(nqn='testnqn', mode='create')
+        for n in s.namespaces:
+            self.assertTrue(False, 'Found Namespace in new Subsystem')
+
+        # create mode
+        n1 = nvme.Namespace(s, nsid=3, mode='create')
+        self.assertIsNotNone(n1)
+        self.assertEqual(len(list(s.namespaces)), 1)
+
+        # any mode, should create
+        n2 = nvme.Namespace(s, nsid=2, mode='any')
+        self.assertIsNotNone(n2)
+        self.assertEqual(len(list(s.namespaces)), 2)
+
+        # create without nsid, should pick lowest available
+        n3 = nvme.Namespace(s, mode='create')
+        self.assertIsNotNone(n3)
+        self.assertEqual(n3.nsid, 1)
+        self.assertEqual(len(list(s.namespaces)), 3)
+
+        n4 = nvme.Namespace(s, mode='create')
+        self.assertIsNotNone(n4)
+        self.assertEqual(n4.nsid, 4)
+        self.assertEqual(len(list(s.namespaces)), 4)
+
+        # duplicate
+        self.assertRaises(nvme.CFSError, nvme.Namespace, 1, mode='create')
+        self.assertEqual(len(list(s.namespaces)), 4)
+
+        # lookup using any, should not create
+        n = nvme.Namespace(s, nsid=3, mode='any')
+        self.assertEqual(n1, n)
+        self.assertEqual(len(list(s.namespaces)), 4)
+
+        # lookup only
+        n = nvme.Namespace(s, nsid=2, mode='lookup')
+        self.assertEqual(n2, n)
+        self.assertEqual(len(list(s.namespaces)), 4)
+
+        # lookup without nsid
+        self.assertRaises(nvme.CFSError, nvme.Namespace, None, mode='lookup')
+
+        # and delete them all
+        for n in s.namespaces:
+            n.delete()
+        self.assertEqual(len(list(s.namespaces)), 0)
+
+    @unittest.skipUnless(test_devices_present(),
+                         "Devices %s not available or suitable" % ','.join(NVMET_TEST_DEVICES))
+    def test_namespace_attrs(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        s = nvme.Subsystem(nqn='testnqn', mode='create')
+        n = nvme.Namespace(s, mode='create')
+
+        self.assertFalse(n.get_enable())
+        self.assertTrue('device' in n.attr_groups)
+        self.assertTrue('path' in n.list_attrs('device'))
+
+        # no device set yet, should fail
+        self.assertRaises(nvme.CFSError, n.set_enable, 1)
+
+        # now set a path and enable
+        n.set_attr('device', 'path', NVMET_TEST_DEVICES[0])
+        n.set_enable(1)
+        self.assertTrue(n.get_enable())
+
+        # test double enable
+        n.set_enable(1)
+
+        # test that we can't write to attrs while enabled
+        self.assertRaises(nvme.CFSError, n.set_attr, 'device', 'path',
+                          NVMET_TEST_DEVICES[1])
+        self.assertRaises(nvme.CFSError, n.set_attr, 'device', 'nguid',
+                          '15f7767b-50e7-4441-949c-75b99153dea7')
+
+        # disable: once and twice
+        n.set_enable(0)
+        n.set_enable(0)
+
+        # enable again, and remove while enabled
+        n.set_enable(1)
+        n.delete()
+
+    def test_recursive_delete(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        s = nvme.Subsystem(nqn='testnqn', mode='create')
+        n1 = nvme.Namespace(s, mode='create')
+        n2 = nvme.Namespace(s, mode='create')
+
+        s.delete()
+        self.assertEqual(len(list(root.subsystems)), 0)
+
+    def test_port(self):
+        root = nvme.Root()
+        root.clear_existing()
+        for p in root.ports:
+            self.assertTrue(False, 'Found Port after clear')
+
+        # create mode
+        p1 = nvme.Port(portid=0, mode='create')
+        self.assertIsNotNone(p1)
+        self.assertEqual(len(list(root.ports)), 1)
+
+        # any mode, should create
+        p2 = nvme.Port(portid=1, mode='any')
+        self.assertIsNotNone(p2)
+        self.assertEqual(len(list(root.ports)), 2)
+
+        # duplicate
+        self.assertRaises(nvme.CFSError, nvme.Port,
+                          portid=0, mode='create')
+        self.assertEqual(len(list(root.ports)), 2)
+
+        # lookup using any, should not create
+        p = nvme.Port(portid=0, mode='any')
+        self.assertEqual(p1, p)
+        self.assertEqual(len(list(root.ports)), 2)
+
+        # lookup only
+        p = nvme.Port(portid=1, mode='lookup')
+        self.assertEqual(p2, p)
+        self.assertEqual(len(list(root.ports)), 2)
+
+        # and delete them all
+        for p in root.ports:
+            p.delete()
+        self.assertEqual(len(list(root.ports)), 0)
+
+    def test_loop_port(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        s = nvme.Subsystem(nqn='testnqn', mode='create')
+        p = nvme.Port(portid=0, mode='create')
+
+        # subsystem doesn't exists, should fail
+        self.assertRaises(nvme.CFSError, p.add_subsystem, 'invalidnqn')
+
+        self.assertTrue('addr' in p.attr_groups)
+
+        # no trtype set yet, should fail
+        self.assertRaises(nvme.CFSError, p.add_subsystem, 'testnqn')
+
+        # now set trtype to loop and other attrs and enable
+        p.set_attr('addr', 'trtype', 'loop')
+        p.set_attr('addr', 'adrfam', 'ipv4')
+        p.set_attr('addr', 'traddr', '192.168.0.1')
+        p.set_attr('addr', 'treq', 'not required')
+        p.set_attr('addr', 'trsvcid', '1023')
+        p.add_subsystem('testnqn')
+
+        # test double add
+        self.assertRaises(nvme.CFSError, p.add_subsystem, 'testnqn')
+
+        # test that we can't write to attrs while enabled
+        self.assertRaises(nvme.CFSError, p.set_attr, 'addr', 'trtype',
+                          'rdma')
+        self.assertRaises(nvme.CFSError, p.set_attr, 'addr', 'adrfam',
+                          'ipv6')
+        self.assertRaises(nvme.CFSError, p.set_attr, 'addr', 'traddr',
+                          '10.0.0.1')
+        self.assertRaises(nvme.CFSError, p.set_attr, 'addr', 'treq',
+                          'required')
+        self.assertRaises(nvme.CFSError, p.set_attr, 'addr', 'trsvcid',
+                          '21')
+
+        # remove: once and twice
+        p.remove_subsystem('testnqn')
+        self.assertRaises(nvme.CFSError, p.remove_subsystem, 'testnqn')
+
+        # check that the attrs haven't been tampered with
+        self.assertEqual(p.get_attr('addr', 'trtype'), 'loop')
+        self.assertEqual(p.get_attr('addr', 'adrfam'), 'ipv4')
+        self.assertEqual(p.get_attr('addr', 'traddr'), '192.168.0.1')
+        self.assertEqual(p.get_attr('addr', 'treq'), 'not required')
+        self.assertEqual(p.get_attr('addr', 'trsvcid'), '1023')
+
+        # add again, and try to remove while enabled
+        p.add_subsystem('testnqn')
+        p.delete()
+
+    def test_host(self):
+        root = nvme.Root()
+        root.clear_existing()
+        for p in root.hosts:
+            self.assertTrue(False, 'Found Host after clear')
+
+        # create mode
+        h1 = nvme.Host(nqn='foo', mode='create')
+        self.assertIsNotNone(h1)
+        self.assertEqual(len(list(root.hosts)), 1)
+
+        # any mode, should create
+        h2 = nvme.Host(nqn='bar', mode='any')
+        self.assertIsNotNone(h2)
+        self.assertEqual(len(list(root.hosts)), 2)
+
+        # duplicate
+        self.assertRaises(nvme.CFSError, nvme.Host,
+                          'foo', mode='create')
+        self.assertEqual(len(list(root.hosts)), 2)
+
+        # lookup using any, should not create
+        h = nvme.Host('foo', mode='any')
+        self.assertEqual(h1, h)
+        self.assertEqual(len(list(root.hosts)), 2)
+
+        # lookup only
+        h = nvme.Host('bar', mode='lookup')
+        self.assertEqual(h2, h)
+        self.assertEqual(len(list(root.hosts)), 2)
+
+        # and delete them all
+        for h in root.hosts:
+            h.delete()
+        self.assertEqual(len(list(root.hosts)), 0)
+
+    def test_referral(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        # create port
+        p = nvme.Port(portid=1, mode='create')
+        self.assertEqual(len(list(p.referrals)), 0)
+
+        # create mode
+        r1 = nvme.Referral(p, name="1", mode='create')
+        self.assertIsNotNone(r1)
+        self.assertEqual(len(list(p.referrals)), 1)
+
+        # any mode, should create
+        r2 = nvme.Referral(p, name="2", mode='any')
+        self.assertIsNotNone(r2)
+        self.assertEqual(len(list(p.referrals)), 2)
+
+        # duplicate
+        self.assertRaises(nvme.CFSError, nvme.Referral,
+                          p, name="2", mode='create')
+        self.assertEqual(len(list(p.referrals)), 2)
+
+        # lookup using any, should not create
+        r = nvme.Referral(p, name="1", mode='any')
+        self.assertEqual(r1, r)
+        self.assertEqual(len(list(p.referrals)), 2)
+
+        # lookup only
+        r = nvme.Referral(p, name="2", mode='lookup')
+        self.assertEqual(r2, r)
+        self.assertEqual(len(list(p.referrals)), 2)
+
+        # non-existent lookup
+        self.assertRaises(nvme.CFSError, nvme.Referral, p, name="foo",
+                          mode='lookup')
+
+        # basic state
+        self.assertTrue('addr' in r.attr_groups)
+        self.assertFalse(r.get_enable())
+
+        # now set trtype to loop and other attrs and enable
+        r.set_attr('addr', 'trtype', 'loop')
+        r.set_attr('addr', 'adrfam', 'ipv4')
+        r.set_attr('addr', 'traddr', '192.168.0.1')
+        r.set_attr('addr', 'treq', 'not required')
+        r.set_attr('addr', 'trsvcid', '1023')
+        r.set_enable(1)
+
+        # test double enable
+        r.set_enable(1)
+
+        # test that we can't write to attrs while enabled
+        self.assertRaises(nvme.CFSError, r.set_attr, 'addr', 'trtype',
+                          'rdma')
+        self.assertRaises(nvme.CFSError, r.set_attr, 'addr', 'adrfam',
+                          'ipv6')
+        self.assertRaises(nvme.CFSError, r.set_attr, 'addr', 'traddr',
+                          '10.0.0.1')
+        self.assertRaises(nvme.CFSError, r.set_attr, 'addr', 'treq',
+                          'required')
+        self.assertRaises(nvme.CFSError, r.set_attr, 'addr', 'trsvcid',
+                          '21')
+
+        # disable: once and twice
+        r.set_enable(0)
+        r.set_enable(0)
+
+        # check that the attrs haven't been tampered with
+        self.assertEqual(r.get_attr('addr', 'trtype'), 'loop')
+        self.assertEqual(r.get_attr('addr', 'adrfam'), 'ipv4')
+        self.assertEqual(r.get_attr('addr', 'traddr'), '192.168.0.1')
+        self.assertEqual(r.get_attr('addr', 'treq'), 'not required')
+        self.assertEqual(r.get_attr('addr', 'trsvcid'), '1023')
+
+        # enable again, and try to remove while enabled
+        r.set_enable(1)
+        r.delete()
+
+        # remove the other one while disabled:
+        r1.delete()
+        self.assertEqual(len(list(p.referrals)), 0)
+
+    def test_allowed_hosts(self):
+        root = nvme.Root()
+
+        h = nvme.Host(nqn='hostnqn', mode='create')
+
+        s = nvme.Subsystem(nqn='testnqn', mode='create')
+
+        # add allowed_host
+        s.add_allowed_host(nqn='hostnqn')
+
+        # duplicate
+        self.assertRaises(nvme.CFSError, s.add_allowed_host, 'hostnqn')
+
+        # invalid
+        self.assertRaises(nvme.CFSError, s.add_allowed_host, 'invalid')
+
+        # remove again
+        s.remove_allowed_host('hostnqn')
+
+        # duplicate removal
+        self.assertRaises(nvme.CFSError, s.remove_allowed_host, 'hostnqn')
+
+        # invalid removal
+        self.assertRaises(nvme.CFSError, s.remove_allowed_host, 'foobar')
+
+    def test_invalid_input(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        self.assertRaises(nvme.CFSError, nvme.Subsystem,
+                          nqn='', mode='create')
+        self.assertRaises(nvme.CFSError, nvme.Subsystem,
+                          nqn='/', mode='create')
+
+        for l in [ 257, 512, 1024, 2048 ]:
+            toolong = ''.join(random.choice(string.ascii_lowercase)
+                              for i in range(l))
+            self.assertRaises(nvme.CFSError, nvme.Subsystem,
+                              nqn=toolong, mode='create')
+
+        discover_nqn = "nqn.2014-08.org.nvmexpress.discovery"
+        self.assertRaises(nvme.CFSError, nvme.Subsystem,
+                          nqn=discover_nqn, mode='create')
+
+        self.assertRaises(nvme.CFSError, nvme.Port,
+                          portid=1 << 17, mode='create')
+
+    @unittest.skipUnless(test_devices_present(),
+                         "Devices %s not available or suitable" % ','.join(
+                             NVMET_TEST_DEVICES))
+    def test_save_restore(self):
+        root = nvme.Root()
+        root.clear_existing()
+
+        h = nvme.Host(nqn='hostnqn', mode='create')
+
+        s = nvme.Subsystem(nqn='testnqn', mode='create')
+        s.add_allowed_host(nqn='hostnqn')
+
+        s2 = nvme.Subsystem(nqn='testnqn2', mode='create')
+        s2.set_attr('attr', 'allow_any_host', 1)
+
+        n = nvme.Namespace(s, nsid=42, mode='create')
+        n.set_attr('device', 'path', NVMET_TEST_DEVICES[0])
+        n.set_enable(1)
+
+        nguid = n.get_attr('device', 'nguid')
+
+        p = nvme.Port(portid=66, mode='create')
+        p.set_attr('addr', 'trtype', 'loop')
+        p.set_attr('addr', 'adrfam', 'ipv4')
+        p.set_attr('addr', 'traddr', '192.168.0.1')
+        p.set_attr('addr', 'treq', 'not required')
+        p.set_attr('addr', 'trsvcid', '1023')
+        p.add_subsystem('testnqn')
+
+        # save, clear, and restore
+        root.save_to_file('test.json')
+        root.clear_existing()
+        root.restore_from_file('test.json')
+
+        # additional restores should fai
+        self.assertRaises(nvme.CFSError, root.restore_from_file,
+                          'test.json', False)
+
+        # ... unless forced!
+        root.restore_from_file('test.json', True)
+
+        # rebuild our view of the world
+        h = nvme.Host(nqn='hostnqn', mode='lookup')
+        s = nvme.Subsystem(nqn='testnqn', mode='lookup')
+        s2 = nvme.Subsystem(nqn='testnqn2', mode='lookup')
+        n = nvme.Namespace(s, nsid=42, mode='lookup')
+        p = nvme.Port(portid=66, mode='lookup')
+
+        self.assertEqual(s.get_attr('attr', 'allow_any_host'), "0")
+        self.assertEqual(s2.get_attr('attr', 'allow_any_host'), "1")
+        self.assertIn('hostnqn', s.allowed_hosts)
+
+        # and check everything is still the same
+        self.assertTrue(n.get_enable())
+        self.assertEqual(n.get_attr('device', 'path'), NVMET_TEST_DEVICES[0])
+        self.assertEqual(n.get_attr('device', 'nguid'), nguid)
+
+        self.assertEqual(p.get_attr('addr', 'trtype'), 'loop')
+        self.assertEqual(p.get_attr('addr', 'adrfam'), 'ipv4')
+        self.assertEqual(p.get_attr('addr', 'traddr'), '192.168.0.1')
+        self.assertEqual(p.get_attr('addr', 'treq'), 'not required')
+        self.assertEqual(p.get_attr('addr', 'trsvcid'), '1023')
+        self.assertIn('testnqn', p.subsystems)
+        self.assertNotIn('testtnqn2', p.subsystems)