Reference¶

Test Plan¶

class spintop_openhtf.TestPlan(name: str = 'testplan', store_result: bool = True, store_location: Union[str, Callable] = None)¶

The core spintop-openhtf interface to create an openhtf sequence.

TestPlan simplifies the creating of openhtf sequences. Instead of declaring functions and adding them as an array to openhtf, the test plan allows declarative addition of function using decorators.

The TestPlan is itself a TestSequence and therefore implements all methods defined there.

In OpenHTF, you would do the following:

import openhtf as htf

@htf.plug(my_plug=...)
def example_test(test, my_plug):
    time.sleep(.2)

@htf.plug(my_plug=...)
def example_test2(test, my_plug):
    time.sleep(.2)

test = htf.Test(example_test, example_test2)
test.run() # Run once only

With the TestPlan, you can do the following (equivalent):

from spintop_openhtf import TestPlan

plan = TestPlan('my-test-name')

@plan.testcase('Test1')
@plan.plug(my_plug=...)
def example_test(test, my_plug):
    time.sleep(.2)

@plan.testcase('Test2')
@plan.plug(my_plug=...)
def example_test2(test, my_plug):
    time.sleep(.2)

plan.run_once()

Parameters

name – The name of the test plan. Used to identify the ‘type’ of the test.
store_result – Whether results should automatically be stored as JSON in the spintop_openhtf site directory.
store_location – When store_result is True, where test results will be stored. This can either be a callable function that receives the test record attributes as kwargs arguments (e.g. fn(**test_record)) or a string that will be formatted with the test record attributes (e.g. ‘{metadata[test_name]}/{outcome}’). This uses LocalStorageOutput, which instead of only writing the result as a JSON file as OpenHTF does, writes that json file as result.json in the folder and writes all test-attached files next to it.

add_callbacks(*callbacks)¶

Add custom callbacks to the underlying openhtf test.

Parameters: callbacks – The callbacks to add.

execute()¶: Execute the configured test using the test_start function as a trigger.

property execute_test¶: Returns a function that takes no arguments and that executes the test described by this test plan.

property history_path¶: The base path where results of the tests defined by this object are stored.

image_url(url: str) → str¶

Creates a temporary hosted image based on the specified file path.

Parameters: url – The image path.
Returns: The temporary url associated with this image. Can be used in custom forms to show images.

property is_runnable¶: Whether this test plan contains any runnable phases.

no_trigger()¶

Removes the need for a trigger and removes the default trigger.

plan = TestPlan('my-plan')
plan.no_trigger()

# ...

plan.run() # Will start immediately.

run(launch_browser=True, once=False)¶

Run this test with the OpenHTF frontend.

Requires the [server] extra to work. If you do not need the server, you should use run_console() which disables completely the GUI server and increases execution speed therefore.

Parameters

launch_browser – When True, a browser page will open automatically when this is called.
once – When False, the test will run in a loop; i.e. when a test ends a new one will start immediately.

run_console(once=False)¶

Run this test without a frontend server and enables console input.

To enable the server, use run() instead.

Parameters: once – When False, the test will run in a loop; i.e. when a test ends a new one will start immediately.

run_once(launch_browser=True)¶: Shortcut for run() with once=True.

trigger(name: str, **options)¶

Decorator factory for the trigger phase.

Similar to testcase(), except that this function will be used as the test trigger.

The test trigger is a special test phase that is executed before test officialy start. Once this phase is complete, the test will start. Usually used to configure the test with the DUT id for example.

property trigger_phase¶: Returns the defined

class spintop_openhtf.TestSequence(name)¶

The base sequence object: defines a sequence of setup, test and teardown openhtf phases.

TestSequences can be nested together in two different ways:

Option 1.

If the sequence should be re-usable, create an explicit TestSequence object and add setup(), testcase() and teardown() phases using the decorator methods. Afterwards, add it to a specific TestPlan (or sub TestSequence) using append():

from spintop_openhtf import TestSequence, TestPlan

my_sequence = TestSequence('my-sequence')

@my_sequence.testcase('Test1')
def do_something(test):
    pass

# Elsewhere probably

plan = TestPlan('my-plan')
plan.append(my_sequence)

Option 2.

If the sequence is defined simply to nest phases inside the same test plan, the sub_sequence() simplifies the declaration and usage of a sub sequence. This code block is equivalent to the previous one, but does not allow re-usage of the sub_sequence object:

from spintop_openhtf import TestPlan

plan = TestPlan('my-plan')

my_sequence = plan.sub_sequence('my-sequence')

