Reference¶

Test Plan¶

class spintop_openhtf.TestPlan(name: str = 'testplan', store_result: bool = True)¶

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.

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.

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

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.

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.

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_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.

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(*args, **kwargs)¶: 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.

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-11085885-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= False

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 = None¶: The command stderr output

exit_code = None¶: The command exit code

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

std_output = 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[T] = [], 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