Add child runner phase

Author: jonny12375

openhtf/__init__.py

@@ -25,6 +25,7 @@
 import openhtf.core.measurements
 import openhtf.core.monitors
 import openhtf.core.phase_branches
+import openhtf.core.phase_child_runner
 import openhtf.core.phase_collections
 import openhtf.core.phase_descriptor
 import openhtf.core.phase_group
@@ -70,6 +71,7 @@
     'DiagnosisCheckpoint',
     'DiagnosisCondition',
     'PhaseFailureCheckpoint',
+    'PhaseChildRunner',
     'PhaseSequence',
     'Subtest',
     'PhaseDescriptor',
@@ -108,6 +110,7 @@
 DiagnosisCheckpoint = openhtf.core.phase_branches.DiagnosisCheckpoint
 DiagnosisCondition = openhtf.core.phase_branches.DiagnosisCondition
 PhaseFailureCheckpoint = openhtf.core.phase_branches.PhaseFailureCheckpoint
+ChildRunnerPhase = openhtf.core.phase_child_runner.ChildRunnerPhase
 
 PhaseSequence = openhtf.core.phase_collections.PhaseSequence
 Subtest = openhtf.core.phase_collections.Subtest

openhtf/core/phase_child_runner.py

@@ -0,0 +1,40 @@
+import abc
+import attr
+from typing import Any, Callable, Dict, Text, Type
+
+from openhtf.core import phase_nodes, base_plugs
+from openhtf.core import phase_descriptor, base_plugs
+
+@attr.s(slots=True, frozen=True)
+class ChildRunnerPhase(phase_nodes.PhaseNode, abc.ABC):
+  pass
+
+  def _asdict(self) -> Dict[Text, Any]:
+    ret = attr.asdict(self)  # pytype: disable=wrong-arg-types  # attr-stubs
+    ret.update(name=self.name, doc=self.doc)
+    return ret
+
+  @property
+  def name(self) -> Text:
+    return "child_runner"
+
+  def apply_to_all_phases(self, func: Any) -> 'ChildRunnerPhase':
+    return self
+
+  def with_args(self: phase_nodes.WithModifierT,
+                **kwargs: Any) -> phase_nodes.WithModifierT:
+    """Send these keyword-arguments when phases are called."""
+    del kwargs  # Unused.
+    return self
+
+  def with_plugs(
+      self: phase_nodes.WithModifierT,
+      **subplugs: Type[base_plugs.BasePlug]) -> phase_nodes.WithModifierT:
+    """Substitute plugs for placeholders for this phase, error on unknowns."""
+    del subplugs  # Unused.
+    return self
+
+  def load_code_info(
+      self: phase_nodes.WithModifierT) -> phase_nodes.WithModifierT:
+    """Load coded info for all contained phases."""
+    return self

openhtf/core/test_descriptor.py

@@ -142,6 +142,8 @@ def PhaseTwo(test):
 
   def __init__(self, *nodes: phase_descriptor.PhaseCallableOrNodeT,
                plug_manager: Optional[PlugManager] = None,
+               child_tests: Optional[List["Test"]] = None,
+               is_child_test: bool = False,
                **metadata: Any):
     # Some sanity checks on special metadata keys we automatically fill in.
     if 'config' in metadata:
