Merge pull request #481 from martin-belanger/more-unit-tests

test: expand unit-test coverage across four test files

Author: martin-belanger

coverage.sh.in

@@ -681,21 +681,22 @@ sudo chown -R "${PRIMARY_USR}":"${PRIMARY_GRP}" subprojects/nvme-cli/libnvme/lib
 # Run unit tests
 TEST_DIR=$( realpath ../test )
 run_unit_test ${TEST_DIR}/test-avahi.py
-run_unit_test ${TEST_DIR}/test-avahi.py
 run_unit_test ${TEST_DIR}/test-config.py
 run_unit_test ${TEST_DIR}/test-controller.py
+run_unit_test ${TEST_DIR}/test-defs.py
 run_unit_test ${TEST_DIR}/test-gtimer.py
+run_unit_test ${TEST_DIR}/test-gutil.py
 run_unit_test ${TEST_DIR}/test-iputil.py
 run_unit_test ${TEST_DIR}/test-log.py
-run_unit_test ${TEST_DIR}/test-nbft.py
 run_unit_test ${TEST_DIR}/test-nbft_conf.py
+run_unit_test ${TEST_DIR}/test-nbft.py
 run_unit_test sudo ${TEST_DIR}/test-nvme_options.py  # Test both with super user...
 run_unit_test ${TEST_DIR}/test-nvme_options.py       # ... and with regular user
 run_unit_test ${TEST_DIR}/test-service.py
+run_unit_test ${TEST_DIR}/test-stasadm.py
+run_unit_test ${TEST_DIR}/test-stas.py
 run_unit_test ${TEST_DIR}/test-timeparse.py
 run_unit_test ${TEST_DIR}/test-transport_id.py
-run_unit_test ${TEST_DIR}/test-defs.py
-run_unit_test ${TEST_DIR}/test-gutil.py
 run_unit_test ${TEST_DIR}/test-udev.py
 run_unit_test ${TEST_DIR}/test-version.py
 

test/test-config.py

@@ -264,5 +264,93 @@ def test_config_missing_file(self):
         self.assertIsNone(system_conf.hostsymname)
 
 
+class TestParseController(unittest.TestCase):
+    '''Unit tests for conf._parse_controller() — a pure function.'''
+
+    def test_empty_string_returns_empty_dict(self):
+        self.assertEqual(conf._parse_controller(''), {})
+
+    def test_malformed_token_with_no_equals_is_silently_skipped(self):
+        # Token without '=' causes ValueError in unpacking → silently ignored
+        result = conf._parse_controller('noequalsign')
+        self.assertEqual(result, {})
+
+    def test_token_with_extra_equals_is_silently_skipped(self):
+        # 'key=val=extra' splits into 3 parts → ValueError → silently ignored
+        result = conf._parse_controller('key=val=extra')
+        self.assertEqual(result, {})
+
+    def test_mixed_valid_and_malformed_tokens(self):
+        result = conf._parse_controller('transport=tcp;noequalsign;traddr=10.10.10.10')
+        self.assertEqual(result, {'transport': 'tcp', 'traddr': '10.10.10.10'})
+
+
+class TestSvcConfEdgeCases(unittest.TestCase):
+    '''Edge-case tests for SvcConf validation: out-of-range values, invalid
+    sections/options.  These tests rely on the SvcConf singleton having been
+    initialised with a default_conf (which happens in StasProcessConfUnitTest),
+    so they are defined after that class.
+    '''
+
+    FNAME_OOR = '/tmp/stas-test-svc-oor.conf'
+    FNAME_BADSEC = '/tmp/stas-test-svc-badsec.conf'
+    FNAME_BADOPT = '/tmp/stas-test-svc-badopt.conf'
+    FNAME_VALID = '/tmp/stas-test-svc-valid.conf'
+
+    @classmethod
+    def setUpClass(cls):
+        with open(cls.FNAME_OOR, 'w') as f:
+            f.write('[Global]\nqueue-size=5\n')  # 5 is below the valid range [16, 1024]
+        with open(cls.FNAME_BADSEC, 'w') as f:
+            f.write('[BadSection]\nfoo=bar\n')
+        with open(cls.FNAME_BADOPT, 'w') as f:
+            f.write('[Global]\nbad-option=something\n')
+        with open(cls.FNAME_VALID, 'w') as f:
+            f.write('[Global]\nip-family=ipv4+ipv6\n')
+
+    @classmethod
+    def tearDownClass(cls):
+        for fname in (cls.FNAME_OOR, cls.FNAME_BADSEC, cls.FNAME_BADOPT, cls.FNAME_VALID):
+            if os.path.exists(fname):
+                os.remove(fname)
+
+    def setUp(self):
+        conf.SvcConf().set_conf_file(self.FNAME_VALID)
+
+    def test_queue_size_out_of_range_falls_back_to_none(self):
+        conf.SvcConf().set_conf_file(self.FNAME_OOR)
+        self.assertIsNone(conf.SvcConf().queue_size)
+
+    def test_invalid_section_logs_error(self):
+        with self.assertLogs(level='ERROR'):
+            conf.SvcConf().set_conf_file(self.FNAME_BADSEC)
+
+    def test_invalid_option_in_valid_section_logs_error(self):
+        with self.assertLogs(level='ERROR'):
+            conf.SvcConf().set_conf_file(self.FNAME_BADOPT)
+
+
+class TestSysConfNqnTooLong(unittest.TestCase):
+    '''Tests for SysConf NQN length validation (> 223 chars → sys.exit).'''
+
+    FNAME = '/tmp/stas-test-long-nqn.conf'
+
+    @classmethod
+    def setUpClass(cls):
+        long_nqn = 'nqn.' + 'a' * 220  # 224 chars — exceeds 223-char NVMe spec limit
+        with open(cls.FNAME, 'w') as f:
+            f.write(f'[Host]\nnqn={long_nqn}\n')
+
+    @classmethod
+    def tearDownClass(cls):
+        if os.path.exists(cls.FNAME):
+            os.remove(cls.FNAME)
+
+    def test_hostnqn_too_long_causes_exit(self):
+        system_conf = conf.SysConf()
+        system_conf.set_conf_file(self.FNAME)
+        self.assertRaises(SystemExit, lambda: system_conf.hostnqn)
+
+
 if __name__ == '__main__':
     unittest.main()