@my_sequence.testcase('Test1')
def do_something(test):
    pass

Since TestPlans are TestSequences, and sub sequences are also TestSequences, they can be infinitely nested using append() and sub_sequence().

Parameters: name – The name of this test node.

append(*phases)¶

Append normal phases (or sequences) to this test plan.

Parameters: phases – The phases to append.

measures(*args, **kwargs)¶: Helper method: shortcut to openhtf.measures()

plug(*args, **kwargs)¶: Helper method: shortcut to openhtf.plug()

setup(name, **options)¶

Decorator factory for a setup function.

A setup function will executed if the sequence is entered. All setup functions are executed before the testcases, regardless of the order of declaration.

See testcase() for usage.

sub_sequence(name)¶

Create new empty TestSequence and append it to this sequence.

The following two snippets are equivalent:

my_sequence = TestSequence('Parent')
sub_sequence = my_sequence.sub_sequence('Child')

my_sequence = TestSequence('Parent')
sub_sequence = TestSequence('Child')
my_sequence.append(sub_sequence)

teardown(name, **options)¶

Decorator factory for a teardown phase.

A teardown function will always be executed if the sequence is entered, regardless of the outcome of normal or setup phases.

See testcase() for usage.

testcase(name, **options)¶

Decorator factory for a normal phase.

A testcase function is a normal openhtf phase.

The options parameter is a proxy for the PhaseOptions arguments.
my_sequence = TestSequence('Parent')

@my_sequence.testcase('my-testcase-name')
def setup_fn(test):
    (...)
Options used to override default test phase behaviors.

name¶: Override for the name of the phase. Can be formatted in several different ways as defined in util.format_string.

timeout_s¶: Timeout to use for the phase, in seconds.

run_if¶: Callback that decides whether to run the phase or not; if not run, the phase will also not be logged. Optionally, this callback may take a single parameter: the test.state dictionnary. This allows dynamic test selection based on variables in the user defined state.

requires_state¶: If True, pass the whole TestState into the first argument, otherwise only the TestApi will be passed in. This is useful if a phase needs to wrap another phase for some reason, as PhaseDescriptors can only be invoked with a TestState instance.

repeat_limit¶: Maximum number of repeats. None indicates a phase will be repeated infinitely as long as PhaseResult.REPEAT is returned.

run_under_pdb¶: If True, run the phase under the Python Debugger (pdb). When setting this option, increase the phase timeout as well because the timeout will still apply when under the debugger.

Example Usages:

@PhaseOptions(timeout_s=1) def PhaseFunc(test):

pass

@PhaseOptions(name=’Phase({port})’) def PhaseFunc(test, port, other_info):

pass

Configuration¶

Config Module¶

A singleton class to replace the ‘conf’ module.

This class provides the configuration interface described in the module docstring. All attribuets/methods must not begin with a lowercase letter so as to avoid naming conflicts with configuration keys.

exception openhtf.util.conf.ConfigurationInvalidError¶: Indicates the configuration format was invalid or couldn’t be read.

class openhtf.util.conf.Declaration¶: Record type encapsulating information about a config declaration.

exception openhtf.util.conf.InvalidKeyError¶: Raised when an invalid key is declared or accessed.

exception openhtf.util.conf.KeyAlreadyDeclaredError¶: Indicates that a configuration key was already declared.

exception openhtf.util.conf.UndeclaredKeyError¶: Indicates that a key was required but not predeclared.

exception openhtf.util.conf.UnsetKeyError¶: Raised when a key value is requested but we have no value for it.

openhtf.util.conf.declare()¶

Declare a configuration key with the given name.

Parameters

name – Configuration key to declare, must not have been already declared.
description – If provided, use this as the description for this key.
**kwargs – Other kwargs to pass to the Declaration, only default_value is currently supported.

openhtf.util.conf.inject_positional_args()¶

Decorator for injecting positional arguments from the configuration.

This decorator wraps the given method, so that any positional arguments are passed with corresponding values from the configuration. The name of the positional argument must match the configuration key.

Keyword arguments are NEVER modified, even if their names match configuration keys. Avoid naming keyword args names that are also configuration keys to avoid confusion.

Additional positional arguments may be used that do not appear in the configuration, but those arguments MUST be specified as keyword arguments upon invocation of the method. This is to avoid ambiguity in which positional arguments are getting which values.

Parameters: method – The method to wrap.
Returns: A wrapper that, when invoked, will call the wrapped method, passing in

configuration values for positional arguments.

