Using hardware Simulators for Unit Testing // Issue platformio#4238

HISTORY.rst

@@ -45,9 +45,10 @@ Please check `Migration guide from 5.x to 6.0 <https://docs.platformio.org/en/la
 * **Unit Testing**
 
   - Refactored from scratch `Unit Testing <https://docs.platformio.org/en/latest/advanced/unit-testing/index.html>`_ solution and its documentation
-  - New `Test Hierarchies <https://docs.platformio.org/en/latest/advanced/unit-testing/structure.html>`_ (`issue #4135 <https://github.com/platformio/platformio-core/issues/4135>`_)
-  - New `Custom Testing Framework <https://docs.platformio.org/en/latest/advanced/unit-testing/frameworks/custom/index.html>`_
-  - New "test" `build configuration <https://docs.platformio.org/en/latest/projectconf/build_configurations.html>`__
+  - New: `Test Hierarchies <https://docs.platformio.org/en/latest/advanced/unit-testing/structure.html>`_ (`issue #4135 <https://github.com/platformio/platformio-core/issues/4135>`_)
+  - New: `Custom Testing Framework <https://docs.platformio.org/en/latest/advanced/unit-testing/frameworks/custom/index.html>`_
+  - New: Using hardware `Simulators <https://docs.platformio.org/en/latest/advanced/unit-testing/simulators/index.html>`__ for Unit Testing
+  - Added a new "test" `build configuration <https://docs.platformio.org/en/latest/projectconf/build_configurations.html>`__
   - Added support for the ``socket://`` and ``rfc2217://`` protocols using `test_port <https://docs.platformio.org/en/latest/projectconf/section_env_test.html#test-port>`__ option (`issue #4229 <https://github.com/platformio/platformio-core/issues/4229>`_)
   - Added support for a `Custom Unity Library <https://docs.platformio.org/en/latest/advanced/unit-testing/frameworks/custom/examples/custom_unity_library.html>`__ (`issue #3980 <https://github.com/platformio/platformio-core/issues/3980>`_)
   - 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>`_)

README.rst

@@ -40,7 +40,7 @@ PlatformIO Core
 .. image:: https://raw.githubusercontent.com/platformio/platformio-web/develop/app/images/platformio-ide-laptop.png
     :target: https://platformio.org?utm_source=github&utm_medium=core
 
-`PlatformIO <https://platformio.org>`_ is a professional collaborative platform for safety-critical and declarative embedded development.
+`PlatformIO <https://platformio.org>`_ is a professional collaborative platform for embedded development.
 
 **A place where Developers and Teams have true Freedom! No more vendor lock-in!**
 

platformio/project/options.py

@@ -687,6 +687,15 @@ def get_default_core_dir():
                 type=click.BOOL,
                 default=False,
             ),
