Checkpoints & Branching
Control OpenHTF test flow with checkpoints that stop on failure and branch sequences that conditionally execute phases based on diagnosis results.
Last updated · Verified with OpenHTF 1.6.1
Stop early on failure or run phases conditionally based on diagnosis results.
A checkpoint phase inspects the outcome of the phases before it. When calibration has failed, the checkpoint stops the test so the long burn-in phase never runs.
- Test start
- phase_power
- phase_calibrate
- checkpoint("calibration")
- phase_burn_in10 min
- Test end
Not every test should run every phase. If calibration fails, there is no point running a 10-minute burn-in. OpenHTF provides two mechanisms for this: checkpoints that stop execution when previous phases fail, and branch sequences that conditionally execute phases based on diagnosis results.
Checkpoints
A checkpoint inspects the outcome of previous phases and decides whether to continue or stop. Place a checkpoint before any expensive phase you want to skip on failure.
import openhtf as htf
from openhtf.util import checkpoints
@htf.measures(
htf.Measurement('calibration_offset').in_range(-1, 1)
)
def calibration_phase(test):
test.measurements.calibration_offset = 5 # Will fail validation
def long_burn_in(test):
# This should NOT run if calibration failed
import time
test.logger.info('Starting 10-minute burn-in')
time.sleep(600)
return htf.PhaseResult.CONTINUE
def main():
test = htf.Test(
calibration_phase,
checkpoints.checkpoint('post_calibration'), # Stops if any previous phase failed
long_burn_in,
)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()Since calibration_phase fails its measurement, the checkpoint triggers a STOP and long_burn_in never runs.
Checkpoint types
checkpoint(name)strStop the test if any previous phase has a FAIL outcome. This is the most common usage.
PhaseFailureCheckpoint.last(name)strStop only if the immediately previous phase failed.
PhaseFailureCheckpoint.all_previous(name)strStop if any previous phase in the entire test failed. Same as checkpoint().
PhaseFailureCheckpoint.subtest_previous(name)strStop if any previous phase within the current subtest failed.
action=PhaseResultKeyword accepted by all three: PhaseResult.STOP (default) ends the test; PhaseResult.FAIL_SUBTEST fails only the enclosing subtest.
import openhtf as htf
from openhtf.core.phase_branches import PhaseFailureCheckpoint
def phase_a(test):
return htf.PhaseResult.FAIL_AND_CONTINUE
def phase_b(test):
return htf.PhaseResult.CONTINUE
def phase_c(test):
return htf.PhaseResult.CONTINUE
def main():
test = htf.Test(
phase_a,
phase_b,
PhaseFailureCheckpoint.last('check_b'), # Only checks phase_b → PASS, continues
phase_c,
)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()Stop on first failure
For simpler cases where you want the entire test to stop as soon as any measurement fails, use the stop_on_first_failure test option instead of checkpoints.
import openhtf as htf
@htf.measures(htf.Measurement('voltage').in_range(0, 5))
def phase_one(test):
test.measurements.voltage = 10 # Fails
@htf.measures(htf.Measurement('current').in_range(0, 1))
def phase_two(test):
# This will NOT run
test.measurements.current = 0.5
def main():
test = htf.Test(phase_one, phase_two)
test.configure(stop_on_first_failure=True)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()Subtests
Every checkpoint takes an action — PhaseResult.STOP by default. Inside a subtest, pass action=htf.PhaseResult.FAIL_SUBTEST so the checkpoint fails only that subtest and the remaining subtests still run; PhaseFailureCheckpoint.subtest_previous(name, action=...) limits the check to phases of the current subtest.
Branch sequences
Branch sequences execute phases conditionally based on diagnosis results. Combine them with diagnoses to build adaptive test flows.
import openhtf as htf
from openhtf.core import diagnoses_lib, measurements
from openhtf.core.phase_branches import BranchSequence, DiagnosisCondition
class CalDiag(diagnoses_lib.DiagResultEnum):
NEEDS_RECAL = 'needs_recal'
@diagnoses_lib.PhaseDiagnoser(CalDiag)
def cal_check(phase_record):
for m in phase_record.measurements.values():
if m.outcome == measurements.Outcome.FAIL:
return diagnoses_lib.Diagnosis(
CalDiag.NEEDS_RECAL,
description='Calibration drift detected',
)
return None
@htf.diagnose(cal_check)
@htf.measures(htf.Measurement('offset').in_range(-0.5, 0.5))
def check_calibration(test):
test.measurements.offset = 1.2 # Fails, triggers NEEDS_RECAL
def recalibrate(test):
test.logger.info('Running recalibration sequence')
return htf.PhaseResult.CONTINUE
def final_test(test):
return htf.PhaseResult.CONTINUE
def main():
test = htf.Test(
check_calibration,
BranchSequence(
DiagnosisCondition.on_any(CalDiag.NEEDS_RECAL),
recalibrate, # Only runs if NEEDS_RECAL diagnosed
),
final_test,
)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()Diagnosis checkpoints
DiagnosisCheckpoint(name, condition) stops the test (or fails the subtest) when a diagnosis condition holds — the checkpoint counterpart of a branch:
from openhtf.core.phase_branches import DiagnosisCheckpoint, DiagnosisCondition
DiagnosisCheckpoint('stop_if_defective', DiagnosisCondition.on_any(TestDiag.BOARD_DEFECTIVE))Condition types
DiagnosisCondition.on_all(*results)Execute branch only if ALL specified diagnosis results are present.
DiagnosisCondition.on_any(*results)Execute branch if ANY of the specified diagnosis results are present.
DiagnosisCondition.on_not_all(*results)Execute branch if NOT ALL of the specified results are present.
DiagnosisCondition.on_not_any(*results)Execute branch if NONE of the specified results are present.
Related
Phase Groups
Structure your OpenHTF tests with Phase Groups for guaranteed teardown execution, setup/main/teardown patterns, and nested group composition.
Subtests
Isolate groups of OpenHTF phases into subtests so one failing block does not stop the rest of the test, using htf.Subtest and PhaseResult.FAIL_SUBTEST.
