Add subtest_previous PhaseFailureCheckpoint

openhtf/core/phase_branches.py

@@ -14,13 +14,14 @@
 
 """Implements phase node branches.
 
-A BranchSequence is a phase node sequence that runs conditiionally based on the
+A BranchSequence is a phase node sequence that runs conditionally based on the
 diagnosis results of the test run.
 """
 
 import abc
 import enum
-from typing import Any, Callable, Dict, Iterator, Text, Tuple, TYPE_CHECKING, Union
+from typing import (TYPE_CHECKING, Any, Callable, Dict, Iterator, Optional,
+                    Text, Tuple, Union)
 
 import attr
 from openhtf import util
@@ -54,6 +55,9 @@ class PreviousPhases(enum.Enum):
   # Check all phases.
   ALL = 'ALL'
 
+  # Check all previous phases in the current subtest.
+  SUBTEST = "SUBTEST"
+
 
 def _not_any(iterable: Iterator[bool]) -> bool:
   return not any(iterable)
@@ -179,15 +183,20 @@ def apply_to_all_phases(
     return self
 
   def get_result(
-      self, running_test_state: 'test_state.TestState'
+    self, 
+    running_test_state: 'test_state.TestState', 
+    subtest_rec: Optional[test_record.SubtestRecord] = None
   ) -> phase_descriptor.PhaseReturnT:
-    if self._check_for_action(running_test_state):
+    if self._check_for_action(running_test_state, subtest_rec):
       return self.action
     return phase_descriptor.PhaseResult.CONTINUE
 
   @abc.abstractmethod
-  def _check_for_action(self,
-                        running_test_state: 'test_state.TestState') -> bool:
+  def _check_for_action(
+    self,
+    running_test_state: 'test_state.TestState',
+    subtest_rec: Optional[test_record.SubtestRecord] = None
+  ) -> bool:
     """Returns True when the action should be taken."""
 
   @abc.abstractmethod
@@ -197,7 +206,7 @@ def record_conditional(self) -> Union[PreviousPhases, DiagnosisCondition]:
 
 @attr.s(slots=True, frozen=True)
 class PhaseFailureCheckpoint(Checkpoint):
-  """Node that checks if a previous phase or all previous phases failed.
+  """Node that checks if the specified previous phase(s) failed.
 
   If the phases fail, this will be resolved as `action`.
 
@@ -219,6 +228,12 @@ def all_previous(cls, *args, **kwargs) -> 'PhaseFailureCheckpoint':
     kwargs['previous_phases_to_check'] = PreviousPhases.ALL
     return cls(*args, **kwargs)
 
+  @classmethod
+  def subtest_previous(cls, *args, **kwargs) -> 'PhaseFailureCheckpoint':
+    """Checks if any node in the current subtest has failed."""
+    kwargs['previous_phases_to_check'] = PreviousPhases.SUBTEST
+    return cls(*args, **kwargs)
+
   def _asdict(self) -> Dict[Text, Any]:
     ret = super(PhaseFailureCheckpoint, self)._asdict()
     ret.update(previous_phases_to_check=self.previous_phases_to_check)
@@ -228,14 +243,23 @@ def _phase_failed(self, phase_rec: test_record.PhaseRecord) -> bool:
     """Returns True if the phase_rec failed; ignores ERRORs."""
     return phase_rec.outcome == test_record.PhaseOutcome.FAIL
 
-  def _check_for_action(self,
-                        running_test_state: 'test_state.TestState') -> bool:
+  def _check_for_action(
+    self,
+    running_test_state: 'test_state.TestState',
+    subtest_rec: Optional[test_record.SubtestRecord] = None
+  ) -> bool:
     """Returns True when the specific set of phases fail."""
     phase_records = running_test_state.test_record.phases
     if not phase_records:
       raise NoPhasesFoundError('No phases found in the test record.')
     if self.previous_phases_to_check == PreviousPhases.LAST:
       return self._phase_failed(phase_records[-1])
+    elif (self.previous_phases_to_check == PreviousPhases.SUBTEST and 
+          subtest_rec is not None):
+      for phase_rec in phase_records:
+        if (phase_rec.subtest_name == subtest_rec.name and 
+            self._phase_failed(phase_rec)):
+          return True
     else:
       for phase_rec in phase_records:
         if self._phase_failed(phase_rec):
@@ -261,8 +285,11 @@ def _asdict(self) -> Dict[Text, Any]:
     ret.update(diag_condition=self.diag_condition._asdict())
     return ret
 
-  def _check_for_action(self,
-                        running_test_state: 'test_state.TestState') -> bool:
+  def _check_for_action(
+    self,
+    running_test_state: 'test_state.TestState',
+    subtest_rec: Optional[test_record.SubtestRecord] = None
+  ) -> bool:
     """Returns True if the condition is true."""
     return self.diag_condition.check(running_test_state.diagnoses_manager.store)
 

openhtf/core/phase_executor.py

@@ -362,7 +362,7 @@ def evaluate_checkpoint(
       subtest_name = None
     evaluated_millis = util.time_millis()
     try:
-      outcome = PhaseExecutionOutcome(checkpoint.get_result(self.test_state))
+      outcome = PhaseExecutionOutcome(checkpoint.get_result(self.test_state, subtest_rec))
       self.logger.debug('Checkpoint %s result: %s', checkpoint.name,
                         outcome.phase_result)
       if outcome.is_fail_subtest and not subtest_rec:

test/core/phase_branches_test.py

@@ -433,6 +433,170 @@ def test_last_fail_subtest__fail_in_subtest(self):
             evaluated_millis=htf_test.VALID_TIMESTAMP),
     ], test_rec.checkpoints)
 