+            ConfigEnvOption(
+                group="test",
+                name="test_testing_command",
+                multiple=True,
+                description=(
+                    "A custom testing command that runs test cases "
+                    "and returns results to the standard output"
+                ),
+            ),
             # Debug
             ConfigEnvOption(
                 group="debug",

platformio/test/runners/base.py

@@ -144,11 +144,17 @@ def stage_testing(self):
             return None
         click.secho("Testing...", bold=self.options.verbose)
         test_port = self.get_test_port()
-        serial_conds = [self.platform.is_embedded(), test_port and "://" in test_port]
+        program_conds = [
+            not self.platform.is_embedded()
+            and (not test_port or "://" not in test_port),
+            self.project_config.get(
+                f"env:{self.test_suite.env_name}", "test_testing_command"
+            ),
+        ]
         reader = (
-            SerialTestOutputReader(self)
-            if any(serial_conds)
-            else ProgramTestOutputReader(self)
+            ProgramTestOutputReader(self)
+            if any(program_conds)
+            else SerialTestOutputReader(self)
         )
         return reader.begin()
 

platformio/test/runners/readers/program.py

@@ -12,28 +12,88 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import os
 import signal
+import subprocess
+import time
 
-from platformio import proc
+from platformio.compat import IS_WINDOWS, get_filesystem_encoding, get_locale_encoding
 from platformio.test.exception import UnitTestError
 
 
+class ProgramProcessProtocol(asyncio.SubprocessProtocol):
+    def __init__(self, test_runner, exit_future):
+        self.test_runner = test_runner
+        self.exit_future = exit_future
+
+    def pipe_data_received(self, _, data):
+        try:
+            data = data.decode(get_locale_encoding() or get_filesystem_encoding())
+        except UnicodeDecodeError:
+            data = data.decode("latin-1")
+        self.test_runner.on_test_output(data)
+        if self.test_runner.test_suite.is_finished():
+            self._stop_testing()
+
+    def process_exited(self):
+        self._stop_testing()
+
+    def _stop_testing(self):
+        if not self.exit_future.done():
+            self.exit_future.set_result(True)
+
+
 class ProgramTestOutputReader:
+
+    KILLING_TIMEOUT = 5  # seconds
+
     def __init__(self, test_runner):
         self.test_runner = test_runner
+        self.aio_loop = (
+            asyncio.ProactorEventLoop() if IS_WINDOWS else asyncio.new_event_loop()
+        )
+        asyncio.set_event_loop(self.aio_loop)
 
-    def begin(self):
+    def get_testing_command(self):
+        custom_testing_command = self.test_runner.project_config.get(
+            f"env:{self.test_runner.test_suite.env_name}", "test_testing_command"
+        )
+        if custom_testing_command:
+            return custom_testing_command
         build_dir = self.test_runner.project_config.get("platformio", "build_dir")
-        result = proc.exec_command(
-            [os.path.join(build_dir, self.test_runner.test_suite.env_name, "program")],
-            stdout=proc.LineBufferedAsyncPipe(self.test_runner.on_test_output),
-            stderr=proc.LineBufferedAsyncPipe(self.test_runner.on_test_output),
+        return [
+            os.path.join(build_dir, self.test_runner.test_suite.env_name, "program")
+        ]
+
+    async def gather_results(self):
+        exit_future = asyncio.Future(loop=self.aio_loop)
+        transport, _ = await self.aio_loop.subprocess_exec(
+            lambda: ProgramProcessProtocol(self.test_runner, exit_future),
+            *self.get_testing_command(),
+            stdin=None,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
         )
-        if result["returncode"] == 0:
-            return True
+        await exit_future
+        last_return_code = transport.get_returncode()
+        transport.close()
+
+        # wait until subprocess will be killed
+        start = time.time()
+        while (
+            start > (time.time() - self.KILLING_TIMEOUT)
+            and transport.get_returncode() is None
+        ):
+            await asyncio.sleep(0.5)
+
+        if last_return_code:
+            self.raise_for_status(last_return_code)
+
+    @staticmethod
+    def raise_for_status(return_code):
         try:
-            sig = signal.Signals(abs(result["returncode"]))
+            sig = signal.Signals(abs(return_code))
             try:
                 signal_description = signal.strsignal(sig)
             except AttributeError:
@@ -42,4 +102,11 @@ def begin(self):
                 f"Program received signal {sig.name} ({signal_description})"
             )
         except ValueError:
-            raise UnitTestError("Program errored with %d code" % result["returncode"])
+            raise UnitTestError("Program errored with %d code" % return_code)
+
+    def begin(self):
+        try:
+            self.aio_loop.run_until_complete(self.gather_results())
+        finally:
+            self.aio_loop.run_until_complete(self.aio_loop.shutdown_asyncgens())
+            self.aio_loop.close()

tests/commands/test_test.py

@@ -13,9 +13,12 @@
 # limitations under the License.
 
 import os
+import sys
 import xml.etree.ElementTree as ET
 from pathlib import Path
 
+import pytest
+
 from platformio import proc
 from platformio.test.command import test_cmd as pio_test_cmd
 
@@ -212,6 +215,69 @@ def test_crashed_program(clirunner, tmpdir):
     )
 
 
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="runs only on Unix (issue with SimAVR)"
+)
+def test_custom_testing_command(clirunner, validate_cliresult, tmp_path: Path):
+    project_dir = tmp_path / "project"
+    project_dir.mkdir()
+    (project_dir / "platformio.ini").write_text(
+        """
+[env:uno]
+platform = atmelavr
+framework = arduino
+board = uno
+
+platform_packages =
+    platformio/tool-simavr @ ^1
+test_speed = 9600
+test_testing_command =
+    ${platformio.packages_dir}/tool-simavr/bin/simavr
+    -m
+    atmega328p
+    -f
+    16000000L
+    ${platformio.build_dir}/${this.__env__}/firmware.elf
+"""
+    )
+    test_dir = project_dir / "test" / "test_dummy"
+    test_dir.mkdir(parents=True)
+    (test_dir / "test_main.cpp").write_text(
+        """
+#include <Arduino.h>
+#include <unity.h>
+
+void setUp(void) {
+    // set stuff up here
+}
+
+void tearDown(void) {
+    // clean stuff up here
+}
+
+void dummy_test(void) {
+    TEST_ASSERT_EQUAL(1, 1);
+}
+
+void setup() {
+    UNITY_BEGIN();
+    RUN_TEST(dummy_test);
+    UNITY_END();
+}
+
+void loop() {
+    delay(1000);
+}
+"""
+    )
+    result = clirunner.invoke(
+        pio_test_cmd,
+        ["-d", str(project_dir), "--without-uploading"],
+    )
+    validate_cliresult(result)
+    assert "dummy_test" in result.output
+
+
 def test_unity_setup_teardown(clirunner, validate_cliresult, tmpdir):
     project_dir = tmpdir.mkdir("project")
     project_dir.join("platformio.ini").write(