Factory Test Framework

Build interactive production test sequences with operator UI

Create step-based production test sequences with interactive operator UI, data persistence, and result reporting.

Import

from lager.factory import Step, Failure, run, hash_file, get_secret
from lager.cache import PersistentCache

Overview

The factory test framework provides:

• Step-based test execution with automatic sequencing
• Interactive UI elements (buttons, text input, selections)
• State management and data persistence between steps
• Error handling with automatic stop-on-fail behavior
• Result streaming to external systems

Classes

Class	Description
Step	Base class for defining test steps
Failure	Exception class for test failures
PersistentCache	Data persistence between test runs

Functions

Function	Description
run()	Execute a sequence of test steps
hash_file()	Calculate SHA1 hash of a file
get_secret()	Get environment variable secret

Step Class

Creating a Step

Create test steps by subclassing Step and implementing the run() method:

from lager.factory import Step, Failure

class MyTestStep(Step):
    """Example test step."""

    def run(self):
        # Access shared state
        self.log("Starting test...")

        # Perform test
        result = self.do_some_test()

        if not result:
            raise Failure("Test failed: result was invalid")

        # Store data for later steps
        self.state['test_result'] = result

        # Return True for pass, False for fail
        return True

Step Properties

Property	Type	Description
state	dict	Shared state dictionary between steps
StopOnFail	bool	If True (default), stop sequence on failure

Step Methods

log(message, file=sys.stdout)

Log a message to stdout or stderr.

def run(self):
    self.log("Processing...")
    self.log("Warning!", file=sys.stderr)

present_pass_fail_buttons(*, timeout=30)

Display Pass/Fail buttons and wait for operator response. Note: timeout is keyword-only.

def run(self):
    self.log("Check LED is blinking")
    result = self.present_pass_fail_buttons(timeout=60)
    return result  # True for Pass, False for Fail

Returns: bool - True if Pass clicked, False if Fail clicked

present_buttons(buttons, *, timeout=30)

Display custom buttons and wait for selection. Note: timeout is keyword-only.

def run(self):
    choice = self.present_buttons([
        ('Option A', 'value_a'),
        ('Option B', 'value_b'),
        ('Option C', 'value_c'),
    ], timeout=30)

    self.state['user_choice'] = choice
    return True

Parameters:

Parameter	Type	Description
buttons	list	List of (label, value) tuples
timeout	float	Timeout in seconds

Returns: Selected value from tuple

present_text_input(prompt, size=50, *, timeout=30)

Display a text input field. Note: timeout is keyword-only.

def run(self):
    serial = self.present_text_input("Enter serial number:", size=20)
    self.state['serial_number'] = serial
    return True

Parameters:

Parameter	Type	Description
prompt	str	Input prompt text
size	int	Input field size (default: 50)
timeout	float	Timeout in seconds

Returns: str - User-entered text

present_select(label, choices, allow_multiple=False, *, timeout=30)

Display a dropdown selection. Note: timeout is keyword-only.

def run(self):
    hw_rev = self.present_select("Select hardware revision:", [
        ('Rev A', 'rev_a'),
        ('Rev B', 'rev_b'),
        ('Rev C', 'rev_c'),
    ])
    self.state['hw_revision'] = hw_rev
    return True

Parameters:

Parameter	Type	Description
label	str	Selection label
choices	list	List of (label, value) tuples
allow_multiple	bool	Allow multiple selections
timeout	float	Timeout in seconds

Returns: Selected value(s)

present_checkboxes(label, choices, *, timeout=30)

Display checkboxes for multiple selection. Note: timeout is keyword-only.

def run(self):
    options = self.present_checkboxes("Select failed components:", [
        ('LED', 'led'),
        ('Button', 'button'),
        ('Display', 'display'),
    ])
    self.state['failed_components'] = options
    return True

present_radios(label, choices, *, timeout=30)

Display radio buttons for single selection. Note: timeout is keyword-only.

def run(self):
    failure_mode = self.present_radios("Select failure mode:", [
        'No power',
        'Intermittent',
        'Complete failure',
    ])
    return True

update_heading(text)

Update the step heading text.

def run(self):
    self.update_heading("Testing Power Supply...")
    # ... perform test ...
    self.update_heading("Power Supply Test Complete")
    return True

update_image(filename)

Display an image to the operator.

