nvme-stas/test/test-controller.py

#!/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()