trunner: add support for pytests

trunner/config.py

@@ -8,7 +8,7 @@
 import yaml
 
 from trunner.ctx import TestContext
-from trunner.harness import PyHarness, unity_harness
+from trunner.harness import PyHarness, unity_harness, pytest_harness
 from trunner.types import AppOptions, BootloaderOptions, TestOptions, ShellOptions
 
 
@@ -62,6 +62,8 @@ def _parse_type(self, config: dict):
             self._parse_pyharness(config)
         elif test_type == "unity":
             self._parse_unity()
+        elif test_type == "pytest":
+            self._parse_pytest()
         else:
             raise ParserError("unknown key!")
 
@@ -99,6 +101,9 @@ def _parse_pyharness(self, config: dict):
     def _parse_unity(self):
         self.test.harness = PyHarness(self.ctx.target.dut, self.ctx, unity_harness, self.test.kwargs)
 
+    def _parse_pytest(self):
+        self.test.harness = PyHarness(self.ctx.host.dut, self.ctx, pytest_harness, self.test.kwargs)
+
     def _parse_load(self, config: dict):
         apps = config.get("load", [])
         apps_to_boot = []

trunner/harness/__init__.py

@@ -23,6 +23,7 @@
 from .psh import ShellHarness
 from .pyharness import PyHarness
 from .unity import unity_harness
+from .pytest import pytest_harness
 
 __all__ = [
     "HarnessBuilder",
@@ -46,4 +47,5 @@
     "PloImageProperty",
     "PloJffsImageProperty",
     "unity_harness",
+    "pytest_harness"
 ]

trunner/harness/pytest.py

@@ -0,0 +1,35 @@
+from typing import Optional
+
+from trunner.ctx import TestContext
+from trunner.dut import HostDut
+from trunner.types import Status, TestResult
+
+def pytest_harness(dut: HostDut, ctx: TestContext, result: TestResult, **kwargs) -> Optional[TestResult]:
+    test_re = r"\:\:(?P<group>.+)\:\:(?P<name>.+)(?P<status>PASSED|SKIPPED|FAILED|ERROR)"
+    final_re = r"\[100\%\]"
+
+    status = Status.OK
+
+    test_path = kwargs.get("path")
+    cmd = f"pytest -v --tb=no {test_path}"
+
+    dut.set_args(cmd, encoding="utf-8")
+    dut.open()
+
+    while True:
+        idx = dut.expect([test_re, final_re], timeout=200)
+        parsed = dut.match.groupdict()
+
+        if idx == 0:
+            sub_status = Status.from_str(parsed["status"])
+            if sub_status is Status.FAIL:
+                status = sub_status
+            subname = f"{parsed['name']}"
+            # result.add_subresult(subname, sub_status)
+            continue
+
+        elif idx == 1:
+            break
+
+
+    return TestResult(status=status, msg="")

trunner/host.py

@@ -2,12 +2,16 @@
 from abc import ABC, abstractmethod
 
 from trunner.ctx import TestContext
-
+from trunner.dut import HostDut
 
 class Host(ABC):  # pylint: disable=too-few-public-methods
     """Base class for Host abstraction"""
     name: str
 
+    def __init__(self) -> None:
+        super().__init__()
+        self.dut: HostDut = HostDut()
+
     @abstractmethod
     def has_gpio(self) -> bool:
         """Return true if host has gpio utility and can restart the target device using it."""

trunner/test_runner.py

@@ -263,7 +263,7 @@ def run_tests(self, tests: Sequence[TestOptions]) -> Sequence[TestResult]:
             if test.ignore:
                 result.skip()
             else:
-                set_logfiles(self.target.dut, self.ctx)
+                set_logfiles(self.ctx.target.dut and self.ctx.host.dut, self.ctx)
                 harness = self.target.build_test(test)
 
                 if not test.should_reboot:  # WARN: build_test may change TestOptions

trunner/types.py

@@ -34,7 +34,7 @@ class Status(Enum):
 
     @classmethod
     def from_str(cls, s):
-        if s in ("FAIL", "FAILED", "BAD"):
+        if s in ("FAIL", "FAILED", "BAD", "ERROR"):
             return Status.FAIL
         if s in ("OK", "PASS", "PASSED"):
             return Status.OK