Move environment related mocks into classes

Author: bridadan

test/test/host_test_black_box.py

@@ -107,69 +107,216 @@ def on_sync(self):
 def _process_side_effect(target=None, args=None):
     return MockThread(target=target, args=args)
 
-class BlackBoxHostTestTestCase(fake_filesystem_unittest.TestCase):
+class MockTestEnvironment(object):
 
-    def setUp(self):
-        self.setUpPyfakefs()
+    def __init__(self, test_case, platform_info, image_path):
+        self._test_case = test_case
+        self._platform_info = platform_info
+        self._image_path = image_path
+        self._patch_definitions = []
+        self.patches = {}
 
-    def tearDown(self):
-        pass
-
-    def test_host_test_has_setup_teardown_attribute(self):
-        platform_info = {
-            "platform_name": "K64F",
-            "target_id": "0240000031754e45000c0018948500156461000097969900",
-            "mount_point": os.path.normpath("/mnt/DAPLINK"),
-            "serial_port": os.path.normpath("/dev/ttyACM0"),
-        }
-        image_path = os.path.normpath(
-            "BUILD/tests/K64F/GCC_ARM/TESTS/network/interface/interface.bin"
-        )
-        self.fs.create_file(image_path)
-        self.fs.create_dir(platform_info['mount_point'])
         args = (
             'mbedhtrun -m {} -p {}:9600 -f '
             '"{}" -e "TESTS/host_tests" -d {} -c default '
             '-t {} -r default '
             '-C 4 --sync 5 -P 60'
         ).format(
-            platform_info['platform_name'],
-            platform_info['serial_port'],
-            image_path,
-            platform_info['mount_point'],
-            platform_info['target_id']
+            self._platform_info['platform_name'],
+            self._platform_info['serial_port'],
+            self._image_path,
+            self._platform_info['mount_point'],
+            self._platform_info['target_id']
         ).split()
+        self.patch('sys.argv', new=args)
+
+        # Mock detect
+        detect_mock = MagicMock()
+        detect_mock.return_value.list_mbeds.return_value = [
+            self._platform_info
+        ]
+        self.patch('mbed_os_tools.detect.create', new=detect_mock)
+
+        # Mock process
+        self.patch(
+            'mbed_os_tools.test.host_tests_runner.host_test_default.Process',
+            new=MagicMock(side_effect=_process_side_effect)
+        )
+        self.patch(
+            'mbed_os_tools.test.host_tests_plugins.host_test_plugins.call',
+            new=MagicMock(return_value=0)
+        )
+
+        # Mock serial
+        mock_serial = MockSerial()
+        mock_device = MockMbedDevice(mock_serial)
+        self.patch(
+            'mbed_os_tools.test.host_tests_conn_proxy.conn_primitive_serial.Serial',
+            new=MagicMock(return_value=mock_serial)
+        )
+
+
+    def patch(self, path, **kwargs):
+        self._patch_definitions.append((path, patch(path, **kwargs)))
+
+    def __enter__(self):
+        self._test_case.fs.create_file(self._image_path)
+        self._test_case.fs.create_dir(self._platform_info['mount_point'])
+
+        for path, patcher in self._patch_definitions:
+            self.patches[path] = patcher.start()
+
+    def __exit__(self, type, value, traceback):
+        for _, patcher in self._patch_definitions:
+            patcher.stop()
+
+class MockTestEnvironmentPosix(MockTestEnvironment):
+
+    def __init__(self, test_case, platform_info, image_path):
+        super().__init__(test_case, platform_info, image_path)
+
+        self.patch('os.name', new='posix')
+
+    def __exit__(self, type, value, traceback):
+        super().__exit__(type, value, traceback)
+
+        # Assert for proper image copy
+        mocked_call = self.patches[
+            'mbed_os_tools.test.host_tests_plugins.host_test_plugins.call'
+        ]
+
+        first_call_args = mocked_call.call_args_list[0][0][0]
+        self._test_case.assertEqual(first_call_args[0], "cp")
+        self._test_case.assertEqual(first_call_args[1], self._image_path)
+        self._test_case.assertTrue(first_call_args[2].startswith(self._platform_info["mount_point"]))
+        self._test_case.assertTrue(first_call_args[2].endswith(os.path.splitext(self._image_path)[1]))
+
+
+class MockTestEnvironmentLinux(MockTestEnvironmentPosix):
+
+    def __init__(self, test_case, platform_info, image_path):
+        super().__init__(test_case, platform_info, image_path)
+
+        self.patch(
+            'os.uname',
+            new=MagicMock(return_value=('Linux',)),
+            create=True
+        )
+
+    def __exit__(self, type, value, traceback):
+        super().__exit__(type, value, traceback)
+
+        # Assert for proper image copy
+        mocked_call = self.patches[
+            'mbed_os_tools.test.host_tests_plugins.host_test_plugins.call'
+        ]
+
+        second_call_args = mocked_call.call_args_list[1][0][0]
+        destination_path = os.path.normpath(
+            os.path.join(
+                self._platform_info["mount_point"],
+                os.path.basename(self._image_path)
+            )
+        )
+
+        self._test_case.assertEqual(
+            second_call_args,
+            ["sync", "-f", destination_path]
+        )
+
+        # Ensure only two subprocesses were started
+        self._test_case.assertEqual(len(mocked_call.call_args_list), 2)
+
+class MockTestEnvironmentDarwin(MockTestEnvironmentPosix):
+
+    def __init__(self, test_case, platform_info, image_path):
+        super().__init__(test_case, platform_info, image_path)
+
+        self.patch(
+            'os.uname',
+            new=MagicMock(return_value=('Darwin',)),
+            create=True
+        )
+
+    def __exit__(self, type, value, traceback):
+        super().__exit__(type, value, traceback)
+
+        # Assert for proper image copy
+        mocked_call = self.patches[
+            'mbed_os_tools.test.host_tests_plugins.host_test_plugins.call'
+        ]
+
+        second_call_args = mocked_call.call_args_list[1][0][0]
+        self._test_case.assertEqual(second_call_args, ["sync"])
+
+        # Ensure only two subprocesses were started
+        self._test_case.assertEqual(len(mocked_call.call_args_list), 2)
+
+class MockTestEnvironmentWindows(MockTestEnvironment):
+
+    def __init__(self, test_case, platform_info, image_path):
+        super().__init__(test_case, platform_info, image_path)
+
+        self.patch('os.name', new='nt')
+
+    def __exit__(self, type, value, traceback):
+        super().__exit__(type, value, traceback)
+
+        # Assert for proper image copy
+        mocked_call = self.patches[
+            'mbed_os_tools.test.host_tests_plugins.host_test_plugins.call'
+        ]
+
+        first_call_args = mocked_call.call_args_list[0][0][0]
+        self._test_case.assertEqual(first_call_args[0], "copy")
+        self._test_case.assertEqual(first_call_args[1], self._image_path)
+        self._test_case.assertTrue(first_call_args[2].startswith(self._platform_info["mount_point"]))
+        self._test_case.assertTrue(first_call_args[2].endswith(os.path.splitext(self._image_path)[1]))
+
+        # Ensure only one subprocess was started
+        self._test_case.assertEqual(len(mocked_call.call_args_list), 1)
 
-        with patch('sys.argv', new=args) as _argv,\
-             patch('os.name', new='posix'),\
-             patch('os.uname', return_value=('Linux',), create=True),\
-             patch('mbed_os_tools.detect.create') as _detect,\
-             patch('mbed_os_tools.test.host_tests_plugins.host_test_plugins.call') as _call,\
-             patch('mbed_os_tools.test.host_tests_runner.host_test_default.Process') as _process,\
-             patch('mbed_os_tools.test.host_tests_conn_proxy.conn_primitive_serial.Serial') as _serial:
-            _process.side_effect = _process_side_effect
-            _detect.return_value.list_mbeds.return_value = [
-                platform_info
-            ]
-            _call.return_value = 0
-
-            mock_serial = MockSerial()
-            mock_device = MockMbedDevice(mock_serial)
-            _serial.return_value = mock_serial
+mock_platform_info = {
+    "platform_name": "K64F",
+    "target_id": "0240000031754e45000c0018948500156461000097969900",
+    "mount_point": os.path.normpath("/mnt/DAPLINK"),
+    "serial_port": os.path.normpath("/dev/ttyACM0"),
+}
+mock_image_path = os.path.normpath(
+    "BUILD/tests/K64F/GCC_ARM/TESTS/network/interface/interface.bin"
+)
 
+class BlackBoxHostTestTestCase(fake_filesystem_unittest.TestCase):
+
+    def setUp(self):
+        self.setUpPyfakefs()
+
+    def tearDown(self):
+        pass
+
+    def test_host_test_linux(self):
+        with MockTestEnvironmentLinux(self, mock_platform_info, mock_image_path) as _env:
             test_selector = DefaultTestSelector(init_host_test_cli_params())
             result = test_selector.execute()
             test_selector.finish()
 
-            matched_calls = [
-                ca for ca in _call.call_args_list if ca[0][0][0] == "cp"
-            ]
-            cmd_args = matched_calls[0][0][0]
-            self.assertEqual(cmd_args[1], image_path)
-            self.assertTrue(cmd_args[2].startswith(platform_info["mount_point"]))
-            self.assertTrue(cmd_args[2].endswith(os.path.splitext(image_path)[1]))
+        self.assertEqual(result, 0)
+
+    def test_host_test_darwin(self):
+        with MockTestEnvironmentDarwin(self, mock_platform_info, mock_image_path) as _env:
+            test_selector = DefaultTestSelector(init_host_test_cli_params())
+            result = test_selector.execute()
+            test_selector.finish()
+
+        self.assertEqual(result, 0)
+
+    def test_host_test_windows(self):
+        with MockTestEnvironmentWindows(self, mock_platform_info, mock_image_path) as _env:
+            test_selector = DefaultTestSelector(init_host_test_cli_params())
+            result = test_selector.execute()
+            test_selector.finish()
 
-            self.assertEqual(result, 0)
+        self.assertEqual(result, 0)
 
 if __name__ == '__main__':
     unittest.main()