#!/usr/bin/python3
import logging
import unittest
from staslib import conf, ctrl, timeparse, trid
from pyfakefs.fake_filesystem_unittest import TestCase
class TestController(ctrl.Controller):
def _find_existing_connection(self):
pass
def _on_aen(self, aen: int):
pass
def _on_nvme_event(self, nvme_event):
pass
def reload_hdlr(self):
pass
class TestDc(ctrl.Dc):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._connected = True
class Ctrl:
def __init__(this):
this.name = 'nvme666'
def connected(this):
return self._connected
def disconnect(this):
pass
self._ctrl = Ctrl()
def _find_existing_connection(self):
pass
def _on_aen(self, aen: int):
pass
def _on_nvme_event(self, nvme_event):
pass
def reload_hdlr(self):
pass
def set_connected(self, value):
self._connected = value
def connected(self):
return self._connected
class TestStaf:
def is_avahi_reported(self, tid):
return False
def controller_unresponsive(self, tid):
pass
@property
def tron(self):
return True
stafd_conf_1 = '''
[Global]
tron=false
hdr-digest=false
data-digest=false
kato=30
queue-size=128
reconnect-delay=10
ctrl-loss-tmo=600
disable-sqflow=false
ignore-iface=false
ip-family=ipv4+ipv6
pleo=enabled
[Service Discovery]
zeroconf=enabled
[Discovery controller connection management]
persistent-connections=true
zeroconf-connections-persistence=10 seconds
'''
stafd_conf_2 = '''
[Discovery controller connection management]
zeroconf-connections-persistence=-1
'''
class Test(TestCase):
'''Unit tests for class Controller'''
def setUp(self):
self.setUpPyfakefs()
self.fs.create_file(
'/etc/nvme/hostnqn', contents='nqn.2014-08.org.nvmexpress:uuid:01234567-0123-0123-0123-0123456789ab\n'
)
self.fs.create_file('/etc/nvme/hostid', contents='01234567-89ab-cdef-0123-456789abcdef\n')
self.fs.create_file(
'/dev/nvme-fabrics',
contents='instance=-1,cntlid=-1,transport=%s,traddr=%s,trsvcid=%s,nqn=%s,queue_size=%d,nr_io_queues=%d,reconnect_delay=%d,ctrl_loss_tmo=%d,keep_alive_tmo=%d,hostnqn=%s,host_traddr=%s,host_iface=%s,hostid=%s,disable_sqflow,hdr_digest,data_digest,nr_write_queues=%d,nr_poll_queues=%d,tos=%d,fast_io_fail_tmo=%d,discovery,dhchap_secret=%s,dhchap_ctrl_secret=%s\n',
)
self.NVME_TID = trid.TID(
{
'transport': 'tcp',
'traddr': '10.10.10.10',
'subsysnqn': 'nqn.1988-11.com.dell:SFSS:2:20220208134025e8',
'trsvcid': '8009',
'host-traddr': '1.2.3.4',
'host-iface': 'wlp0s20f3',
'host-nqn': 'nqn.1988-11.com.dell:poweredge:1234',
}
)
default_conf = {
('Global', 'tron'): False,
('Global', 'hdr-digest'): False,
('Global', 'data-digest'): False,
('Global', 'kato'): None, # None to let the driver decide the default
('Global', 'queue-size'): None, # None to let the driver decide the default
('Global', 'reconnect-delay'): None, # None to let the driver decide the default
('Global', 'ctrl-loss-tmo'): None, # None to let the driver decide the default
('Global', 'disable-sqflow'): None, # None to let the driver decide the default
('Global', 'persistent-connections'): True,
('Discovery controller connection management', 'persistent-connections'): True,
('Discovery controller connection management', 'zeroconf-connections-persistence'): timeparse.timeparse(
'72hours'
),
('Global', 'ignore-iface'): False,
('Global', 'ip-family'): (4, 6),
('Global', 'pleo'): True,
('Service Discovery', 'zeroconf'): True,
('Controllers', 'controller'): list(),
('Controllers', 'exclude'): list(),
}
self.stafd_conf_file1 = '/etc/stas/stafd1.conf'
self.fs.create_file(self.stafd_conf_file1, contents=stafd_conf_1)
self.stafd_conf_file2 = '/etc/stas/stafd2.conf'
self.fs.create_file(self.stafd_conf_file2, contents=stafd_conf_2)
self.svcconf = conf.SvcConf(default_conf=default_conf)
self.svcconf.set_conf_file(self.stafd_conf_file1)
def tearDown(self):
pass
def test_cannot_instantiate_concrete_classes_if_abstract_method_are_not_implemented(self):
# Make sure we can't instantiate the ABC directly (Abstract Base Class).
class Controller(ctrl.Controller):
pass
self.assertRaises(TypeError, lambda: ctrl.Controller(tid=self.NVME_TID))
def test_get_device(self):
controller = TestController(tid=self.NVME_TID, service=TestStaf())
self.assertEqual(controller._connect_attempts, 0)
controller._try_to_connect()
self.assertEqual(controller._connect_attempts, 1)
self.assertEqual(
controller.id, "(tcp, 10.10.10.10, 8009, nqn.1988-11.com.dell:SFSS:2:20220208134025e8, wlp0s20f3, 1.2.3.4)"
)
# raise Exception(controller._connect_op)
self.assertEqual(
str(controller.tid),
"(tcp, 10.10.10.10, 8009, nqn.1988-11.com.dell:SFSS:2:20220208134025e8, wlp0s20f3, 1.2.3.4)",
)
self.assertEqual(controller.device, 'nvme?')
self.assertEqual(
controller.controller_id_dict(),
{
'transport': 'tcp',
'traddr': '10.10.10.10',
'trsvcid': '8009',
'host-traddr': '1.2.3.4',
'host-iface': 'wlp0s20f3',
'subsysnqn': 'nqn.1988-11.com.dell:SFSS:2:20220208134025e8',
'device': 'nvme?',
'host-nqn': 'nqn.1988-11.com.dell:poweredge:1234',
},
)
self.assertEqual(
controller.info(),
{
'transport': 'tcp',
'traddr': '10.10.10.10',
'subsysnqn': 'nqn.1988-11.com.dell:SFSS:2:20220208134025e8',
'trsvcid': '8009',
'host-traddr': '1.2.3.4',
'host-iface': 'wlp0s20f3',
'host-nqn': 'nqn.1988-11.com.dell:poweredge:1234',
'device': 'nvme?',
'connect attempts': '1',
'retry connect timer': '60.0s [off]',
'connect operation': "{'fail count': 0, 'completed': False, 'alive': True}",
},
)
self.assertEqual(
controller.details(),
{
'dctype': '',
'cntrltype': '',
'connected': 'False',
'transport': 'tcp',
'traddr': '10.10.10.10',
'trsvcid': '8009',
'host-traddr': '1.2.3.4',
'host-iface': 'wlp0s20f3',
'host-nqn': 'nqn.1988-11.com.dell:poweredge:1234',
'subsysnqn': 'nqn.1988-11.com.dell:SFSS:2:20220208134025e8',
'device': 'nvme?',
'connect attempts': '1',
'retry connect timer': '60.0s [off]',
'hostid': '',
'hostnqn': '',
'model': '',
'serial': '',
'connect operation': "{'fail count': 0, 'completed': False, 'alive': True}",
},
)
# print(controller._connect_op)
self.assertEqual(controller.cancel(), None)
self.assertEqual(controller.kill(), None)
self.assertIsNone(controller.disconnect(lambda *args: None, True))
def test_connect(self):
controller = TestController(tid=self.NVME_TID, service=TestStaf())
self.assertEqual(controller._connect_attempts, 0)
controller._find_existing_connection = lambda: None
with self.assertLogs(logger=logging.getLogger(), level='DEBUG') as captured:
controller._try_to_connect()
self.assertTrue(len(captured.records) > 0)
self.assertTrue(
captured.records[0]
.getMessage()
.startswith(
"Controller._do_connect() - (tcp, 10.10.10.10, 8009, nqn.1988-11.com.dell:SFSS:2:20220208134025e8, wlp0s20f3, 1.2.3.4) Connecting to nvme control with cfg={"
)
)
self.assertEqual(controller._connect_attempts, 1)
def test_dlp_supp_opts_as_string(self):
dlp_supp_opts = 0x7
opts = ctrl.dlp_supp_opts_as_string(dlp_supp_opts)
self.assertEqual(['EXTDLPES', 'PLEOS', 'ALLSUBES'], opts)
def test_ncc(self):
dlpe = {'eflags': '4'}
ncc = ctrl.get_ncc(ctrl.get_eflags(dlpe))
self.assertTrue(ncc)
dlpe = {}
ncc = ctrl.get_ncc(ctrl.get_eflags(dlpe))
self.assertFalse(ncc)
def test_dc(self):
self.svcconf.set_conf_file(self.stafd_conf_file1)
controller = TestDc(TestStaf(), tid=self.NVME_TID)
controller.set_connected(True)
controller.origin = 'discovered'
with self.assertLogs(logger=logging.getLogger(), level='DEBUG') as captured:
controller.origin = 'blah'
self.assertEqual(len(captured.records), 1)
self.assertNotEqual(-1, captured.records[0].getMessage().find("Trying to set invalid origin to blah"))
controller.set_connected(False)
with self.assertLogs(logger=logging.getLogger(), level='DEBUG') as captured:
controller.origin = 'discovered'
self.assertEqual(len(captured.records), 1)
self.assertNotEqual(
-1, captured.records[0].getMessage().find("Controller is not responding. Will be removed by")
)
self.svcconf.set_conf_file(self.stafd_conf_file2)
with self.assertLogs(logger=logging.getLogger(), level='DEBUG') as captured:
controller.origin = 'discovered'
self.assertEqual(len(captured.records), 1)
self.assertNotEqual(-1, captured.records[0].getMessage().find("Controller not responding. Retrying..."))
controller.set_connected(True)
with self.assertLogs(logger=logging.getLogger(), level='DEBUG') as captured:
controller.disconnect(lambda *args: None, keep_connection=False)
self.assertEqual(len(captured.records), 2)
self.assertNotEqual(-1, captured.records[0].getMessage().find("nvme666: keep_connection=False"))
self.assertNotEqual(-1, captured.records[1].getMessage().find("nvme666 - Disconnect initiated"))
# def test_disconnect(self):
if __name__ == '__main__':
unittest.main()