test/test-gtimer.py

@@ -38,6 +38,30 @@ def test_clear(self):
         self.assertEqual(tmr.time_remaining(), 0)
         self.assertEqual(str(tmr), '1.0s [0s]')
 
+    def test_callback_source_remove_clears_source(self):
+        from gi.repository import GLib
+
+        tmr = gutil.GTimer(interval_sec=1, user_cback=lambda: GLib.SOURCE_REMOVE)
+        tmr.start()
+        self.assertIsNotNone(tmr._source)
+        result = tmr._callback()
+        self.assertEqual(result, GLib.SOURCE_REMOVE)
+        # _callback() must clear _source when the user callback returns SOURCE_REMOVE
+        self.assertIsNone(tmr._source)
+
+    def test_restart_running_timer_reschedules_in_place(self):
+        from gi.repository import GLib
+
+        tmr = gutil.GTimer(interval_sec=10, user_cback=lambda: GLib.SOURCE_REMOVE)
+        tmr.start()
+        self.assertIsNotNone(tmr._source)
+        source_before = tmr._source
+        # start() on an already-running timer must reuse the existing GLib source
+        # (via set_ready_time) rather than creating a new one.
+        tmr.start()
+        self.assertIs(tmr._source, source_before)
+        tmr.kill()
+
 
 if __name__ == '__main__':
     unittest.main()

test/test-gutil.py

@@ -1,6 +1,10 @@
 #!/usr/bin/python3
+import os
 import unittest
-from staslib import gutil
+from staslib import conf, gutil, trid
+
+SUBSYSNQN = 'nqn.1988-11.com.dell:SFSS:2:20220208134025e8'
+HOSTNQN = 'nqn.2014-08.org.nvmexpress:uuid:01234567-0123-0123-0123-0123456789ab'
 
 
 class GutilUnitTest(unittest.TestCase):
@@ -45,5 +49,78 @@ def test_Deferred(self):
         self.assertFalse(d.is_scheduled())
 
 
