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()