openhtf.util.conf.load()¶: load configuration values from kwargs, see load_from_dict().

openhtf.util.conf.load_flag_values()¶

Load flag values given from command line flags.

Parameters: flags – An argparse Namespace containing the command line flags.

openhtf.util.conf.load_from_dict()¶

Loads the config with values from a dictionary instead of a file.

This is meant for testing and bin purposes and shouldn’t be used in most applications.

Parameters

dictionary – The dictionary containing config keys/values to update.
_override – If True, new values will override previous values.
_allow_undeclared – If True, silently load undeclared keys, otherwise warn and ignore the value. Typically used for loading config files before declarations have been evaluated.

openhtf.util.conf.load_from_file()¶

Loads the configuration from a file.

Parsed contents must be a single dict mapping config key to value.

Parameters

yamlfile – The opened file object to load configuration from.
load_from_dict() for other args' descriptions. (See) –

Raises

ConfigurationInvalidError – If configuration file can’t be read, or can’t be parsed as either YAML (or JSON, which is a subset of YAML).

openhtf.util.conf.load_from_filename()¶: Opens the filename and calls load_from_file.

openhtf.util.conf.reset()¶

Reset the loaded state of the configuration to what it was at import.

Note that this does not reset values set by commandline flags or loaded from –config-file (in fact, any values loaded from –config-file that have been overridden are reset to their value from –config-file).

openhtf.util.conf.save_and_restore()¶

Decorator for saving conf state and restoring it after a function.

This decorator is primarily for use in tests, where conf keys may be updated for individual test cases, but those values need to be reverted after the test case is done.

Examples

conf.declare(‘my_conf_key’)

@conf.save_and_restore def MyTestFunc():

conf.load(my_conf_key=’baz’) SomeFuncUnderTestThatUsesMyConfKey()

conf.load(my_conf_key=’foo’) MyTestFunc() print conf.my_conf_key # Prints ‘foo’, NOT ‘baz’

# Without the save_and_restore decorator, MyTestFunc() would have had the # side effect of altering the conf value of ‘my_conf_key’ to ‘baz’.

# Config keys can also be initialized for the context inline at decoration # time. This is the same as setting them at the beginning of the # function, but is a little clearer syntax if you know ahead of time what # config keys and values you need to set.

@conf.save_and_restore(my_conf_key=’baz’) def MyOtherTestFunc():

print conf.my_conf_key # Prints ‘baz’

MyOtherTestFunc() print conf.my_conf_key # Prints ‘foo’ again, for the same reason.

Parameters

_func – The function to wrap. The returned wrapper will invoke the function and restore the config to the state it was in at invocation.
**config_values – Config keys can be set inline at decoration time, see examples. Note that config keys can’t begin with underscore, so there can be no name collision with _func.

Returns

Wrapper to replace _func, as per Python decorator semantics.

Built-in Configuration Keys¶

class spintop_openhtf.testplan._default_conf.ConfigHelpText¶

plug_teardown_timeout_s¶

Timeout (in seconds) for each plug tearDown function if > 0; otherwise, will wait an unlimited time.

Default Value= 0

allow_unset_measurements¶

If True, unset measurements do not cause Tests to FAIL.

Default Value= False

station_id¶

The name of this test station

Default Value= ‘build-12175381-project-581279-spintop-openhtf’

cancel_timeout_s¶

Timeout (in seconds) when the test has been cancelledto wait for the running phase to exit.

Default Value= 2

stop_on_first_failure¶

Stop current test execution and return Outcome FAILon first phase with failed measurement.

Default Value= False

capture_source¶

Whether to capture the source of phases and the test module. This defaults to False since this potentially reads many files and makes large string copies. If True, will capture docstring also. Set to ‘true’ if you want to capture your test’s source.

Default Value= False

capture_docstring¶

Whether to capture the docstring of phases and the test module. If True, will capture docstring. Set to ‘true’ if you want to capture your test’s docstring.

Default Value= False

teardown_timeout_s¶

Default timeout (in seconds) for test teardown functions; this option is deprecated and only applies to the deprecated Test level teardown function.

Default Value= 30

user_input_enable_console¶

If True, enables user input collection in console prompt.

Default Value= True

frontend_throttle_s¶

Min wait time between successive updates to the frontend.

Default Value= 0.15

station_server_port¶

Port on which to serve the app. If set to zero (the default) then an arbitrary port will be chosen.

Default Value= 0

station_discovery_address¶

(no description)

Default Value= None

station_discovery_port¶

(no description)

Default Value= None

station_discovery_ttl¶