def run(self):
    self.update_image("/path/to/test_setup.png")
    result = self.present_pass_fail_buttons()
    return result

hide_image()

Hide the currently displayed image.

def run(self):
    self.hide_image()

present_link(url, text=None)

Display a clickable link.

def run(self):
    self.present_link("https://docs.example.com/setup", "Setup Instructions")

StopOnFail Behavior

By default, the test sequence stops when a step fails. Override this behavior:

class NonCriticalStep(Step):
    StopOnFail = False  # Continue even if this step fails

    def run(self):
        # This step's failure won't stop the sequence
        return self.some_optional_test()

Failure Exception

Use Failure to indicate a test failure with a message:

from lager.factory import Step, Failure

class VoltageTestStep(Step):
    def run(self):
        voltage = self.measure_voltage()

        if voltage < 3.0:
            raise Failure(f"Voltage too low: {voltage}V (expected >= 3.0V)")

        if voltage > 3.6:
            raise Failure(f"Voltage too high: {voltage}V (expected <= 3.6V)")

        self.log(f"Voltage OK: {voltage}V")
        return True

run() Function

Execute a sequence of test steps:

from lager.factory import run

# Define your step classes
steps = [
    InitializeStep,
    PowerOnStep,
    VoltageTestStep,
    FunctionalTestStep,
    CleanupStep,
]

# Run the test sequence
run(steps)

# Or with a finalizer step (always runs)
run(steps, finalizer_cls=FinalReportStep)

Parameters:

Parameter	Type	Description
steps	list	List of Step subclasses
finalizer_cls	class	Optional Step class that always runs

State Management

Shared State

Steps share a state dictionary:

class Step1(Step):
    def run(self):
        self.state['serial'] = '12345'
        return True

class Step2(Step):
    def run(self):
        serial = self.state['serial']  # Access from previous step
        self.log(f"Testing device {serial}")
        return True

Persistent Cache

For data that persists between test runs:

from lager.cache import PersistentCache

class CalibrateStep(Step):
    def run(self):
        cache = PersistentCache()

        # Check for existing calibration
        cal_data = cache.load('calibration_data')
        if cal_data:
            self.log("Using cached calibration")
            return True

        # Perform calibration
        cal_data = self.perform_calibration()
        cache.store('calibration_data', cal_data)
        return True

Helper Functions

hash_file(path)

Calculate SHA1 hash of a file:

from lager.factory import hash_file

firmware_hash = hash_file("/path/to/firmware.bin")
print(f"Firmware hash: {firmware_hash}")

get_secret(name, default=None)

Get a secret from environment variables (prefixed with LAGER_SECRET_):

from lager.factory import get_secret

api_key = get_secret('API_KEY')  # Gets LAGER_SECRET_API_KEY
db_pass = get_secret('DB_PASSWORD', default='testing')

Examples

Complete Production Test

from lager.factory import Step, Failure, run
from lager import Net, NetType

class InitializeStep(Step):
    """Initialize test equipment and get serial number."""

    def run(self):
        self.update_heading("Initializing Test Station")

        # Get serial number from operator
        serial = self.present_text_input("Scan device serial number:")
        self.state['serial'] = serial
        self.log(f"Testing device: {serial}")

        return True


class PowerOnStep(Step):
    """Apply power and verify voltage."""

    def run(self):
        self.update_heading("Power On Test")

        # Get power supply net
        psu = Net.get('DUT_POWER', type=NetType.PowerSupply)
        psu.set_voltage(3.3)
        psu.set_current(0.5)
        psu.enable()

        # Store for cleanup
        self.state['psu'] = psu

        # Read voltage
        import time
        time.sleep(0.5)
        voltage = psu.voltage()

        if voltage < 3.2 or voltage > 3.4:
            raise Failure(f"Voltage out of range: {voltage}V")

        self.log(f"Power OK: {voltage}V")
        return True


class CurrentTestStep(Step):
    """Verify current consumption."""

    def run(self):
        self.update_heading("Current Consumption Test")

        psu = self.state['psu']
        current = psu.current()

        # Check current limits
        if current < 0.010:  # 10mA minimum
            raise Failure(f"Current too low: {current*1000:.1f}mA")

        if current > 0.100:  # 100mA maximum
            raise Failure(f"Current too high: {current*1000:.1f}mA")

        self.log(f"Current OK: {current*1000:.1f}mA")
        self.state['current_ma'] = current * 1000
        return True


