Handle Test Failures and Retries

When a measurement fails in OpenHTF, the default behavior is to mark the phase as failed and continue running subsequent phases. In production, you need deliberate control: when to stop early, when to retry, and how to preserve failure data in TofuPilot for analytics.

Prerequisites

TofuPilot Python client installed: pip install tofupilot
OpenHTF installed: pip install openhtf
Basic familiarity with OpenHTF phases and measurements

Phase Failure Behavior by Default

OpenHTF phases return a PhaseResult. When a measurement fails, the phase is marked FAIL but execution continues unless you explicitly stop it.

tests/voltage_check.py

import openhtf as htf
from openhtf.util import units
from tofupilot.openhtf import TofuPilot

@htf.measures(
    htf.Measurement('rail_3v3_voltage').in_range(3.2, 3.4).with_units(units.VOLT),
    htf.Measurement('rail_5v_voltage').in_range(4.85, 5.15).with_units(units.VOLT),
)
def check_power_rails(test):
    test.measurements.rail_3v3_voltage = 3.1  # fails: below minimum
    test.measurements.rail_5v_voltage = 5.02  # passes

@htf.measures(
    htf.Measurement('uart_echo').equals('OK'),
)
def check_uart(test):
    # This phase still runs even though check_power_rails failed
    test.measurements.uart_echo = 'OK'

def main():
    test = htf.Test(check_power_rails, check_uart)
    with TofuPilot(test):
        test.execute(test_start=lambda: 'SN-001')

Both phases run. The run is uploaded to TofuPilot as FAIL with the specific failing measurement recorded.

Stop on First Failure

For hardware tests where a failed power rail makes downstream tests meaningless, stop early using PhaseResult.STOP.

tests/production_sequence.py

import openhtf as htf
from openhtf.util import units
from tofupilot.openhtf import TofuPilot

@htf.measures(
    htf.Measurement('rail_3v3_voltage').in_range(3.2, 3.4).with_units(units.VOLT),
)
def check_power_rails(test):
    test.measurements.rail_3v3_voltage = read_adc_channel(0)

    if not test.measurements.rail_3v3_voltage.is_pass:
        test.logger.error('Power rail out of spec, aborting sequence')
        return htf.PhaseResult.STOP

@htf.measures(
    htf.Measurement('uart_echo').equals('OK'),
)
def check_uart(test):
    response = send_uart_command('PING')
    test.measurements.uart_echo = response

def main():
    test = htf.Test(check_power_rails, check_uart)
    with TofuPilot(test):
        test.execute(test_start=lambda: 'SN-001')

When check_power_rails returns PhaseResult.STOP, check_uart is skipped. TofuPilot records the run as FAIL with check_uart phases marked as not executed.

Retry a Phase on Failure

Retries are useful for transient failures: communication timeouts, settling voltages, or flaky contacts. Use PhaseResult.REPEAT with a counter to avoid infinite loops.

tests/comms_test.py

import openhtf as htf
from openhtf.util import units
from tofupilot.openhtf import TofuPilot

MAX_RETRIES = 3

@htf.measures(
    htf.Measurement('i2c_device_present').equals(True),
)
def check_i2c_device(test, retries=[0]):
    found = probe_i2c_address(0x48)
    test.measurements.i2c_device_present = found

    if not found and retries[0] < MAX_RETRIES:
        retries[0] += 1
        test.logger.warning('I2C device not found, retry %d/%d', retries[0], MAX_RETRIES)
        return htf.PhaseResult.REPEAT

    retries[0] = 0  # reset for future runs

@htf.measures(
    htf.Measurement('i2c_temperature').in_range(-10, 85).with_units(units.DEGREE_CELSIUS),
)
def read_i2c_temperature(test):
    temp = read_temperature_sensor(0x48)
    test.measurements.i2c_temperature = temp

def main():
    test = htf.Test(check_i2c_device, read_i2c_temperature)
    with TofuPilot(test):
        test.execute(test_start=lambda: 'SN-002')

Using a mutable default argument (retries=[0]) persists the counter across REPEAT calls within the same phase execution. Reset it before returning to avoid stale state on the next DUT.

Retry with Backoff Using a Plug

For production lines with many fixtures, a plug keeps retry logic reusable and keeps phase code clean.

plugs/comms_plug.py

import time
import openhtf as htf
from openhtf.plugs import BasePlug

class CommPlug(BasePlug):
    def setUp(self):
        import serial
        self._port = serial.Serial('/dev/ttyUSB0', 115200, timeout=1)

    def send_with_retry(self, command, expected, attempts=3, delay=0.5):
        response = ''
        for attempt in range(attempts):
            self._port.write(f'{command}
'.encode())
            response = self._port.readline().decode().strip()
            if response == expected:
                return response, attempt + 1
            time.sleep(delay)
        return response, attempts

    def tearDown(self):
        if self._port:
            self._port.close()

tests/serial_test.py

import openhtf as htf
from tofupilot.openhtf import TofuPilot
from plugs.comms_plug import CommPlug

@htf.plug(comm=CommPlug)
@htf.measures(
    htf.Measurement('firmware_version').matches_regex(r'v\d+\.\d+\.\d+'),
    htf.Measurement('version_retry_count').in_range(minimum=1),
)
def check_firmware_version(test, comm):
    response, attempts = comm.send_with_retry('VERSION', expected='v2.1.0', attempts=3)
    test.measurements.firmware_version = response
    test.measurements.version_retry_count = attempts

def main():
    test = htf.Test(check_firmware_version)
    with TofuPilot(test):
        test.execute(test_start=lambda: 'SN-003')

Recording version_retry_count as a measurement lets TofuPilot surface retry frequency across your production fleet. A spike in retries on a specific station points to a fixture problem before it becomes a yield problem.

Clean Teardown After Failures

Plugs with a tearDown method are always called by OpenHTF, even when a phase fails or the test stops early. Use this to release hardware resources reliably.

plugs/fixture_plug.py

import openhtf as htf
from openhtf.plugs import BasePlug

class FixturePlug(BasePlug):
    def setUp(self):
        self._power_on = False
        self._relay_closed = False

    def power_on_dut(self):
        enable_power_rail()
        self._power_on = True

    def close_test_relay(self):
        close_relay(1)
        self._relay_closed = True

    def tearDown(self):
        # Always runs, even on STOP or exception
        if self._relay_closed:
            open_relay(1)
        if self._power_on:
            disable_power_rail()

tests/powered_test.py

import openhtf as htf
from openhtf.util import units
from tofupilot.openhtf import TofuPilot
from plugs.fixture_plug import FixturePlug

@htf.plug(fixture=FixturePlug)
@htf.measures(
    htf.Measurement('leakage_current').in_range(maximum=50).with_units(units.AMPERE),
)
def measure_leakage(test, fixture):
    fixture.power_on_dut()
    fixture.close_test_relay()

    current_ua = read_current_meter() * 1e6
    test.measurements.leakage_current = current_ua

    if not test.measurements.leakage_current.is_pass:
        test.logger.error('Leakage %.1f uA exceeds 50 uA limit', current_ua)
        return htf.PhaseResult.STOP
    # fixture.tearDown() called automatically by OpenHTF

def main():
    test = htf.Test(measure_leakage)
    with TofuPilot(test):
        test.execute(test_start=lambda: 'SN-004')

Failure Handling Strategy Comparison

Strategy	Use when	PhaseResult	Downstream phases
Continue on fail	Failures are independent, collect all data	(default)	Run
Stop on first fail	Downstream tests are invalid after failure	`STOP`	Skipped
Retry transient	Failure may be a fixture fluke	`REPEAT`	Run after retry
Retry with limit	Same, but cap attempts	`REPEAT` + counter	Run or STOP

Logging Failures for Analytics

TofuPilot captures every failed measurement automatically. To make failures actionable in analytics, add structured context using test.logger.

tests/full_sequence.py

import openhtf as htf
from openhtf.util import units
from tofupilot.openhtf import TofuPilot

@htf.measures(
    htf.Measurement('oscillator_freq_hz').in_range(7_990_000, 8_010_000).with_units(units.HERTZ),
)
def check_oscillator(test):
    freq = measure_frequency(channel=2)
    test.measurements.oscillator_freq_hz = freq

    if not test.measurements.oscillator_freq_hz.is_pass:
        deviation_ppm = abs(freq - 8_000_000) / 8_000_000 * 1e6
        test.logger.error(
            'Oscillator %.0f Hz (%.1f ppm from 8 MHz)',
            freq, deviation_ppm,
        )

@htf.measures(
    htf.Measurement('adc_offset_mv').in_range(-5, 5),
    htf.Measurement('adc_gain_error_pct').within_percent(1.0, 0.1),
)
def calibrate_adc(test):
    offset = measure_adc_offset()
    gain = measure_adc_gain()

    test.measurements.adc_offset_mv = offset * 1000
    test.measurements.adc_gain_error_pct = gain

def main():
    test = htf.Test(check_oscillator, calibrate_adc, name='PCB-Rev-C')
    with TofuPilot(test):
        test.execute(test_start=lambda: 'SN-005')

Log messages appear in the TofuPilot run detail view alongside the failing measurement. Quantitative context (deviation in ppm, not just pass/fail) makes root cause analysis faster when reviewing failures across a batch.