+  @htf_test.yields_phases
+  def test_subtest_previous_fail__fail(self):
+    test_rec = yield htf.Test(
+        fail_phase,
+        phase0,
+        phase_branches.PhaseFailureCheckpoint.subtest_previous(
+            'subtest_previous_fail', action=htf.PhaseResult.STOP), error_phase)
+
+    self.assertTestFail(test_rec)
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.PASS, test_rec,
+                                   'phase0')
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.FAIL, test_rec,
+                                   'fail_phase')
+
+    self.assertEqual([
+        test_record.CheckpointRecord(
+            name='subtest_previous_fail',
+            action=htf.PhaseResult.STOP,
+            conditional=phase_branches.PreviousPhases.SUBTEST,
+            subtest_name=None,
+            result=phase_executor.PhaseExecutionOutcome(
+                htf.PhaseResult.STOP),
+            evaluated_millis=htf_test.VALID_TIMESTAMP),
+    ], test_rec.checkpoints)
+
+  @htf_test.yields_phases
+  def test_subtest_previous_fail__pass(self):
+    test_rec = yield htf.Test(
+        phase0,
+        phase1,
+        phase_branches.PhaseFailureCheckpoint.subtest_previous(
+            'subtest_previous_pass', action=htf.PhaseResult.STOP), phase2)
+
+    self.assertTestPass(test_rec)
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.PASS, test_rec,
+                                   'phase0', 'phase1', 'phase2')
+
+    self.assertEqual([
+        test_record.CheckpointRecord(
+            name='subtest_previous_pass',
+            action=htf.PhaseResult.STOP,
+            conditional=phase_branches.PreviousPhases.SUBTEST,
+            subtest_name=None,
+            result=phase_executor.PhaseExecutionOutcome(
+                htf.PhaseResult.CONTINUE),
+            evaluated_millis=htf_test.VALID_TIMESTAMP),
+    ], test_rec.checkpoints)
+
+  @htf_test.yields_phases
+  def test_subtest_previous_fail__fail_in_subtest(self):
+    test_rec = yield htf.Test(
+        phase0,
+        htf.Subtest(
+          'sub', fail_phase, phase1, 
+          phase_branches.PhaseFailureCheckpoint.subtest_previous(
+            'subtest_previous_fail_in_subtest', action=htf.PhaseResult.STOP), 
+        ),
+        error_phase)
+
+    self.assertTestFail(test_rec)
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.PASS, test_rec,
+                                   'phase0', 'phase1')
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.FAIL, test_rec,
+                                   'fail_phase')
+
+    self.assertEqual([
+        test_record.CheckpointRecord(
+            name='subtest_previous_fail_in_subtest',
+            action=htf.PhaseResult.STOP,
+            conditional=phase_branches.PreviousPhases.SUBTEST,
+            subtest_name='sub',
+            result=phase_executor.PhaseExecutionOutcome(
+                htf.PhaseResult.STOP),
+            evaluated_millis=htf_test.VALID_TIMESTAMP),
+    ], test_rec.checkpoints)
+
+  @htf_test.yields_phases
+  def test_subtest_previous_fail__fail_out_of_subtest(self):
+    test_rec = yield htf.Test(
+        fail_phase,
+        htf.Subtest(
+          'sub', phase0, 
+          phase_branches.PhaseFailureCheckpoint.subtest_previous(
+            'subtest_previous_fail_out_of_subtest', action=htf.PhaseResult.STOP), 
+          phase1,
+        ),
+        phase2)
+
+    self.assertTestFail(test_rec)
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.PASS, test_rec,
+                                   'phase0', 'phase1', 'phase2')
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.FAIL, test_rec,
+                                   'fail_phase')
+
+    self.assertEqual([
+        test_record.CheckpointRecord(
+            name='subtest_previous_fail_out_of_subtest',
+            action=htf.PhaseResult.STOP,
+            conditional=phase_branches.PreviousPhases.SUBTEST,
+            subtest_name='sub',
+            result=phase_executor.PhaseExecutionOutcome(
+                htf.PhaseResult.CONTINUE),
+            evaluated_millis=htf_test.VALID_TIMESTAMP),
+    ], test_rec.checkpoints)
+
+  @htf_test.yields_phases
+  def test_subtest_previous_fail__pass_in_subtest(self):
+    test_rec = yield htf.Test(
+        phase0,
+        htf.Subtest(
+          'sub', phase1, 
+          phase_branches.PhaseFailureCheckpoint.subtest_previous(
+            'subtest_previous_pass_in_subtest', action=htf.PhaseResult.STOP),
+          phase2,
+        ), phase3)
+
+    self.assertTestPass(test_rec)
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.PASS, test_rec,
+                                   'phase0', 'phase1', 'phase2', 'phase3')
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.FAIL, test_rec,
+                                   'fail_phase')
+
+    self.assertEqual([
+        test_record.CheckpointRecord(
+            name='subtest_previous_pass_in_subtest',
+            action=htf.PhaseResult.STOP,
+            conditional=phase_branches.PreviousPhases.SUBTEST,
+            subtest_name='sub',
+            result=phase_executor.PhaseExecutionOutcome(
+                htf.PhaseResult.CONTINUE),
+            evaluated_millis=htf_test.VALID_TIMESTAMP),
+    ], test_rec.checkpoints)
+
+  @htf_test.yields_phases
+  def test_subtest_previous_fail_subtest__fail_in_subtest(self):
+    test_rec = yield htf.Test(
+        phase0,
+        htf.Subtest(
+          'sub', fail_phase, phase1, 
+          phase_branches.PhaseFailureCheckpoint.subtest_previous(
+            'subtest_previous_fail_subtest_in_subtest', action=htf.PhaseResult.FAIL_SUBTEST), 
+          skip0,
+        ),
+        phase2)
+
+    self.assertTestFail(test_rec)
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.PASS, test_rec,
+                                   'phase0', 'phase1', 'phase2')
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.FAIL, test_rec,
+                                   'fail_phase')
+    self.assertPhasesOutcomeByName(test_record.PhaseOutcome.SKIP, test_rec, 'skip0')
+
+    self.assertEqual([
+        test_record.CheckpointRecord(
+            name='subtest_previous_fail_subtest_in_subtest',
+            action=htf.PhaseResult.FAIL_SUBTEST,
+            conditional=phase_branches.PreviousPhases.SUBTEST,
+            subtest_name='sub',
+            result=phase_executor.PhaseExecutionOutcome(
+                htf.PhaseResult.FAIL_SUBTEST),
+            evaluated_millis=htf_test.VALID_TIMESTAMP),
+    ], test_rec.checkpoints)
+
+
   @htf_test.yields_phases
   def test_all__no_previous_phases(self):
     self.test_start_function = None