Refactor test runner mixins to the test output readers

HISTORY.rst

@@ -50,7 +50,7 @@ Please check `Migration guide from 5.x to 6.0 <https://docs.platformio.org/en/la
   - New "test" `build configuration <https://docs.platformio.org/en/latest/projectconf/build_configurations.html>`__
   - Generate reports in JUnit and JSON formats using the `pio test --output-format <https://docs.platformio.org/en/latest/core/userguide/cmd_test.html#cmdoption-pio-test-output-format>`__ option  (`issue #2891 <https://github.com/platformio/platformio-core/issues/2891>`_)
   - Provide more information when the native program crashed on a host (errored with a negative return code) (`issue #3429 <https://github.com/platformio/platformio-core/issues/3429>`_)
-  - Fixed an issue when command line parameters ("--ignore", "--filter") do not override values defined in the |PIOCONF| (`issue #3845 <https://github.com/platformio/platformio-core/issues/3845>`_)
+  - Fixed an issue when command line parameters (``--ignore``, ``--filter``) do not override values defined in the |PIOCONF| (`issue #3845 <https://github.com/platformio/platformio-core/issues/3845>`_)
   - Renamed the "test_build_project_src" project configuration option to `test_build_src <https://docs.platformio.org/en/latest//projectconf/section_env_test.html#test-build-src>`__
   - Removed the "test_transport" option in favor of the `Custom "unity_config.h" <https://docs.platformio.org/en/latest/advanced/unit-testing/frameworks/unity.html>`_
 

platformio/test/runners/base.py

@@ -18,8 +18,8 @@ from platformio.exception import ReturnErrorCode
 from platformio.platform.factory import PlatformFactory
 from platformio.test.exception import UnitTestSuiteError
 from platformio.test.result import TestCase, TestCaseSource, TestStatus
-from platformio.test.runners.mixins.embedded import TestRunnerEmbeddedMixin
+from platformio.test.runners.readers.program import ProgramTestOutputReader
-from platformio.test.runners.mixins.native import TestRunnerNativeMixin
+from platformio.test.runners.readers.serial import SerialTestOutputReader
 
 CTX_META_TEST_IS_RUNNING = __name__ + ".test_running"
 CTX_META_TEST_RUNNING_NAME = __name__ + ".test_running_name"
@@ -51,7 +51,7 @@ class TestRunnerOptions:  # pylint: disable=too-many-instance-attributes
         self.monitor_dtr = monitor_dtr
 
 
-class TestRunnerBase(TestRunnerNativeMixin, TestRunnerEmbeddedMixin):
+class TestRunnerBase:
 
     NAME = None
     EXTRA_LIB_DEPS = None
@@ -138,9 +138,12 @@ class TestRunnerBase(TestRunnerNativeMixin, TestRunnerEmbeddedMixin):
         if self.options.without_testing:
             return None
         click.secho("Testing...", bold=self.options.verbose)
-        if self.platform.is_embedded():
+        reader = (
-            return self.stage_testing_on_target()
+            SerialTestOutputReader(self)
-        return self.stage_testing_on_host()
+            if self.platform.is_embedded()
+            else ProgramTestOutputReader(self)
+        )
+        return reader.begin()
 
     def teardown(self):
         pass
@@ -170,9 +173,9 @@ class TestRunnerBase(TestRunnerNativeMixin, TestRunnerEmbeddedMixin):
 
     def on_test_output(self, data):
         click.echo(data, nl=False)
-        self.parse_testcases(data)
+        self.parse_test_cases(data)
 
-    def parse_testcases(self, data):
+    def parse_test_cases(self, data):
         if not self.TESTCASE_PARSE_RE:
             raise NotImplementedError()
 

platformio/test/runners/readers/program.py

@@ -19,13 +19,16 @@ from platformio import proc
 from platformio.test.exception import UnitTestError
 
 