(no description)

Default Value= None

Plugs¶

Base Interface¶

class spintop_openhtf.plugs.base.UnboundPlug¶

A generic interface base class that allows intelligent creation of OpenHTF plugs without limiting its usage to OpenHTF.

logger¶: If inside an OpenHTF plug, this is the OpenHTF provided logger. If not, it is a logger with the object ID as name.

classmethod as_plug(name, **kwargs_values)¶

Create a bound plug that will retrieve values from conf or the passed values here.

Take SSHInterface for example.

from spintop_openhtf.plugs import from_conf
from spintop_openhtf.plugs.ssh import SSHInterface

MySSHInterface = SSHInterface.as_plug(
    'MySSHInterface', # The name of this plug as it will appear in logs
    addr=from_conf( # from_conf will retrieve the conf value named like this.
        'my_ssh_addr',
        description="The addr of my device."
    ),
    username='x', # Always the same
    password='y'  # Always the same
)

close()¶: Abstract method: Close resources related to this interface.

close_log()¶: Close all logger handlers.

log_to_filename(filename, **kwargs)¶: Create and add to logger a FileHandler that logs to filename.

log_to_stream(stream=None, **kwargs)¶: Create and add to logger a StreamHandler that streams to stream

open(*args, **kwargs)¶: Abstract method: Open resources related to this interface.

tearDown()¶: Tear down the plug instance. This is part of the OpenHTF Plug contract

COM Port Interface¶

class spintop_openhtf.plugs.comport.ComportInterface(comport, baudrate=115200)¶

An interface to a comport.

Allows reading and writing. A background thread reads any data that comes in and those lines can be accessed using the next_line function.

classmethod as_plug(name, **kwargs_values)¶

Create a bound plug that will retrieve values from conf or the passed values here.

Take SSHInterface for example.

from spintop_openhtf.plugs import from_conf
from spintop_openhtf.plugs.ssh import SSHInterface

MySSHInterface = SSHInterface.as_plug(
    'MySSHInterface', # The name of this plug as it will appear in logs
    addr=from_conf( # from_conf will retrieve the conf value named like this.
        'my_ssh_addr',
        description="The addr of my device."
    ),
    username='x', # Always the same
    password='y'  # Always the same
)

clear_lines()¶: Clear all lines in the buffer.

close()¶: Attempts to close the serial port if it exists.

close_log()¶: Close all logger handlers.

com_target(*args, **kwargs)¶: Alias for message_target

connection_lost(exc)¶

Called when the connection is lost or closed.

The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).

connection_made(transport)¶

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

data_received(data)¶

Called when some data is received.

The argument is a bytes object.

eof_received()¶

Called when the other end calls write_eof() or equivalent.

If this returns a false value (including None), the transport will close itself. If it returns a true value, closing the transport is up to the protocol.

execute_command(command=None, timeout=None, target=None)¶: Adds the self.eol to command and call message_target using either target as target or self._target if defined. This is used for executing commands in a shell-like environment and to wait for the prompt. self._target or target should be the expected prompt.

keep_lines(lines_to_keep)¶: Clear all lines in the buffer except the last lines_to_keep lines.

log_to_filename(filename, **kwargs)¶: Create and add to logger a FileHandler that logs to filename.

log_to_stream(stream=None, **kwargs)¶: Create and add to logger a StreamHandler that streams to stream

message_target(message, target, timeout=None, keeplines=0, _timeout_raises=True)¶

Sends the message string and waits for any string in target to be received.

Parameters

message – The string to write into the io interface
target – A string or a list of string to check for in the read lines.
timeout – (default=None, no timeout) Wait up to this seconds for the targets. If busted, this will raise a IOTargetTimeout.
keeplines – (default=0, discard all) Before sending the message, call self.keep_lines with this value.
_timeout_raises – (default=True) If True, a timeout will raise an error.

next_line(timeout=10)¶: Waits up to timeout seconds and return the next line available in the buffer.

open(_serial=None)¶

Opens the serial port using the comport and baudrate object attributes.

Parameters: _serial – Optionnal underlying serial.Serial object to use. Used for mock testing.

pause_writing()¶

Called when the transport’s buffer goes over the high-water mark.

Pause and resume calls are paired – pause_writing() is called once when the buffer goes strictly over the high-water mark (even if subsequent writes increases the buffer size even more), and eventually resume_writing() is called once when the buffer size reaches the low-water mark.

Note that if the buffer size equals the high-water mark, pause_writing() is not called – it must go strictly over. Conversely, resume_writing() is called when the buffer size is equal or lower than the low-water mark. These end conditions are important to ensure that things go as expected when either mark is zero.

