summaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/NBFTbin0 -> 1017 bytes
-rw-r--r--test/NBFT-Empty0
-rw-r--r--test/meson.build49
-rwxr-xr-xtest/test-avahi.py20
-rwxr-xr-xtest/test-config.py5
-rwxr-xr-xtest/test-defs.py59
-rwxr-xr-xtest/test-gutil.py33
-rwxr-xr-xtest/test-iputil.py16
-rwxr-xr-xtest/test-nbft.py105
-rwxr-xr-xtest/test-nbft_conf.py56
-rwxr-xr-xtest/test-nvme_options.py27
-rwxr-xr-xtest/test-udev.py644
12 files changed, 986 insertions, 28 deletions
diff --git a/test/NBFT b/test/NBFT
new file mode 100644
index 0000000..2dea936
--- /dev/null
+++ b/test/NBFT
Binary files differ
diff --git a/test/NBFT-Empty b/test/NBFT-Empty
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/test/NBFT-Empty
diff --git a/test/meson.build b/test/meson.build
index 251140c..086a3e8 100644
--- a/test/meson.build
+++ b/test/meson.build
@@ -7,6 +7,7 @@
# Authors: Martin Belanger <Martin.Belanger@dell.com>
#
+srce_dir = meson.current_source_dir()
test_env = environment({'MALLOC_PERTURB_': '0'})
libnvme_location = '?'
@@ -22,6 +23,9 @@ if get_option('libnvme-sel') == 'pre-installed'
rr = run_command(python3, '-c', 'import libnvme; print(f"{libnvme.__path__[0]}")', check: false, env: test_env)
if rr.returncode() == 0
libnvme_location = rr.stdout().strip()
+ pythonpath = fs.parent(libnvme_location)
+ test_env.prepend('PYTHONPATH', pythonpath) # Look in standard location first
+ test_env.append('PYTHONPATH', PYTHONPATH) # Look in the build directory second
endif
endif
@@ -40,7 +44,6 @@ endif
if libnvme_location == '?'
warning('Missing runtime package needed to run the tests: python3-libnvme.')
else
- message('\n\n\u001b[32m\u001b[1mNOTE: Tests will be using @0@\u001b[0m\n'.format(libnvme_location))
#---------------------------------------------------------------------------
# pylint and pyflakes
if modules_to_lint.length() != 0
@@ -53,7 +56,7 @@ else
endif
endif
- rcfile = meson.current_source_dir() / 'pylint.rc'
+ rcfile = srce_dir / 'pylint.rc'
if pylint.found()
test('pylint', pylint, args: ['--rcfile=' + rcfile] + modules_to_lint, env: test_env)
@@ -91,32 +94,37 @@ else
#---------------------------------------------------------------------------
# Unit tests
things_to_test = [
- ['Test Configuration', 'test-config.py', []],
- ['Test Controller', 'test-controller.py', ['pyfakefs']],
- ['Test GTimer', 'test-gtimer.py', []],
- ['Test iputil', 'test-iputil.py', []],
- ['Test KernelVersion', 'test-version.py', []],
- ['Test log', 'test-log.py', ['pyfakefs']],
- ['Test NvmeOptions', 'test-nvme_options.py', ['pyfakefs']],
- ['Test Service', 'test-service.py', ['pyfakefs']],
- ['Test TID', 'test-transport_id.py', []],
- ['Test Udev', 'test-udev.py', []],
- ['Test timeparse', 'test-timeparse.py', []],
+ ['Test Configuration', [], [srce_dir / 'test-config.py', ]],
+ ['Test Controller', ['pyfakefs'], [srce_dir / 'test-controller.py', ]],
+ ['Test GTimer', [], [srce_dir / 'test-gtimer.py', ]],
+ ['Test iputil', [], [srce_dir / 'test-iputil.py', ]],
+ ['Test KernelVersion', [], [srce_dir / 'test-version.py', ]],
+ ['Test log', ['pyfakefs'], [srce_dir / 'test-log.py', ]],
+ ['Test NBFT', [], [srce_dir / 'test-nbft.py', ]],
+ ['Test NbftConf', [], [srce_dir / 'test-nbft_conf.py', ]],
+ ['Test NvmeOptions', ['pyfakefs'], [srce_dir / 'test-nvme_options.py', ]],
+ ['Test Service', ['pyfakefs'], [srce_dir / 'test-service.py', ]],
+ ['Test TID', [], [srce_dir / 'test-transport_id.py', ]],
+ ['Test defs.py', [], [srce_dir / 'test-defs.py', ]],
+ ['Test gutil.py', [], [srce_dir / 'test-gutil.py', ]],
+ ['Test Udev', [], [srce_dir / 'test-udev.py', ]],
+ ['Test timeparse', [], [srce_dir / 'test-timeparse.py', ]],
]
# The Avahi test requires the Avahi and the Dbus daemons to be running.
if want_avahi_test
- things_to_test += [['Test Avahi', 'test-avahi.py', []]]
+ things_to_test += [['Test Avahi', [], [srce_dir / 'test-avahi.py']]]
else
warning('Skip Avahi Test due to missing dependencies')
endif
foreach thing: things_to_test
msg = thing[0]
+ deps = thing[1]
+ args = thing[2]
# Check whether all dependencies can be found
missing_deps = []
- deps = thing[2]
foreach dep : deps
rr = run_command(python3, '-c', 'import @0@'.format(dep), check: false)
if rr.returncode() != 0
@@ -126,8 +134,7 @@ else
if missing_deps.length() == 0
# Allow the test to run if all dependencies are available
- script = meson.current_source_dir() / thing[1]
- test(msg, python3, args: script, env: test_env)
+ test(msg, python3, args: args, env: test_env)
else
warning('"@0@" requires python module "@1@"'.format(msg, missing_deps))
endif
@@ -138,15 +145,15 @@ endif
#-------------------------------------------------------------------------------
# Make sure code complies with minimum Python version requirement.
tools = [
- meson.current_source_dir() / '../doc',
- meson.current_source_dir() / '../utils',
+ srce_dir / '../doc',
+ srce_dir / '../utils',
]
vermin = find_program('vermin', required: false)
if vermin.found()
if modules_to_lint.length() != 0
- test('vermin code', vermin, args: ['--config-file', meson.current_source_dir() / 'vermin.conf'] + modules_to_lint, env: test_env)
+ test('vermin code', vermin, args: ['--config-file', srce_dir / 'vermin.conf'] + modules_to_lint, env: test_env)
endif
- test('vermin tools', vermin, args: ['--config-file', meson.current_source_dir() / 'vermin-tools.conf'] + tools, env: test_env)
+ test('vermin tools', vermin, args: ['--config-file', srce_dir / 'vermin-tools.conf'] + tools, env: test_env)
else
warning('Skiping some of the tests because "vermin" is missing.')
endif
diff --git a/test/test-avahi.py b/test/test-avahi.py
index 3529104..1081947 100755
--- a/test/test-avahi.py
+++ b/test/test-avahi.py
@@ -36,6 +36,26 @@ class Test(unittest.TestCase):
srv.kill()
self.assertEqual(srv.info(), {'avahi wake up timer': 'None', 'service types': [], 'services': {}})
+ def test__txt2dict(self):
+ txt = [
+ list('NqN=Starfleet'.encode('utf-8')),
+ list('p=tcp'.encode('utf-8')),
+ ]
+ self.assertEqual(avahi._txt2dict(txt), {'nqn': 'Starfleet', 'p': 'tcp'})
+
+ txt = [
+ list('Nqn=Starfleet'.encode('utf-8')),
+ list('p='.encode('utf-8')), # Try with a missing value for p
+ list('blah'.encode('utf-8')), # Missing '='
+ list('='.encode('utf-8')), # Just '='
+ ]
+ self.assertEqual(avahi._txt2dict(txt), {'nqn': 'Starfleet', 'p': ''})
+
+ txt = [
+ [1000, ord('='), 123456], # Try with non printable characters
+ ]
+ self.assertEqual(avahi._txt2dict(txt), {})
+
if __name__ == '__main__':
unittest.main()
diff --git a/test/test-config.py b/test/test-config.py
index 1480fc3..1ecae93 100755
--- a/test/test-config.py
+++ b/test/test-config.py
@@ -157,6 +157,11 @@ class StasProcessConfUnitTest(unittest.TestCase):
self.assertRaises(KeyError, service_conf.get_option, 'Babylon', 5)
+ def test__parse_single_val(self):
+ self.assertEqual(conf._parse_single_val('hello'), 'hello')
+ self.assertIsNone(conf._parse_single_val(None))
+ self.assertEqual(conf._parse_single_val(['hello', 'goodbye']), 'goodbye')
+
class StasSysConfUnitTest(unittest.TestCase):
'''Sys config unit tests'''
diff --git a/test/test-defs.py b/test/test-defs.py
new file mode 100755
index 0000000..3f8b02b
--- /dev/null
+++ b/test/test-defs.py
@@ -0,0 +1,59 @@
+#!/usr/bin/python3
+import os
+import sys
+import unittest
+from unittest import mock
+
+
+class MockLibnvmeTestCase(unittest.TestCase):
+ '''Testing defs.py by mocking the libnvme package'''
+
+ def test_libnvme_version(self):
+ # For unknown reasons, this test does
+ # not work when run from GitHub Actions.
+ if not os.getenv('GITHUB_ACTIONS'):
+ from staslib import defs
+
+ libnvme_ver = defs.LIBNVME_VERSION
+ self.assertEqual(libnvme_ver, '?.?')
+
+ @classmethod
+ def setUpClass(cls): # called once before all the tests
+ # define what to patch sys.modules with
+ cls._modules_patcher = mock.patch.dict(sys.modules, {'libnvme': mock.Mock()})
+
+ # actually patch it
+ cls._modules_patcher.start()
+
+ # make the package globally visible and import it,
+ # just like if you have imported it in a usual way
+ # placing import statement at the top of the file,
+ # but relying on a patched dependency
+ global libnvme
+ import libnvme
+
+ @classmethod # called once after all tests
+ def tearDownClass(cls):
+ # restore initial sys.modules state back
+ cls._modules_patcher.stop()
+
+
+class RealLibnvmeUnitTest(unittest.TestCase):
+ '''Testing defs.py with the real libnvme package'''
+
+ def test_libnvme_version(self):
+ try:
+ # We can't proceed with this test if the
+ # module libnvme is not installed.
+ import libnvme
+ except ModuleNotFoundError:
+ return
+
+ from staslib import defs
+
+ libnvme_ver = defs.LIBNVME_VERSION
+ self.assertNotEqual(libnvme_ver, '?.?')
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test/test-gutil.py b/test/test-gutil.py
new file mode 100755
index 0000000..3039867
--- /dev/null
+++ b/test/test-gutil.py
@@ -0,0 +1,33 @@
+#!/usr/bin/python3
+import unittest
+from staslib import gutil
+
+
+class GutilUnitTest(unittest.TestCase):
+ '''Run unit test for gutil.py'''
+
+ def _on_success(self, op_obj: gutil.AsyncTask, data):
+ op_obj.kill()
+
+ def _on_fail(self, op_obj: gutil.AsyncTask, err, fail_cnt):
+ op_obj.kill()
+
+ def _operation(self, data):
+ return data
+
+ def test_AsyncTask(self):
+ op = gutil.AsyncTask(self._on_success, self._on_fail, self._operation, 'hello')
+
+ self.assertIsInstance(str(op), str)
+ self.assertEqual(op.as_dict(), {'fail count': 0, 'completed': None, 'alive': True})
+
+ op.retry(10)
+ self.assertIsNotNone(op.as_dict().get('retry timer'))
+
+ errmsg = 'something scarry happened'
+ op._errmsg = errmsg
+ self.assertEqual(op.as_dict().get('error'), errmsg)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test/test-iputil.py b/test/test-iputil.py
index 8f364ad..3af734f 100755
--- a/test/test-iputil.py
+++ b/test/test-iputil.py
@@ -1,11 +1,12 @@
#!/usr/bin/python3
import json
import shutil
+import socket
import logging
import unittest
import ipaddress
import subprocess
-from staslib import iputil, log, trid
+from staslib import iputil, log, stas, trid
IP = shutil.which('ip')
@@ -39,6 +40,12 @@ class Test(unittest.TestCase):
self.assertEqual('', iputil.get_interface('255.255.255.255'))
+ def test_mac2iface(self):
+ for iface in self.ifaces:
+ address = iface.get('address', None)
+ if address:
+ self.assertEqual(iface['ifname'], iputil.mac2iface(address))
+
def test_remove_invalid_addresses(self):
good_tcp = trid.TID({'transport': 'tcp', 'traddr': '1.1.1.1', 'subsysnqn': '', 'trsvcid': '8009'})
bad_tcp = trid.TID({'transport': 'tcp', 'traddr': '555.555.555.555', 'subsysnqn': '', 'trsvcid': '8009'})
@@ -51,7 +58,7 @@ class Test(unittest.TestCase):
any_fc,
bad_trtype,
]
- l2 = iputil.remove_invalid_addresses(l1)
+ l2 = stas.remove_invalid_addresses(l1)
self.assertNotEqual(l1, l2)
@@ -61,6 +68,11 @@ class Test(unittest.TestCase):
self.assertNotIn(bad_tcp, l2)
self.assertNotIn(bad_trtype, l2)
+ def test__data_matches_ip(self):
+ self.assertFalse(iputil._data_matches_ip(None, None, None))
+ self.assertFalse(iputil._data_matches_ip(socket.AF_INET, None, None))
+ self.assertFalse(iputil._data_matches_ip(socket.AF_INET6, None, None))
+
if __name__ == "__main__":
unittest.main()
diff --git a/test/test-nbft.py b/test/test-nbft.py
new file mode 100755
index 0000000..cf62214
--- /dev/null
+++ b/test/test-nbft.py
@@ -0,0 +1,105 @@
+#!/usr/bin/python3
+import os
+import unittest
+from staslib import defs, nbft
+from libnvme import nvme
+from argparse import ArgumentParser
+
+TEST_DIR = os.path.dirname(__file__)
+NBFT_FILE = os.path.join(TEST_DIR, "NBFT")
+EMPTY_NBFT_FILE = os.path.join(TEST_DIR, "NBFT-Empty")
+NBFT_DATA = {
+ "discovery": [
+ {
+ "hfi_index": 0,
+ "nqn": "nqn.2014-08.org.nvmexpress.discovery",
+ "uri": "nvme+tcp://100.71.103.50:8009/",
+ }
+ ],
+ "hfi": [
+ {
+ "dhcp_override": True,
+ "dhcp_server_ipaddr": "100.71.245.254",
+ "gateway_ipaddr": "100.71.245.254",
+ "ip_origin": 82,
+ "ipaddr": "100.71.245.232",
+ "mac_addr": "b0:26:28:e8:7c:0e",
+ "pcidev": "0:40:0.0",
+ "primary_dns_ipaddr": "100.64.0.5",
+ "route_metric": 500,
+ "secondary_dns_ipaddr": "100.64.0.6",
+ "subnet_mask_prefix": 24,
+ "this_hfi_is_default_route": True,
+ "trtype": "tcp",
+ "vlan": 0,
+ }
+ ],
+ "host": {
+ "host_id_configured": True,
+ "host_nqn_configured": True,
+ "id": "44454c4c-3400-1036-8038-b2c04f313233",
+ "nqn": "nqn.1988-11.com.dell:PowerEdge.R760.1234567",
+ "primary_admin_host_flag": "not " "indicated",
+ },
+ "subsystem": [
+ {
+ "asqsz": 0,
+ "controller_id": 5,
+ "data_digest_required": False,
+ "hfi_indexes": [0],
+ "nid": "c82404ed9c15f53b8ccf0968002e0fca",
+ "nid_type": "nguid",
+ "nsid": 148,
+ "pdu_header_digest_required": False,
+ "subsys_nqn": "nqn.1988-11.com.dell:powerstore:00:2a64abf1c5b81F6C4549",
+ "subsys_port_id": 0,
+ "traddr": "100.71.103.48",
+ "trsvcid": "4420",
+ "trtype": "tcp",
+ },
+ {
+ "asqsz": 0,
+ "controller_id": 4166,
+ "data_digest_required": False,
+ "hfi_indexes": [0],
+ "nid": "c82404ed9c15f53b8ccf0968002e0fca",
+ "nid_type": "nguid",
+ "nsid": 148,
+ "pdu_header_digest_required": False,
+ "subsys_nqn": "nqn.1988-11.com.dell:powerstore:00:2a64abf1c5b81F6C4549",
+ "subsys_port_id": 0,
+ "traddr": "100.71.103.49",
+ "trsvcid": "4420",
+ "trtype": "tcp",
+ },
+ ],
+}
+
+
+class Test(unittest.TestCase):
+ """Unit tests for NBFT"""
+
+ def setUp(self):
+ # Depending on the version of libnvme installed
+ # we may or may not have access to NBFT support.
+ nbft_get = getattr(nvme, "nbft_get", None)
+ if defs.HAS_NBFT_SUPPORT:
+ self.expected_nbft = {
+ NBFT_FILE: NBFT_DATA,
+ EMPTY_NBFT_FILE: {},
+ }
+ else:
+ self.expected_nbft = {}
+
+ def test_dir_with_nbft_files(self):
+ """Make sure we get expected data when reading from binary NBFT file"""
+ actual_nbft = nbft.get_nbft_files(TEST_DIR)
+ self.assertEqual(actual_nbft, self.expected_nbft)
+
+ def test_dir_without_nbft_files(self):
+ actual_nbft = nbft.get_nbft_files("/tmp")
+ self.assertEqual(actual_nbft, {})
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/test/test-nbft_conf.py b/test/test-nbft_conf.py
new file mode 100755
index 0000000..85cb35d
--- /dev/null
+++ b/test/test-nbft_conf.py
@@ -0,0 +1,56 @@
+#!/usr/bin/python3
+import os
+import logging
+import unittest
+from staslib import conf
+
+TEST_DIR = os.path.dirname(__file__)
+EXPECTED_DCS = [
+ {
+ 'subsysnqn': 'nqn.2014-08.org.nvmexpress.discovery',
+ 'traddr': '100.71.103.50',
+ 'transport': 'tcp',
+ 'trsvcid': '8009',
+ }
+]
+EXPECTED_IOCS = [
+ {
+ 'data-digest': False,
+ 'hdr-digest': False,
+ 'subsysnqn': 'nqn.1988-11.com.dell:powerstore:00:2a64abf1c5b81F6C4549',
+ 'traddr': '100.71.103.48',
+ 'transport': 'tcp',
+ 'trsvcid': '4420',
+ },
+ {
+ 'data-digest': False,
+ 'hdr-digest': False,
+ 'subsysnqn': 'nqn.1988-11.com.dell:powerstore:00:2a64abf1c5b81F6C4549',
+ 'traddr': '100.71.103.49',
+ 'transport': 'tcp',
+ 'trsvcid': '4420',
+ },
+]
+
+
+class Test(unittest.TestCase):
+ """Unit tests for class NbftConf"""
+
+ def test_dir_with_nbft_files(self):
+ conf.NbftConf.destroy() # Make sure singleton does not exist
+ with self.assertLogs(logger=logging.getLogger(), level='DEBUG') as captured:
+ nbft_conf = conf.NbftConf(TEST_DIR)
+ self.assertNotEqual(-1, captured.records[0].getMessage().find("NBFT location(s):"))
+ self.assertEqual(nbft_conf.dcs, EXPECTED_DCS)
+ self.assertEqual(nbft_conf.iocs, EXPECTED_IOCS)
+
+ def test_dir_without_nbft_files(self):
+ conf.NbftConf.destroy() # Make sure singleton does not exist
+ with self.assertNoLogs(logger=logging.getLogger(), level='DEBUG'):
+ nbft_conf = conf.NbftConf('/tmp')
+ self.assertEqual(nbft_conf.dcs, [])
+ self.assertEqual(nbft_conf.iocs, [])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/test/test-nvme_options.py b/test/test-nvme_options.py
index 428f22a..adfba5f 100755
--- a/test/test-nvme_options.py
+++ b/test/test-nvme_options.py
@@ -6,6 +6,19 @@ from staslib import conf, log
from pyfakefs.fake_filesystem_unittest import TestCase
+class TestStandardNvmeFabricsFile(unittest.TestCase):
+ def test_regular_user(self):
+ conf.NvmeOptions.destroy() # Make sure singleton does not exist
+ if os.path.exists('/dev/nvme-fabrics'):
+ if os.geteuid() != 0:
+ with self.assertRaises(PermissionError):
+ nvme_options = conf.NvmeOptions()
+ else:
+ nvme_options = conf.NvmeOptions()
+ self.assertIsInstance(nvme_options.discovery_supp, bool)
+ self.assertIsInstance(nvme_options.host_iface_supp, bool)
+
+
class Test(TestCase):
"""Unit tests for class NvmeOptions"""
@@ -19,24 +32,30 @@ class Test(TestCase):
# No longer need self.tearDownPyfakefs()
pass
+ def test_file_missing(self):
+ self.assertFalse(os.path.exists("/dev/nvme-fabrics"))
+ conf.NvmeOptions.destroy() # Make sure singleton does not exist
+ nvme_options = conf.NvmeOptions()
+ self.assertIsInstance(nvme_options.discovery_supp, bool)
+ self.assertIsInstance(nvme_options.host_iface_supp, bool)
+
def test_fabrics_empty_file(self):
self.assertFalse(os.path.exists("/dev/nvme-fabrics"))
- # TODO: this is a bug
self.fs.create_file("/dev/nvme-fabrics")
self.assertTrue(os.path.exists('/dev/nvme-fabrics'))
+ conf.NvmeOptions.destroy() # Make sure singleton does not exist
nvme_options = conf.NvmeOptions()
self.assertIsInstance(nvme_options.discovery_supp, bool)
self.assertIsInstance(nvme_options.host_iface_supp, bool)
- del nvme_options
def test_fabrics_wrong_file(self):
self.assertFalse(os.path.exists("/dev/nvme-fabrics"))
self.fs.create_file("/dev/nvme-fabrics", contents="blah")
self.assertTrue(os.path.exists('/dev/nvme-fabrics'))
+ conf.NvmeOptions.destroy() # Make sure singleton does not exist
nvme_options = conf.NvmeOptions()
self.assertIsInstance(nvme_options.discovery_supp, bool)
self.assertIsInstance(nvme_options.host_iface_supp, bool)
- del nvme_options
def test_fabrics_correct_file(self):
self.assertFalse(os.path.exists("/dev/nvme-fabrics"))
@@ -44,6 +63,7 @@ class Test(TestCase):
'/dev/nvme-fabrics', contents='host_iface=%s,discovery,dhchap_secret=%s,dhchap_ctrl_secret=%s\n'
)
self.assertTrue(os.path.exists('/dev/nvme-fabrics'))
+ conf.NvmeOptions.destroy() # Make sure singleton does not exist
nvme_options = conf.NvmeOptions()
self.assertTrue(nvme_options.discovery_supp)
self.assertTrue(nvme_options.host_iface_supp)
@@ -54,7 +74,6 @@ class Test(TestCase):
{'discovery': True, 'host_iface': True, 'dhchap_secret': True, 'dhchap_ctrl_secret': True},
)
self.assertTrue(str(nvme_options).startswith("supported options:"))
- del nvme_options
if __name__ == "__main__":
diff --git a/test/test-udev.py b/test/test-udev.py
index 3798d6c..71e5f8a 100755
--- a/test/test-udev.py
+++ b/test/test-udev.py
@@ -1,6 +1,196 @@
#!/usr/bin/python3
+import json
+import shutil
+import logging
import unittest
-from staslib import udev
+import ipaddress
+import subprocess
+from staslib import defs, iputil, log, trid, udev
+
+IP = shutil.which('ip')
+
+TRADDR4 = '1.2.3.4'
+TRADDR4_REV = '4.3.2.1'
+TRADDR6 = 'FE80::aaaa:BBBB:cccc:dddd'
+TRADDR6_REV = 'fe80::DDDD:cccc:bbbb:AAAA'
+
+
+def traddr(family, reverse=False):
+ if reverse:
+ return TRADDR4_REV if family == 4 else TRADDR6_REV
+ return TRADDR4 if family == 4 else TRADDR6
+
+
+def get_tids_to_test(family, src_ip, ifname):
+ return [
+ (
+ 1,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ 'host-iface': ifname,
+ }
+ ),
+ True,
+ ),
+ (
+ 2,
+ trid.TID(
+ {
+ 'transport': 'blah',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ 'host-iface': ifname,
+ }
+ ),
+ False,
+ ),
+ (
+ 3,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family, reverse=True),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ 'host-iface': ifname,
+ }
+ ),
+ False,
+ ),
+ (
+ 4,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8010',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ 'host-iface': ifname,
+ }
+ ),
+ False,
+ ),
+ (
+ 5,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '255.255.255.255',
+ 'host-iface': ifname,
+ }
+ ),
+ False,
+ ),
+ (
+ 6,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ 'host-iface': 'blah',
+ }
+ ),
+ False,
+ ),
+ (
+ 7,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'bob',
+ 'host-traddr': src_ip,
+ 'host-iface': ifname,
+ }
+ ),
+ False,
+ ),
+ (
+ 8,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-iface': ifname,
+ }
+ ),
+ True,
+ ),
+ (
+ 9,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ }
+ ),
+ True,
+ ),
+ (
+ 10,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ }
+ ),
+ True,
+ ),
+ (
+ 11,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ip,
+ 'host-iface': ifname,
+ }
+ ),
+ True,
+ ),
+ (
+ 12,
+ trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(family),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-iface': ifname,
+ }
+ ),
+ True,
+ ),
+ ]
+
+
+class DummyDevice:
+ ...
class Test(unittest.TestCase):
@@ -9,6 +199,30 @@ class Test(unittest.TestCase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
+ def setUp(self):
+ log.init(syslog=False)
+ self.logger = logging.getLogger()
+ self.logger.setLevel(logging.INFO)
+
+ # Retrieve the list of Interfaces and all the associated IP addresses
+ # using standard bash utility (ip address).
+ try:
+ cmd = [IP, '-j', 'address', 'show']
+ p = subprocess.run(cmd, stdout=subprocess.PIPE, check=True)
+ ifaces = json.loads(p.stdout.decode().strip())
+ except subprocess.CalledProcessError:
+ ifaces = []
+
+ self.ifaces = {}
+ for iface in ifaces:
+ addr_info = iface.get('addr_info')
+ if addr_info:
+ ifname = iface['ifname']
+ self.ifaces[ifname] = {}
+ for info in addr_info:
+ family = 4 if info['family'] == 'inet' else 6
+ self.ifaces[ifname].setdefault(family, []).append(info['local'])
+
@classmethod
def tearDownClass(cls):
'''Release resources'''
@@ -36,6 +250,434 @@ class Test(unittest.TestCase):
bogus = udev.UDEV.get_key_from_attr(device, 'bogus', 'BOGUS', '\n')
self.assertEqual(bogus, '')
+ def test_is_dc_device(self):
+ device = DummyDevice()
+ device.children = ['vera', 'Chuck', 'Dave']
+ device.attributes = {}
+
+ self.assertFalse(udev.UDEV.is_dc_device(device))
+
+ device.attributes = {'subsysnqn': defs.WELL_KNOWN_DISC_NQN.encode('utf-8')}
+ self.assertTrue(udev.UDEV.is_dc_device(device))
+
+ device.attributes = {'cntrltype': 'discovery'.encode('utf-8')}
+ self.assertTrue(udev.UDEV.is_dc_device(device))
+
+ device.attributes = {}
+ device.children = []
+ self.assertTrue(udev.UDEV.is_dc_device(device))
+
+ def test_is_ioc_device(self):
+ device = DummyDevice()
+ device.children = []
+ device.attributes = {}
+
+ self.assertFalse(udev.UDEV.is_ioc_device(device))
+
+ device.attributes = {'cntrltype': 'io'.encode('utf-8')}
+ self.assertTrue(udev.UDEV.is_ioc_device(device))
+
+ device.attributes = {}
+ device.children = ['vera', 'Chuck', 'Dave']
+ self.assertTrue(udev.UDEV.is_ioc_device(device))
+
+ def test__cid_matches_tid(self):
+ for ifname, addrs in self.ifaces.items():
+ ##############################################
+ # IPV4
+
+ ipv4_addrs = addrs.get(4, [])
+ for src_ipv4 in ipv4_addrs:
+ cid = {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ipv4,
+ 'host-iface': ifname,
+ 'src-addr': src_ipv4,
+ }
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ipv4,
+ 'host-iface': ifname,
+ 'src-addr': '', # Legacy
+ }
+ for case_id, tid, match in get_tids_to_test(4, src_ipv4, ifname):
+ self.assertEqual(match, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case {case_id} failed')
+ if case_id != 8:
+ self.assertEqual(
+ match, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case {case_id} failed'
+ )
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '',
+ 'host-iface': '',
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '1.1.1.1',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case A4.1 failed')
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '',
+ 'host-iface': ifname,
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case A4.2 failed')
+ self.assertEqual(
+ True, udev.UDEV._cid_matches_tcp_tid_legacy(tid, cid_legacy), msg=f'Legacy Test Case A4.3 failed'
+ )
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ipv4,
+ 'host-iface': '',
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '1.1.1.1',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case B4 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-iface': 'blah',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case C4 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-iface': ifname,
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case D4 failed')
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '',
+ 'host-iface': ifname,
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '1.1.1.1',
+ 'host-iface': 'blah',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case E4 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '1.1.1.1',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case F4 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(4),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': ipv4_addrs[0],
+ }
+ )
+ match = len(ipv4_addrs) == 1 and iputil.get_ipaddress_obj(
+ ipv4_addrs[0], ipv4_mapped_convert=True
+ ) == iputil.get_ipaddress_obj(tid.host_traddr, ipv4_mapped_convert=True)
+ self.assertEqual(match, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case G4 failed')
+
+ ##############################################
+ # IPV6
+
+ ipv6_addrs = addrs.get(6, [])
+ for src_ipv6 in ipv6_addrs:
+ cid = {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ipv6,
+ 'host-iface': ifname,
+ 'src-addr': src_ipv6,
+ }
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ipv6,
+ 'host-iface': ifname,
+ 'src-addr': '', # Legacy
+ }
+ for case_id, tid, match in get_tids_to_test(6, src_ipv6, ifname):
+ self.assertEqual(match, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case {case_id} failed')
+ self.assertEqual(
+ match, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case {case_id} failed'
+ )
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '',
+ 'host-iface': '',
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'AAAA::FFFF',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case A6.1 failed')
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '',
+ 'host-iface': ifname,
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case A6.2 failed')
+ self.assertEqual(
+ True, udev.UDEV._cid_matches_tcp_tid_legacy(tid, cid_legacy), msg=f'Legacy Test Case A6.3 failed'
+ )
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': src_ipv6,
+ 'host-iface': '',
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'AAAA::FFFF',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case B6 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-iface': 'blah',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case C6 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-iface': ifname,
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case D6 failed')
+
+ cid_legacy = {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '',
+ 'host-iface': ifname,
+ 'src-addr': '', # Legacy
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'AAA::BBBB',
+ 'host-iface': 'blah',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case E6 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'AAA::BBB',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case F6 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'tcp',
+ 'traddr': traddr(6),
+ 'trsvcid': '8009',
+ 'subsysnqn': 'hello',
+ 'host-traddr': ipv6_addrs[0],
+ }
+ )
+ match = len(ipv6_addrs) == 1 and iputil.get_ipaddress_obj(
+ ipv6_addrs[0], ipv4_mapped_convert=True
+ ) == iputil.get_ipaddress_obj(tid.host_traddr, ipv4_mapped_convert=True)
+ self.assertEqual(match, udev.UDEV._cid_matches_tid(tid, cid_legacy), msg=f'Legacy Test Case G6 failed')
+
+ ##############################################
+ # FC
+ cid = {
+ 'transport': 'fc',
+ 'traddr': 'ABC',
+ 'trsvcid': '',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'AAA::BBBB',
+ 'host-iface': '',
+ 'src-addr': '',
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'fc',
+ 'traddr': 'ABC',
+ 'trsvcid': '',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'AAA::BBBB',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case FC-1 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'fc',
+ 'traddr': 'ABC',
+ 'trsvcid': '',
+ 'subsysnqn': 'hello',
+ 'host-traddr': 'BBBB::AAA',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case FC-2 failed')
+
+ ##############################################
+ # RDMA
+ cid = {
+ 'transport': 'rdma',
+ 'traddr': '2.3.4.5',
+ 'trsvcid': '4444',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '5.4.3.2',
+ 'host-iface': '',
+ 'src-addr': '',
+ }
+ tid = trid.TID(
+ {
+ 'transport': 'rdma',
+ 'traddr': '2.3.4.5',
+ 'trsvcid': '4444',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '5.4.3.2',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case RDMA-1 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'rdma',
+ 'traddr': '2.3.4.5',
+ 'trsvcid': '4444',
+ 'subsysnqn': 'hello',
+ 'host-traddr': '5.5.6.6',
+ }
+ )
+ self.assertEqual(False, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case RDMA-2 failed')
+
+ tid = trid.TID(
+ {
+ 'transport': 'rdma',
+ 'traddr': '2.3.4.5',
+ 'trsvcid': '4444',
+ 'subsysnqn': 'hello',
+ }
+ )
+ self.assertEqual(True, udev.UDEV._cid_matches_tid(tid, cid), msg=f'Test Case RDMA-3 failed')
+
if __name__ == '__main__':
unittest.main()