All the Things Example

import os.path
import time

import openhtf as htf
from openhtf import util
from openhtf.core import base_plugs
from openhtf.output import callbacks
from openhtf.output.callbacks import console_summary
from openhtf.output.callbacks import json_factory
from openhtf.plugs import user_input
from openhtf.util import configuration
from openhtf.util import units

CONF = configuration.CONF

# ----- Plugs (upstream: example_plugs.py) -----

EXAMPLE_PLUG_INCREMENT_SIZE = CONF.declare(
    'example_plug_increment_size', default_value=1,
    description='increment constant for example plug.')


class ExamplePlug(base_plugs.BasePlug):
    """Keeps a value and increments it. Constructed once per test."""

    def __init__(self, example_plug_increment_size):
        self.increment_size = example_plug_increment_size
        self.value = 0

    def tearDown(self):
        self.logger.info('Tearing down %s', self)

    def increment(self):
        self.value += self.increment_size
        return self.value - self.increment_size


# Bind the constructor argument to the configuration key.
example_plug_configured = configuration.bind_init_args(ExamplePlug, EXAMPLE_PLUG_INCREMENT_SIZE)


class ExampleFrontendAwarePlug(base_plugs.FrontendAwareBasePlug):
    """Calls notify_update() when its state changes so the Operator UI refreshes."""

    def __init__(self):
        super().__init__()
        self.value = 0

    def _asdict(self):
        return {'value': self.value}

    def increment(self):
        self.value += 1
        self.notify_update()


# ----- Phases -----

@htf.plug(example=example_plug_configured)
@htf.plug(frontend_aware=ExampleFrontendAwarePlug)
def example_monitor(example, frontend_aware):
    """Monitor function: sampled in the background by set_measurements."""
    time.sleep(.2)
    frontend_aware.increment()
    return example.increment()


@htf.measures(
    htf.Measurement('widget_type').matches_regex(r'.*Widget$').doc('Type of widget'),
    htf.Measurement('widget_color').doc('Color of the widget'),
    htf.Measurement('widget_size').in_range(1, 4).doc('Size of widget'))
@htf.measures('specified_as_args', docstring='Helpful docstring', units=units.HERTZ,
              validators=[util.validators.matches_regex('Measurement')])
@htf.plug(example=example_plug_configured)
@htf.plug(prompts=user_input.UserInput)
def hello_world(test, example, prompts):
    test.logger.info('Hello World!')
    test.measurements.widget_type = prompts.prompt(
        "What's the widget type? (Hint: try `MyWidget` to PASS)", text_input=True)
    if test.measurements.widget_type == 'raise':
        raise Exception()
    test.measurements.widget_color = 'Black'
    test.measurements.widget_size = 3
    test.measurements.specified_as_args = 'Measurement args specified directly'
    test.logger.info('Plug value: %s', example.increment())


@htf.PhaseOptions(timeout_s=10)
@htf.measures(*(htf.Measurement('level_%s' % i) for i in ['none', 'some', 'all']))
@htf.monitors('monitor_measurement', example_monitor)
def set_measurements(test):
    test.measurements.level_none = 0
    time.sleep(1)
    test.measurements.level_some = 8
    time.sleep(1)
    test.measurements.level_all = 9
    time.sleep(1)
    assert test.get_measurement('level_all').value == 9


@htf.measures(
    htf.Measurement('dimensions').with_dimensions(units.HERTZ),
    htf.Measurement('lots_of_dims').with_dimensions(
        units.HERTZ, units.SECOND,
        htf.Dimension(description='my_angle', unit=units.RADIAN)))
def dimensions(test):
    for dim in range(5):
        test.measurements.dimensions[dim] = 1 << dim
    for x, y, z in zip(range(1, 5), range(21, 25), range(101, 105)):
        test.measurements.lots_of_dims[x, y, z] = x + y + z


@htf.measures(
    htf.Measurement('replaced_min_only').in_range('{minimum}', 5, type=int),
    htf.Measurement('replaced_max_only').in_range(0, '{maximum}', type=int),
    htf.Measurement('replaced_min_max').in_range('{minimum}', '{maximum}', type=int),
)
def measures_with_args(test, minimum, maximum):
    del minimum, maximum  # Used by the validators, not the body.
    test.measurements.replaced_min_only = 1
    test.measurements.replaced_max_only = 1
    test.measurements.replaced_min_max = 1


def attachments(test):
    test.attach('test_attachment', 'This is test attachment data.'.encode('utf-8'))
    test.attach_from_file(os.path.join(os.path.dirname(__file__), 'example_attachment.txt'))
    assert test.get_attachment('test_attachment').data == b'This is test attachment data.'


@htf.PhaseOptions(run_if=lambda: False)
def skip_phase():
    """Never runs and leaves no record."""


def analysis(test):
    assert test.get_measurement('level_all').value == 9
    lots_of_dims = test.get_measurement('lots_of_dims')
    assert lots_of_dims.value.value == [
        (1, 21, 101, 123), (2, 22, 102, 126), (3, 23, 103, 129), (4, 24, 104, 132),
    ]
    test.logger.info('lots_of_dims as a DataFrame:\n%s', lots_of_dims.value.to_dataframe())


def teardown(test):
    test.logger.info('Running teardown')


# ----- Test -----

def make_test():
    return htf.Test(
        htf.PhaseGroup.with_teardown(teardown)(
            hello_world,
            set_measurements,
            dimensions,
            attachments,
            skip_phase,
            measures_with_args.with_args(minimum=1, maximum=4),
            analysis,
        ),
        # Metadata: stored in the record, used by mfg_inspector and by TofuPilot.
        test_name='MyTest',
        test_description='OpenHTF Example Test',
        test_version='1.0.0')


def main():
    test = make_test()
    test.add_output_callbacks(
        callbacks.OutputToFile('./{dut_id}.{metadata[test_name]}.{start_time_millis}.pickle'))
    test.add_output_callbacks(
        json_factory.OutputToJSON('./{dut_id}.{metadata[test_name]}.{start_time_millis}.json', indent=4))
    test.add_output_callbacks(console_summary.ConsoleSummary())
    test.execute(test_start=user_input.prompt_for_test_start())


if __name__ == '__main__':
    main()

$ python all_the_things.py
Enter a DUT ID in order to start the test.
--> SN1
What's the widget type? (Hint: try `MyWidget` to PASS)
--> MyWidget
MyTest:PASS

======================= test: MyTest  outcome: PASS ======================