NOTE: This is the only Protocol callback that is not called through EventLoop.call_soon() – if it were, it would have no effect when it’s most needed (when the app keeps writing without yielding until pause_writing() is called).

resume_writing()¶

Called when the transport’s buffer drains below the low-water mark.

See pause_writing() for details.

tearDown()¶: Tear down the plug instance.

write(string)¶: Write the string into the io interface.

SSH Interface¶

class spintop_openhtf.plugs.ssh.SSHInterface(addr, username, password, create_timeout=3, port=22)¶

An interface to an SSH Server.

class SSHResponse(exit_code: int, err_output: str, std_output: str)¶

err_output: str = None¶: The command stderr output

exit_code: int = None¶: The command exit code

property output¶: Combines both the err output and the std_output.

std_output: str = None¶: The command stdout output

classmethod as_plug(name, **kwargs_values)¶

Create a bound plug that will retrieve values from conf or the passed values here.

Take SSHInterface for example.

from spintop_openhtf.plugs import from_conf
from spintop_openhtf.plugs.ssh import SSHInterface

MySSHInterface = SSHInterface.as_plug(
    'MySSHInterface', # The name of this plug as it will appear in logs
    addr=from_conf( # from_conf will retrieve the conf value named like this.
        'my_ssh_addr',
        description="The addr of my device."
    ),
    username='x', # Always the same
    password='y'  # Always the same
)

close()¶: Abstract method: Close resources related to this interface.

close_log()¶: Close all logger handlers.

execute_command(command: str, timeout: float = 60, stdin: List = [], get_pty: bool = False, assertexitcode: Union[List[int], int, None] = 0)¶

Send a command and wait for it to execute.

Parameters

command – The command to send. End of lines are automatically managed. For example execute_command(‘ls’) will executed the ls command.
timeout – The timeout in second to wait for the command to finish executing.
stdin – A list of inputs to send into stdin after the command is started. Each entry in this list will be separated with an r’n’ character.
get_pty – Usually required when providing stdin.
assertexitcode – Unless this is None, defines one or a list of exit codes that are expected. After the command is executed, an SSHError will be raised if the exit code is not as expected.

Raises

SSHTimeoutError – Raised when timeout is reached.
SSHError – Raised when the exit code of the command is not in assertexitcode and assertexitcode is not None.

log_to_filename(filename, **kwargs)¶: Create and add to logger a FileHandler that logs to filename.

log_to_stream(stream=None, **kwargs)¶: Create and add to logger a StreamHandler that streams to stream

open(_client=None)¶: Abstract method: Open resources related to this interface.

tearDown()¶: Tear down the plug instance. This is part of the OpenHTF Plug contract

Telnet Interface¶

class spintop_openhtf.plugs.telnet.TelnetInterface(addr, port=4343)¶

classmethod as_plug(name, **kwargs_values)¶

Create a bound plug that will retrieve values from conf or the passed values here.

Take SSHInterface for example.

from spintop_openhtf.plugs import from_conf
from spintop_openhtf.plugs.ssh import SSHInterface

MySSHInterface = SSHInterface.as_plug(
    'MySSHInterface', # The name of this plug as it will appear in logs
    addr=from_conf( # from_conf will retrieve the conf value named like this.
        'my_ssh_addr',
        description="The addr of my device."
    ),
    username='x', # Always the same
    password='y'  # Always the same
)

close()¶: Abstract method: Close resources related to this interface.

close_log()¶: Close all logger handlers.

execute_command(command: str, expected_output: str = '', until_timeout: float = 5)¶

Send a command and wait for it to execute.

Parameters

command (str) – The command to send. End of lines are automatically managed. For example execute_command(‘ls’)
executed the ls command. (will) –
expected_output (str, optional) – If not none, the command executor will read until a given string, expected_output, is encountered or until until_timeout (s) have passed.
until_timeout (float, optional) – The timeout in seconds to wait for the command to finish reading output. Defaults to 5.

Returns

output response from the Telnet client

Return type

output (str)

Raises

TelnetError – Raised when reading the output goes wrong.

log_to_filename(filename, **kwargs)¶: Create and add to logger a FileHandler that logs to filename.

log_to_stream(stream=None, **kwargs)¶: Create and add to logger a StreamHandler that streams to stream

open(_client=None)¶: Abstract method: Open resources related to this interface.

tearDown()¶: Tear down the plug instance. This is part of the OpenHTF Plug contract