Checkpoints & Branching
Control OpenHTF test flow with checkpoints that stop on failure and branch sequences that conditionally execute phases based on diagnosis results.
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.
import openhtf as htf
from openhtf.util.checkpoints 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
Subtests isolate groups of phases so a failure in one subtest does not stop other subtests from running. Use PhaseResult.FAIL_SUBTEST to fail a subtest without stopping the entire test.
import openhtf as htf
def wifi_test(test):
test.logger.info('Testing WiFi')
return htf.PhaseResult.FAIL_SUBTEST # Fails this subtest only
def bluetooth_test(test):
test.logger.info('Testing Bluetooth')
return htf.PhaseResult.CONTINUE
def main():
test = htf.Test(
htf.Subtest('wireless_wifi', wifi_test),
htf.Subtest('wireless_bluetooth', bluetooth_test), # Still runs
)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()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()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.
Diagnoses
Use OpenHTF diagnoses to analyze results after measurement validation, classify failure modes, and drive conditional execution between phases.
Manufacturing Test Analytics
Monitor your test's first-pass yield (FPY) and capability index (Cpk) at the measurement level using TofuPilot with OpenHTF.