-class TestRunnerNativeMixin:
+class ProgramTestOutputReader:
-    def stage_testing_on_host(self):
+    def __init__(self, test_runner):
-        build_dir = self.project_config.get("platformio", "build_dir")
+        self.test_runner = test_runner
+
+    def begin(self):
+        build_dir = self.test_runner.project_config.get("platformio", "build_dir")
         result = proc.exec_command(
-            [os.path.join(build_dir, self.test_suite.env_name, "program")],
+            [os.path.join(build_dir, self.test_runner.test_suite.env_name, "program")],
-            stdout=proc.LineBufferedAsyncPipe(self.on_test_output),
+            stdout=proc.LineBufferedAsyncPipe(self.test_runner.on_test_output),
-            stderr=proc.LineBufferedAsyncPipe(self.on_test_output),
+            stderr=proc.LineBufferedAsyncPipe(self.test_runner.on_test_output),
         )
         if result["returncode"] == 0:
             return True

platformio/test/runners/readers/serial.py

@@ -21,11 +21,14 @@ from platformio import util
 from platformio.exception import UserSideException
 
 
-class TestRunnerEmbeddedMixin:
+class SerialTestOutputReader:
 
     SERIAL_TIMEOUT = 600
 
-    def stage_testing_on_target(self):
+    def __init__(self, test_runner):
+        self.test_runner = test_runner
+
+    def begin(self):
         click.echo(
             "If you don't see any output for the first 10 secs, "
             "please reset board (press reset button)"
@@ -34,17 +37,17 @@ class TestRunnerEmbeddedMixin:
 
         try:
             ser = serial.Serial(
-                baudrate=self.get_test_speed(), timeout=self.SERIAL_TIMEOUT
+                baudrate=self.test_runner.get_test_speed(), timeout=self.SERIAL_TIMEOUT
             )
             ser.port = self.get_test_port()
-            ser.rts = self.options.monitor_rts
+            ser.rts = self.test_runner.options.monitor_rts
-            ser.dtr = self.options.monitor_dtr
+            ser.dtr = self.test_runner.options.monitor_dtr
             ser.open()
         except serial.SerialException as e:
             click.secho(str(e), fg="red", err=True)
             return None
 
-        if not self.options.no_reset:
+        if not self.test_runner.options.no_reset:
             ser.flushInput()
             ser.setDTR(False)
             ser.setRTS(False)
@@ -53,7 +56,7 @@ class TestRunnerEmbeddedMixin:
             ser.setRTS(True)
             sleep(0.1)
 
-        while not self.test_suite.is_finished():
+        while not self.test_runner.test_suite.is_finished():
             line = ser.readline().strip()
 
             # fix non-ascii output from device
@@ -68,19 +71,26 @@ class TestRunnerEmbeddedMixin:
                 continue
             if isinstance(line, bytes):
                 line = line.decode("utf8", "ignore")
-            self.on_test_output(line)
+            self.test_runner.on_test_output(line)
         ser.close()
 
     def get_test_port(self):
         # if test port is specified manually or in config
-        port = self.options.test_port or self.project_config.get(
+        port = (
-            f"env:{self.test_suite.env_name}", "test_port"
+            self.test_runner.options.test_port
+            or self.test_runner.project_config.get(
+                f"env:{self.test_runner.test_suite.env_name}", "test_port"
+            )
         )
         if port:
             return port
 
-        board = self.project_config.get(f"env:{self.test_suite.env_name}", "board")
+        board = self.test_runner.project_config.get(
-        board_hwids = self.platform.board_config(board).get("build.hwids", [])
+            f"env:{self.test_runner.test_suite.env_name}", "board"
+        )
+        board_hwids = self.test_runner.platform.board_config(board).get(
+            "build.hwids", []
+        )
         port = None
         elapsed = 0
         while elapsed < 5 and not port:

platformio/test/runners/unity.py

@@ -269,4 +269,4 @@ void unityOutputComplete(void) { unittest_uart_end(); }
             else:
                 click.echo(line)
 
-        return self.parse_testcases(data)
+        return self.parse_test_cases(data)