class LEDTestStep(Step):
    """Visual LED verification."""
    StopOnFail = False  # Non-critical

    def run(self):
        self.update_heading("LED Verification")
        self.update_image("/etc/lager/images/led_pattern.png")

        self.log("Verify LED is blinking green")
        result = self.present_pass_fail_buttons(timeout=60)

        self.hide_image()
        return result


class FunctionalTestStep(Step):
    """Run functional tests."""

    def run(self):
        self.update_heading("Functional Test")

        # GPIO test
        gpio = Net.get('TEST_PIN', type=NetType.GPIO)

        # Toggle and verify
        gpio.output(1)
        if gpio.input() != 1:
            raise Failure("GPIO stuck low")

        gpio.output(0)
        if gpio.input() != 0:
            raise Failure("GPIO stuck high")

        self.log("GPIO test passed")
        return True


class CleanupStep(Step):
    """Clean up and report results."""
    StopOnFail = False  # Always try to clean up

    def run(self):
        self.update_heading("Cleanup")

        # Disable power supply
        if 'psu' in self.state:
            self.state['psu'].disable()
            self.log("Power supply disabled")

        return True


class FinalReportStep(Step):
    """Generate final report - always runs."""
    StopOnFail = False

    def run(self):
        serial = self.state.get('serial', 'UNKNOWN')
        failed_step = self.state.get('FAILED_STEP', None)

        if failed_step:
            self.log(f"FAILED at step: {failed_step}")
            self.log(f"Device {serial}: FAIL")
        else:
            self.log(f"Device {serial}: PASS")

        return True


# Run the test sequence
if __name__ == "__main__":
    steps = [
        InitializeStep,
        PowerOnStep,
        CurrentTestStep,
        LEDTestStep,
        FunctionalTestStep,
        CleanupStep,
    ]

    run(steps, finalizer_cls=FinalReportStep)

Interactive Configuration Step

class ConfigurationStep(Step):
    """Configure test parameters."""

    def run(self):
        self.update_heading("Test Configuration")

        # Select test mode
        mode = self.present_select("Select test mode:", [
            ('Full Test', 'full'),
            ('Quick Test', 'quick'),
            ('Debug Mode', 'debug'),
        ])
        self.state['test_mode'] = mode

        # Select hardware revision
        hw_rev = self.present_radios("Hardware revision:", [
            'Rev A',
            'Rev B',
            'Rev C',
        ])
        self.state['hw_revision'] = hw_rev

        # Optional tests
        options = self.present_checkboxes("Additional tests:", [
            ('Stress Test', 'stress'),
            ('EMC Pre-scan', 'emc'),
            ('Extended Burn-in', 'burnin'),
        ])
        self.state['optional_tests'] = options

        self.log(f"Configuration: mode={mode}, hw={hw_rev}")
        return True

Calibration with Cache

class CalibrationStep(Step):
    """Perform or load calibration."""

    def run(self):
        self.update_heading("Calibration")

        cache = self.state['cache']

        # Check for recent calibration
        import json
        import time
        data = cache.load('last_calibration')
        if data:
            last_cal = json.loads(data.decode())
            age_hours = (time.time() - last_cal['timestamp']) / 3600

            if age_hours < 24:
                self.log(f"Using cached calibration ({age_hours:.1f} hours old)")
                self.state['cal_offset'] = last_cal['offset']
                return True

        # Perform new calibration
        self.log("Performing calibration...")

        # Measure reference
        adc = Net.get('CAL_REF', type=NetType.ADC)
        measured = adc.input()
        expected = 2.500  # Reference voltage
        offset = expected - measured

        # Store calibration
        cal_data = {
            'timestamp': time.time(),
            'offset': offset,
        }
        cache.store('last_calibration', json.dumps(cal_data).encode())

        self.state['cal_offset'] = offset
        self.log(f"Calibration offset: {offset:.4f}V")
        return True

Notes

• Steps execute in order; sequence stops on first failure (unless StopOnFail = False)
• The finalizer step always runs, regardless of test pass/fail status
• State dictionary is shared between all steps in a sequence
• UI methods block until operator interaction or timeout
• Timeout raises TimeoutError which is handled as a step failure
• FAILED_STEP is automatically set in state when a step fails
• Cache data persists across test runs (stored on disk)
• The framework integrates with Lager’s streaming result system