Merging upstream version 2.3~rc1.
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
parent
972d2d9aa2
commit
ca2ec6771a
GPG key ID:
FBB4F0E80A80222F
37 changed files with 1946 additions and 407 deletions
BIN
test/NBFT
Normal file
BIN
test/NBFT
Normal file
Binary file not shown.
0
test/NBFT-Empty
Normal file
0
test/NBFT-Empty
Normal file
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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'''
|
||||
|
|
|
|||
59
test/test-defs.py
Executable file
59
test/test-defs.py
Executable file
|
|
@ -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()
|
||||
33
test/test-gutil.py
Executable file
33
test/test-gutil.py
Executable file
|
|
@ -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()
|
||||
|
|
@ -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()
|
||||
|
|
|
|||
105
test/test-nbft.py
Executable file
105
test/test-nbft.py
Executable file
|
|
@ -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()
|
||||
56
test/test-nbft_conf.py
Executable file
56
test/test-nbft_conf.py
Executable file
|
|
@ -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()
|
||||
|
|
@ -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_fabrics_empty_file(self):
|
||||
def test_file_missing(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)
|
||||
|
||||
def test_fabrics_empty_file(self):
|
||||
self.assertFalse(os.path.exists("/dev/nvme-fabrics"))
|
||||
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__":
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue