Reworking mocks classes

Author: bridadan

test/test/host_test_black_box.py

@@ -17,277 +17,15 @@
 import unittest
 import os
 import re
-
-import threading
-import shutil
-import tempfile
 from builtins import super
 from copy import copy
 from mbed_os_tools.test import init_host_test_cli_params
 from mbed_os_tools.test.host_tests_runner.host_test_default  import DefaultTestSelector
 from mock import patch, MagicMock
 
-
-class MockThread(threading.Thread):
-    def __init__(self, target=None, args=None):
-        super().__init__(target=target, args=args)
-        self._terminates = 0
-        self.exitcode = 0 # TODO maybe this needs to be setable? Mock sys.exit
-
-    def terminate(self):
-        self._terminates += 1
-
-class MockSerial(object):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self._args = args
-        self._kwargs = kwargs
-        self._open = True
-        self._rx_counter = 0
-        self._tx_buffer = b""
-        self._rx_buffer = b""
-        self._upstream_write_cb = None
-
-    def read(self, count):
-        contents = self._rx_buffer[self._rx_counter:count]
-        self._rx_counter += len(contents)
-        return contents
-
-    def write(self, data):
-        self._tx_buffer += data
-        if self._upstream_write_cb:
-            # TODO this may not work...
-            self._upstream_write_cb(data)
-
-    def close(self):
-        self._open = False
-
-    def downstream_write(self, data):
-        self._rx_buffer += data.encode("utf-8")
-
-    def downstream_write_bytes(self, data):
-        self._rx_buffer += data
-
-    def on_upstream_write(self, func):
-        self._upstream_write_cb = func
-
-kv_regex = re.compile("\{\{([\w\d_-]+);([^\}]+)\}\}")
-
-class MockMbedDevice(object):
-    def __init__(self, serial):
-        self._synced = False
-        self._kvs = []
-        self._serial = serial
-        self._serial.on_upstream_write(self.on_write)
-
-    def handle_kv(self, key, value):
-        if not self._synced:
-            if key == "__sync":
-                self._synced = True
-                self.send_kv(key, value)
-                self.on_sync()
-        else:
-            pass
-
-    def send_kv(self, key, value):
-        self._serial.downstream_write("{{{{{};{}}}}}\r\n".format(key, value))
-
-    def on_write(self, data):
-        kvs = kv_regex.findall(data.decode("utf-8"))
-
-        for key, value in kvs:
-            self.handle_kv(key, value)
-            self._kvs.append((key, value))
-
-    def on_sync(self):
-        self._serial.downstream_write(
-            "{{__timeout;15}}\r\n"
-            "{{__host_test_name;default_auto}}\r\n"
-            "{{end;success}}\n"
-            "{{__exit;0}}\r\n"
-        )
-
-
-def _process_side_effect(target=None, args=None):
-    return MockThread(target=target, args=args)
-
-class MockTestEnvironment(object):
-
-    def __init__(self, test_case, platform_info, image_path):
-        self._test_case = test_case
-        self._tempdir = tempfile.mkdtemp()
-        self._platform_info = copy(platform_info)
-        self._platform_info['mount_point'] = os.path.join(self._tempdir, self._platform_info['mount_point'])
-        self._platform_info['serial_port'] = os.path.join(self._tempdir, self._platform_info['serial_port'])
-        self._image_path = os.path.join(self._tempdir, image_path)
-
-        self._patch_definitions = []
-        self.patches = {}
-
-        args = (
-            'mbedhtrun -m {} -p {}:9600 -f '
-            '"{}" -e "TESTS/host_tests" -d {} -c default '
-            '-t {} -r default '
-            '-C 4 --sync 5 -P 60'
-        ).format(
-            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):
-        os.makedirs(os.path.dirname(self._image_path))
-        with open(self._image_path, 'w') as _:
-            pass
-
-        os.makedirs(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()
-
-        shutil.rmtree(self._tempdir)
-
-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)
+from .mocks.environment.linux import MockTestEnvironmentLinux
+from .mocks.environment.darwin import MockTestEnvironmentDarwin
+from .mocks.environment.windows import MockTestEnvironmentWindows
 
 mock_platform_info = {
     "platform_name": "K64F",

test/test/mocks/__init__.py

@@ -0,0 +1,84 @@
+import tempfile
+import os
+import shutil
+
+from mock import patch, MagicMock
+from copy import copy
+from ..serial import MockSerial
+from ..mbed_device import MockMbedDevice
+from ..thread import MockThread
+
+class MockTestEnvironment(object):
+
+    def __init__(self, test_case, platform_info, image_path):
+        self._test_case = test_case
+        self._tempdir = tempfile.mkdtemp()
+        self._platform_info = copy(platform_info)
+        self._platform_info['mount_point'] = os.path.splitdrive(os.path.join(self._tempdir, self._platform_info['mount_point']))[1]
+        self._platform_info['serial_port'] =  os.path.splitdrive(os.path.join(self._tempdir, self._platform_info['serial_port']))[1]
+        self._image_path =  os.path.splitdrive(os.path.join(self._tempdir, image_path))[1]
+
+        self._patch_definitions = []
+        self.patches = {}
+
+        args = (
+            'mbedhtrun -m {} -p {}:9600 -f '
+            '"{}" -e "TESTS/host_tests" -d {} -c default '
+            '-t {} -r default '
+            '-C 4 --sync 5 -P 60'
+        ).format(
+            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=self._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)
+        )
+
+    @staticmethod
+    def _process_side_effect(target=None, args=None):
+        return MockThread(target=target, args=args)
+
+    def patch(self, path, **kwargs):
+        self._patch_definitions.append((path, patch(path, **kwargs)))
+
+    def __enter__(self):
+        os.makedirs(os.path.dirname(self._image_path))
+        with open(self._image_path, 'w') as _:
+            pass
+
+        os.makedirs(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()
+
+        shutil.rmtree(self._tempdir)

test/test/mocks/environment/__init__.py

@@ -0,0 +1,32 @@
+from builtins import super
+from mock import MagicMock
+
+from .posix import MockTestEnvironmentPosix
+
+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)
+
+        if value:
+            return False
+
+        # 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)