diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/NBFT | bin | 0 -> 1017 bytes | |||
| -rw-r--r-- | test/NBFT-Empty | 0 | ||||
| -rw-r--r-- | test/meson.build | 49 | ||||
| -rwxr-xr-x | test/test-avahi.py | 20 | ||||
| -rwxr-xr-x | test/test-config.py | 5 | ||||
| -rwxr-xr-x | test/test-defs.py | 59 | ||||
| -rwxr-xr-x | test/test-gutil.py | 33 | ||||
| -rwxr-xr-x | test/test-iputil.py | 16 | ||||
| -rwxr-xr-x | test/test-nbft.py | 105 | ||||
| -rwxr-xr-x | test/test-nbft_conf.py | 56 | ||||
| -rwxr-xr-x | test/test-nvme_options.py | 27 | ||||
| -rwxr-xr-x | test/test-udev.py | 644 |
12 files changed, 986 insertions, 28 deletions
diff --git a/test/NBFT b/test/NBFT Binary files differnew file mode 100644 index 0000000..2dea936 --- /dev/null +++ b/test/NBFT 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() |