cmake: Add unit testing framework to MPM

Add unit testing framework to MPM which can be run by calling
"make test". The testing is done using the built in unittest Python
module. Tests can be run on a dev machine or on the USRP itself when
compiling natively.

4 files changed, 243 insertions, 0 deletions

diff --git a/mpm/python/CMakeLists.txt b/mpm/python/CMakeLists.txt
index 7338b5167..84cff3e54 100644
--- a/mpm/python/CMakeLists.txt
+++ b/mpm/python/CMakeLists.txt
@@ -66,3 +66,5 @@ elseif (ENABLE_E320)
         DESTINATION ${RUNTIME_DIR}
     )
 endif (ENABLE_MYKONOS)
+
+add_subdirectory(tests)
diff --git a/mpm/python/tests/CMakeLists.txt b/mpm/python/tests/CMakeLists.txt
new file mode 100755
index 000000000..26dcad40a
--- /dev/null
+++ b/mpm/python/tests/CMakeLists.txt
@@ -0,0 +1,14 @@
+#
+# Copyright 2019 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+
+########################################################################
+# This file included, use CMake directory variables
+########################################################################
+
+add_test(
+  NAME mpm_unit_tests
+  COMMAND ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/run_unit_tests.py ${MPM_DEVICE}
+)
diff --git a/mpm/python/tests/run_unit_tests.py b/mpm/python/tests/run_unit_tests.py
new file mode 100755
index 000000000..d7c288aaa
--- /dev/null
+++ b/mpm/python/tests/run_unit_tests.py
@@ -0,0 +1,64 @@
+#
+# Copyright 2019 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+"""
+USRP MPM Python Unit testing framework
+"""
+
+import unittest
+import sys
+from sys_utils_tests import TestNet
+
+TESTS = {
+    '__all__': {TestNet},
+    'n3xx': set(),
+}
+
+def get_test_suite(device_name=''):
+    """
+    Gets a test suite (collection of test cases) which is relevant for
+    the specified device.
+    """
+    # A collection of test suites, generated by test loaders, which will
+    # be later combined
+    test_suite_list = []
+    test_loader = unittest.TestLoader()
+
+    # Combine generic and device specific tests
+    test_cases = TESTS.get('__all__') | TESTS.get(device_name, set())
+    for case in test_cases:
+        new_suite = test_loader.loadTestsFromTestCase(case)
+        for test in new_suite:
+            # Set up test case class for a specific device.
+            # Each test uses a different test case instance.
+            if (hasattr(test, 'set_device_name')) and (device_name != ''):
+                test.set_device_name(device_name)
+        test_suite_list.append(new_suite)
+
+    # Individual test suites are combined into a master test suite
+    test_suite = unittest.TestSuite(test_suite_list)
+    return test_suite
+
+def run_tests(device_name=''):
+    """
+    Executes the unit tests specified by the test suite.
+    This should be called from CMake.
+    """
+    test_result = unittest.TestResult()
+    test_runner = unittest.TextTestRunner(verbosity=2)
+    test_result = test_runner.run(get_test_suite(device_name))
+    return test_result
+
+def main():
+    if len(sys.argv) >= 2:
+        mpm_device_name = sys.argv[1]
+    else:
+        mpm_device_name = ''
+
+    if not run_tests(mpm_device_name).wasSuccessful():
+        sys.exit(-1)
+
+if __name__ == "__main__":
+    main()
diff --git a/mpm/python/tests/sys_utils_tests.py b/mpm/python/tests/sys_utils_tests.py
new file mode 100755
index 000000000..fc0ae33a1
--- /dev/null
+++ b/mpm/python/tests/sys_utils_tests.py
@@ -0,0 +1,163 @@
+#
+# Copyright 2019 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+"""
+Tests related to usrp_mpm.sys_utils
+"""
+
+import platform
+import unittest
+from usrp_mpm.sys_utils import net
+
+class TestNet(unittest.TestCase):
+    """
+    Tests multiple functions defined in usrp_mpm.sys_utils.net.
+
+    Some tests are system agnostic and some are only run on
+    USRPs with an ARM processor.
+    For tests run on the USRP, it is assumed that the device has at
+    least an active RJ-45 (eth0) connection.
+    """
+    def skipUnlessOnLinux():
+        """
+        Test function decorator which skips tests unless the current
+        execution environment is a linux OS.
+        """
+        if 'linux' in platform.system().lower():
+            return lambda func: func
+        return unittest.skip("This test is only valid when run on a Linux system.")
+
+    def skipUnlessOnUsrp():
+        """
+        Test function decorator which skips tests unless the current
+        execution environment is a USRP.
+
+        Assumes that 'arm' in the machine name constitutes an ARM
+        processor, aka a USRP.
+        """
+        if 'arm' in platform.machine().lower():
+            return lambda func: func
+        return unittest.skip("This test is only valid when run on the USRP.")
+
+    def set_device_name(self, device_name):
+        """
+        Stores a device name attribute for tests whose success condition
+        depends on the current device.
+        """
+        self.device_name = device_name
+
+    def test_get_hostname(self):
+        """
+        Test net.get_hostname() returns the same value as
+        platform.node() which should also be the network hostname of
+        the current system.
+        """
+        expected_hostname = platform.node()
+        self.assertEqual(expected_hostname, net.get_hostname())
+
+    @skipUnlessOnUsrp()
+    def test_get_valid_interfaces(self):
+        """
+        Test that expected network interfaces are returned as valid
+        and that unexpected network interfaces are not.
+
+        This test assumes there is an ethernet connection to the USRP
+        RJ-45 connector and will fail otherwise.
+
+        Note: This test is only valid when run on a USRP because the
+        network interfaces of a dev machine are unknown.
+        """
+        expected_valid_ifaces = ['eth0']
+        expected_invalid_ifaces = ['eth2', 'spf2']
+        all_ifaces = expected_valid_ifaces + expected_invalid_ifaces
+        resulting_valid_ifaces = net.get_valid_interfaces(all_ifaces)
+        self.assertEqual(expected_valid_ifaces, resulting_valid_ifaces)
+
+    @skipUnlessOnUsrp()
+    def test_get_iface_info(self):
+        """
+        Tests the get_iface_info function.
+        Expected ifaces should return information in the correct format
+        while unexpected ifaces should raise an IndexError.
+
+        Note: This test is only valid when run on a USRP because the
+        network interfaces of a dev machine are unknown.
+        """
+        if self.device_name == 'n3xx':
+            possible_ifaces = ['eth0', 'sfp0', 'sfp1']
+        else:
+            possible_ifaces = ['eth0', 'sfp0']
+
+        active_ifaces = net.get_valid_interfaces(possible_ifaces)
+
+        for iface_name in possible_ifaces:
+            iface_info = net.get_iface_info(iface_name)
+            # Verify the output info contains the expected keys
+            self.assertGreaterEqual(set(iface_info), {'mac_addr', 'ip_addr', 'ip_addrs', 'link_speed'})
+            if iface_name in active_ifaces:
+                # Verify interfaces with an active connection have a set IPv4 address
+                self.assertNotEqual(iface_info['ip_addr'], '')
+
+        unknown_name = 'unknown_iface'
+        # Verify that an unknown interface throws a LookupError
+        self.assertRaises(LookupError, net.get_iface_info, unknown_name)
+
+    @skipUnlessOnUsrp()
+    def test_get_link_speed(self):
+        """
+        Tests that the link speed of 'eth0' is the expected 1GB and that
+        when the function is called on unknown interfaces, an exception
+        is raised.
+
+        Note: This test is only valid when run on a USRP because the
+        network interfaces of a dev machine are unknown.
+        """
+        known_iface = 'eth0'
+        self.assertEqual(1000, net.get_link_speed(known_iface))
+        unknown_iface = 'unknown'
+        self.assertRaises(IndexError, net.get_link_speed, unknown_iface)
+
+    def test_ip_addr_to_iface(self):
+        """
+        Tests ip_addr_to_iface to ensure that the iface name is looked
+        up properly.
+        """
+        iface_list = {
+            'eth0': {
+                'mac_addr': None,
+                'ip_addr': '10.2.34.6',
+                'ip_addrs': ['10.2.99.99', '10.2.34.6'],
+                'link_speed': None,
+            },
+            'eth1': {
+                'mac_addr': None,
+                'ip_addr': '10.2.99.99',
+                'ip_addrs': ['10.2.99.99'],
+                'link_speed': None,
+            }
+        }
+        self.assertEqual(net.ip_addr_to_iface('10.2.34.6', iface_list), 'eth0')
+        self.assertEqual(net.ip_addr_to_iface('10.2.99.99', iface_list), 'eth1')
+        # TODO: If the IP address cannot be found it should probably not
+        # raise a KeyError but instead fail more gracefully
+        self.assertRaises(KeyError, net.ip_addr_to_iface, '10.2.100.100', iface_list)
+
+    def test_byte_to_mac(self):
+        """
+        Test the conversion from byte string to formatted MAC address.
+        Compares an expected formatted MAC address with the actual
+        returned value.
+        """
+        mac_addr = 0x2F16ABBF9063
+        byte_str = ""
+        for byte_index in range(0, 6):
+            byte = (mac_addr >> (byte_index * 8)) & 0xFF
+            byte_char = chr(byte)
+            byte_str = byte_char + byte_str
+        expected_string = '2F:16:AB:BF:90:63'
+        self.assertEqual(expected_string, net.byte_to_mac(byte_str).upper())
+
+if __name__ == '__main__':
+    unittest.main()