diff options
Diffstat (limited to 'nvmet/test_nvmet.py')
| -rw-r--r-- | nvmet/test_nvmet.py | 484 |
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) |