+# ==============================================================================
+class TestNameResolver(unittest.TestCase):
+    '''Unit tests for NameResolver.resolve_ctrl_async() — synchronous paths only.
+
+    When traddr is already a valid IP address the resolver skips the async DNS
+    lookup and calls the callback immediately from within resolve_ctrl_async().
+    Non-TCP/RDMA transports also bypass DNS.  Only those two paths are exercised
+    here; async DNS resolution requires a live network and is not unit-testable.
+    '''
+
+    FNAME_BOTH = '/tmp/stas-test-nr-both.conf'
+    FNAME_IPV6 = '/tmp/stas-test-nr-ipv6.conf'
+
+    @classmethod
+    def setUpClass(cls):
+        with open(cls.FNAME_BOTH, 'w') as f:
+            f.write('[Global]\nip-family=ipv4+ipv6\n')
+        with open(cls.FNAME_IPV6, 'w') as f:
+            f.write('[Global]\nip-family=ipv6\n')
+        conf.SvcConf().set_conf_file(cls.FNAME_BOTH)
+
+    @classmethod
+    def tearDownClass(cls):
+        for fname in (cls.FNAME_BOTH, cls.FNAME_IPV6):
+            if os.path.exists(fname):
+                os.remove(fname)
+
+    def setUp(self):
+        # Reset to the "both families" config before every test so that a test
+        # that switches to FNAME_IPV6 does not leak into the next one.
+        conf.SvcConf().set_conf_file(self.FNAME_BOTH)
+
+    def _make_tid(self, transport, traddr):
+        return trid.TID({'transport': transport, 'traddr': traddr, 'subsysnqn': SUBSYSNQN, 'host-nqn': HOSTNQN})
+
+    def test_empty_list_calls_callback_immediately(self):
+        resolver = gutil.NameResolver()
+        result = []
+        resolver.resolve_ctrl_async(None, [], lambda ctrls: result.extend(ctrls))
+        self.assertEqual(result, [])
+
+    def test_ipv4_address_resolves_synchronously(self):
+        resolver = gutil.NameResolver()
+        t = self._make_tid('tcp', '10.10.10.10')
+        result = []
+        resolver.resolve_ctrl_async(None, [t], lambda ctrls: result.extend(ctrls))
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0].traddr, '10.10.10.10')
+
+    def test_ipv6_address_resolves_synchronously(self):
+        resolver = gutil.NameResolver()
+        t = self._make_tid('tcp', '::1')
+        result = []
+        resolver.resolve_ctrl_async(None, [t], lambda ctrls: result.extend(ctrls))
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0].traddr, '::1')
+
+    def test_fc_transport_passes_through_without_dns(self):
+        resolver = gutil.NameResolver()
+        t = self._make_tid('fc', 'nn-0x1000000044001123:pn-0x2000000055001123')
+        result = []
+        resolver.resolve_ctrl_async(None, [t], lambda ctrls: result.extend(ctrls))
+        self.assertEqual(len(result), 1)
+
+    def test_ipv4_excluded_when_only_ipv6_allowed(self):
+        conf.SvcConf().set_conf_file(self.FNAME_IPV6)
+        resolver = gutil.NameResolver()
+        t = self._make_tid('tcp', '10.10.10.10')
+        result = []
+        resolver.resolve_ctrl_async(None, [t], lambda ctrls: result.extend(ctrls))
+        self.assertEqual(result, [])
+
+
 if __name__ == '__main__':
     unittest.main()

test/test-stas.py

@@ -1,7 +1,9 @@
 #!/usr/bin/python3
 import os
 import unittest
+from unittest.mock import patch
 from staslib import conf, stas, trid
+from pyfakefs.fake_filesystem_unittest import TestCase as FakeTestCase
 
 HOSTNQN = 'nqn.2014-08.org.nvmexpress:uuid:01234567-0123-0123-0123-0123456789ab'
 SUBSYSNQN = 'nqn.1988-11.com.dell:SFSS:2:20220208134025e8'
@@ -258,5 +260,48 @@ def test_unknown_transport_always_removed(self):
         self.assertEqual(stas.remove_invalid_addresses([t]), [])
 
 
+# ==============================================================================
+class TestLoadIdl(unittest.TestCase):
+    '''Unit tests for stas.load_idl().'''
+
+    def test_stafd_idl_returns_content(self):
+        content = stas.load_idl('stafd.idl')
+        self.assertIsInstance(content, str)
+        self.assertGreater(len(content), 0)
+
+    def test_stacd_idl_returns_content(self):
+        content = stas.load_idl('stacd.idl')
+        self.assertIsInstance(content, str)
+        self.assertGreater(len(content), 0)
+
+    def test_nonexistent_idl_returns_empty_string(self):
+        content = stas.load_idl('nonexistent.idl')
+        self.assertEqual(content, '')
+
+
+# ==============================================================================
+class TestCheckIfAllowedToContinue(FakeTestCase):
+    '''Unit tests for stas.check_if_allowed_to_continue() using a fake filesystem.'''
+
+    def setUp(self):
+        self.setUpPyfakefs()
+
+    def test_exits_when_not_root(self):
+        with patch('os.geteuid', return_value=1000):
+            with self.assertRaises(SystemExit):
+                stas.check_if_allowed_to_continue()
+
+    def test_exits_when_nvme_fabrics_missing(self):
+        # /dev/nvme-fabrics is absent in the fake filesystem by default
+        with patch('os.geteuid', return_value=0):
+            with self.assertRaises(SystemExit):
+                stas.check_if_allowed_to_continue()
+
+    def test_returns_normally_when_root_and_nvme_fabrics_present(self):
+        self.fs.create_file('/dev/nvme-fabrics')
+        with patch('os.geteuid', return_value=0):
+            stas.check_if_allowed_to_continue()  # Must not raise
+
+
 if __name__ == '__main__':
     unittest.main()