@@ -151,6 +153,8 @@ def __init__(self, *nodes: phase_descriptor.PhaseCallableOrNodeT,
     self.created_time_millis = util.time_millis()
     self.last_run_time_millis = None
     self._test_options = TestOptions()
+    self._test_options.child_tests = child_tests or []
+    self._test_options.is_child_test = is_child_test
     self._lock = threading.Lock()
     self._executor = None
     self._plug_manager = plug_manager
@@ -226,6 +230,10 @@ def state(self) -> Optional[test_state.TestState]:
         return self._executor.test_state
       return None
 
+  @property
+  def is_child_test(self) -> bool:
+      return self._test_options.is_child_test
+
   def get_option(self, option: Text) -> Any:
     return getattr(self._test_options, option)
 
@@ -404,7 +412,6 @@ def trigger_phase(test):
 
     return final_state.test_record.outcome == htf_test_record.Outcome.PASS
 
-
 @attr.s(slots=True)
 class TestOptions(object):
   """Class encapsulating various tunable knobs for Tests and their defaults.
@@ -430,6 +437,8 @@ class TestOptions(object):
   default_dut_id = attr.ib(type=Text, default='UNKNOWN_DUT')
   stop_on_first_failure = attr.ib(type=bool, default=False)
   diagnosers = attr.ib(type=List[diagnoses_lib.BaseTestDiagnoser], factory=list)
+  is_child_test = attr.ib(type=bool, default=False)
+  child_tests = attr.ib(type=List[Test], factory=list)
 
 
 @attr.s(slots=True)

openhtf/core/test_executor.py

@@ -20,20 +20,24 @@
 import sys
 import tempfile
 import threading
+import time
 import traceback
 from typing import Iterator, List, Optional, Text, Type, TYPE_CHECKING
 
 from openhtf import util
 from openhtf.core import base_plugs
 from openhtf.core import diagnoses_lib
 from openhtf.core import phase_branches
+from openhtf.core import phase_child_runner
 from openhtf.core import phase_collections
 from openhtf.core import phase_descriptor
 from openhtf.core import phase_executor
 from openhtf.core import phase_group
 from openhtf.core import phase_nodes
+from openhtf.core import test_descriptor
 from openhtf.core import test_record
 from openhtf.core import test_state
+from openhtf.core.results_collector import ResultsCollector
 from openhtf.plugs import PlugManager
 from openhtf.util import configuration
 from openhtf.util import threads
@@ -383,11 +387,48 @@ def _execute_checkpoint(self, checkpoint: phase_branches.Checkpoint,
       self.phase_executor.skip_checkpoint(checkpoint, subtest_rec)
       return _ExecutorReturn.CONTINUE
 
-    outcome = self.phase_executor.evaluate_checkpoint(checkpoint, subtest_rec)
+  def _execute_children(self, child_runner: phase_child_runner.ChildRunnerPhase,
+                          subtest_rec: Optional[test_record.SubtestRecord],
+                          in_teardown: bool) -> _ExecutorReturn:
+
+    tests = self._test_options.child_tests  # type: list[test_descriptor.Test]
+    if len(tests) == 0:
+      self.logger.warning("There are no child tests to execute.")
+      return _ExecutorReturn.CONTINUE
+
+    results = ResultsCollector()
+
+    def test_runner(child_test: test_descriptor.Test):
+      res = child_test.execute()
+      self.logger.debug(f"test uid {child_test.uid} finished, with result: {res}")
+
+    execution_threads = []
+    for test in tests:
+      test.add_output_callbacks(results.on_test_completed)
+      thread = threading.Thread(target=test_runner, args=(test,))
+      execution_threads.append(thread)
+      thread.start()
+
+    self.logger.debug('Waiting for all tests to complete.')
+    while any([not t.state.is_finalized if t.state is not None else False for t in tests]):
+      time.sleep(0.1)
+
+    for thread in execution_threads:
+      thread.join()
+
+    self.logger.debug('All child tests complete.')
+
+    outcome = phase_executor.PhaseExecutionOutcome(phase_descriptor.PhaseResult.CONTINUE)
+
+    if any(res.outcome != test_record.Outcome.PASS for res in results.get_results()):
+      if self._test_options.stop_on_first_failure:
+        outcome = phase_executor.PhaseExecutionOutcome(
+          phase_descriptor.PhaseResult.STOP)
+
     if outcome.is_terminal:
       if not self._last_outcome:
         self._last_outcome = outcome
-        self._last_execution_unit = checkpoint.name
+        self._last_execution_unit = child_runner.name
       return _ExecutorReturn.TERMINAL
 
     if outcome.is_fail_subtest:
@@ -616,6 +657,8 @@ def _execute_node(self, node: phase_nodes.PhaseNode,
       return self._execute_phase(node, subtest_rec, in_teardown)
     if isinstance(node, phase_branches.Checkpoint):
       return self._execute_checkpoint(node, subtest_rec, in_teardown)
+    if isinstance(node, phase_child_runner.ChildRunnerPhase):
+      return self._execute_children(node, subtest_rec, in_teardown)
     self.logger.error('Unhandled node type: %s', node)
     return _ExecutorReturn.TERMINAL
 
@@ -639,4 +682,4 @@ def _execute_test_diagnoser(
 
   def _execute_test_diagnosers(self) -> None:
     for diagnoser in self._test_options.diagnosers:
-      self._execute_test_diagnoser(diagnoser)
+      self._execute_test_diagnoser(diagnoser)

openhtf/output/servers/station_server.py

@@ -92,6 +92,9 @@ def _get_executing_test():
   if not tests:
     return None, None
 
+  # Filter out child tests
+  tests = [test for test in tests if not test.is_child_test]
+
   if len(tests) > 1:
     _LOG.warning('Station server does not support multiple executing tests.')