API Documentation

Test Cases

Generic Test

module: test_case
synopsis: Basic extensible implementation of a TestCase.

Note

TODO later on will inherit from a metaclass to get the id parameters

class pykiso.test_coordinator.test_case.BasicTest(test_suite_id, test_case_id, aux_list, setup_timeout, run_timeout, teardown_timeout, test_ids, args, kwargs)[source]

Base for test-cases.

Initialize generic test-case.

Parameters

test_suite_id (int) – test suite identification number
test_case_id (int) – test case identification number
aux_list (Optional[List[AuxiliaryInterface]]) – list of used auxiliaries
setup_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during setup execution
run_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during test_run execution
teardown_timeout (Optional[int]) – the maximum time (in seconds) used to wait for a report during teardown execution
test_ids (Optional[dict]) – jama references to get the coverage eg: {“Component1”: [“Req1”, “Req2”], “Component2”: [“Req3”]}

cleanup_and_skip(aux, info_to_print)[source]

Cleanup auxiliary and log reasons.

Parameters

aux (AuxiliaryInterface) – corresponding auxiliary to abort
info_to_print (str) – A message you want to print while cleaning up the test

Return type

None

msg_handler: alias of pykiso.test_coordinator.test_message_handler.TestCaseMsgHandler

setUp()[source]

Hook method for constructing the test fixture.

Return type: None

classmethod setUpClass()[source]

A class method called before tests in an individual class are run.

This implementation is only mandatory to enable logging in junit report. The logging configuration has to be call inside test runner run, otherwise stdout is never caught.

Return type: None

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

Return type: None

test_run()[source]

Hook method from unittest in order to execute test case.

Return type: None

pykiso.test_coordinator.test_case.define_test_parameters(suite_id=0, case_id=0, aux_list=None, setup_timeout=None, run_timeout=None, teardown_timeout=None, test_ids=None)[source]: Decorator to fill out test parameters of the BasicTest automatically.

Connectors

Interface Definition for Connectors, CChannels and Flasher

module: connector
synopsis: Interface for a channel

class pykiso.connector.CChannel(**kwargs)[source]

Abstract class for coordination channel.

constructor

abstract _cc_close()[source]: Close the channel.

abstract _cc_open()[source]: Open the channel.

abstract _cc_receive(timeout, raw=False)[source]

How to receive something from the channel.

Parameters

timeout (float) – Time to wait in second for a message to be received
raw (bool) – send raw message without further work (default: False)

Return type

Union[Message, bytes, str]

Returns

message.Message() - If one received / None - If not

abstract _cc_send(msg, raw=False)[source]

Sends the message on the channel.

Parameters

msg (Union[Message, bytes, str]) – Message to send out
raw (bool) – send raw message without further work (default: False)

Return type

None

cc_receive(timeout=0.1, raw=False)[source]

Read a thread-safe message on the channel and send an acknowledgement.

Parameters

timeout (float) – time in second to wait for reading a message
raw (bool) – should the message be returned raw or should it be interpreted as a pykiso.Message?

Returns

Message if successful, None else

Raises

ConnectionRefusedError – when lock acquire failed

cc_send(msg, raw=False, **kwargs)[source]

Send a thread-safe message on the channel and wait for an acknowledgement.

Parameters: msg (Union[Message, bytes, str]) – message to send
Raises: ConnectionRefusedError – when lock acquire failed

close()[source]

Close a thread-safe channel.

Return type: None

open()[source]

Open a thread-safe channel.

Raises: ConnectionRefusedError – when lock acquire failed
Return type: None

class pykiso.connector.Connector(name=None)[source]

Abstract interface for all connectors to inherit from.

Defines hooks for opening and closing the connector and also defines a contextmanager interface.

Constructor.

Parameters: name (Optional[str]) – alias for the connector, used for repr and logging.

abstract close()[source]: Close the connector, freeing resources.

abstract open()[source]: Initialise the Connector.

class pykiso.connector.Flasher(binary=None, **kwargs)[source]

Interface for devices that can flash firmware on our targets.

Constructor.

Parameters

binary (Union[str, Path, None]) – binary firmware file

Raises

ValueError – if binary doesn’t exist or is not a file
TypeError – if given binary is None

abstract flash()[source]: Flash firmware on the target.

Auxiliary Interface

Auxiliary Interface Definition

module: auxiliary
synopsis: Implementation of basic auxiliary functionality and definition of abstract methods for override

class pykiso.auxiliary.AuxiliaryInterface(name=None, is_proxy_capable=False, is_pausable=False, activate_log=None)[source]

Defines the Interface of Auxiliaries.

Auxiliaries get configured by the Test Coordinator, get instantiated by the TestCases and in turn use Connectors.

Auxiliary thread initialization.

Parameters

name (Optional[str]) – alias of the auxiliary instance
is_proxy_capable (bool) – notify if the current auxiliary could be (or not) associated to a proxy-auxiliary.
is_pausable (bool) – notify if the current auxiliary could be (or not) paused
activate_log (Optional[List[str]]) – loggers to deactivate

abort_command(blocking=True, timeout_in_s=25)[source]

Force test to abort.

Parameters

blocking (bool) – If you want the command request to be blocking or not
timeout_in_s (float) – Number of time (in s) you want to wait for an answer

Return type

bool

Returns

True - Abort was a success / False - if not

create_instance()[source]

Create an auxiliary instance and ensure the communication to it.

Return type: bool
Returns: message.Message() - Contain received message

delete_instance()[source]

Delete an auxiliary instance and its communication to it.

Return type: bool
Returns: message.Message() - Contain received message

static initialize_loggers(loggers)[source]

Deactivate all external loggers except the specified ones.

Parameters: loggers (Optional[List[str]]) – list of logger names to keep activated

lock_it(timeout_in_s)[source]

Lock to ensure exclusivity.

Parameters: timeout_in_s (integer) – How many second you want to wait for the lock
Return type: bool
Returns: True - Lock done / False - Lock failed

resume()[source]

Resume current auxiliary’s run.

Return type: None

run()[source]

Run function of the auxiliary thread.

Return type: None

run_command(cmd_message, cmd_data=None, blocking=True, timeout_in_s=0)[source]

Send a test request.

Parameters

cmd_message (Union[Message, bytes, str]) – command request to the auxiliary
cmd_data (Optional[Any]) – data you would like to populate the command with
blocking (bool) – If you want the command request to be blocking or not
timeout_in_s (int) – Number of time (in s) you want to wait for an answer

Return type

bool

Returns

True - Successfully sent / False - Failed by sending / None

stop()[source]

Force the thread to stop itself.

Return type: None

suspend()[source]

Supend current auxiliary’s run.

Return type: None

unlock_it()[source]

Unlock exclusivity

Return type: None

wait_and_get_report(blocking=False, timeout_in_s=0)[source]

Wait for the report of the previous sent test request.

Parameters

blocking (bool) – True: wait for timeout to expire, False: return immediately
timeout_in_s (int) – if blocking, wait the defined time in seconds

Return type

Union[Message, bytes, str]

Returns

a message.Message() - Message received / None - nothing received

Message Protocol

pykiso Control Message Protocol

module: message
synopsis: Message that will be send though the different agents

class pykiso.message.Message(msg_type=0, sub_type=0, error_code=0, test_suite=0, test_case=0, tlv_dict=None)[source]

A message who fit testApp protocol.

The created message is a tlv style message with the following format: TYPE: msg_type | message_token | sub_type | errorCode |

Create a generic message.

Parameters

msg_type (MessageType) – Message type
sub_type (Message<MessageType>Type) – Message sub-type
error_code (integer) – Error value
test_section (integer) – Section value
test_suite (integer) – Suite value
test_case (integer) – Test value
tlv_dict (dict) – Dictionary containing tlvs elements in the form {‘type’:’value’, …}

check_if_ack_message_is_matching(ack_message)[source]

Check if the ack message was for this sent message.

Parameters: ack_message (Message) – received acknowledge message
Return type: bool
Returns: True if message type and token are valid otherwise False

generate_ack_message(ack_type)[source]

Generate acknowledgement to send out.

Parameters: ack_type (int) – ack or nack
Return type: Optional[Message]
Returns: filled acknowledge message otherwise None

classmethod get_crc(serialized_msg, crc_byte_size=2)[source]

Get the CRC checksum for a bytes message.

Parameters

serialized_msg (bytes) – message used for the crc calculation
crc_byte_size (int) – number of bytes dedicated for the crc

Return type

int

Returns

CRC checksum

get_message_sub_type()[source]

Return actual message subtype.

Return type: int

get_message_tlv_dict()[source]

Return actual message type/length/value dictionary.

Return type: dict

get_message_token()[source]

Return actual message token.

Return type: int

get_message_type()[source]

Return actual message type.

Return type: Union[int, MessageType]

classmethod parse_packet(raw_packet)[source]

Factory function to create a Message object from raw data.

Parameters: raw_packet (bytes) – array of a received message
Return type: Message
Returns: itself

serialize()[source]

Serialize message into raw packet.

Format: | msg_type (1b) | msg_token (1b) | sub_type (1b) | error_code (1b) |: test_section (1b) | test_suite (1b) | test_case (1b) | payload_length (1b) |

tlv_type (1b) | tlv_size (1b) | … | crc_checksum (2b)

Return type: bytes
Returns: bytes representing the Message object

class pykiso.message.MessageAckType(value)[source]: List of possible received messages.

class pykiso.message.MessageCommandType(value)[source]: List of commands allowed.

class pykiso.message.MessageLogType(value)[source]: List of possible received log messages.

class pykiso.message.MessageReportType(value)[source]: List of possible received messages.

class pykiso.message.MessageType(value)[source]: List of messages allowed.

class pykiso.message.TlvKnownTags(value)[source]: List of known / supported tags.

Import Magic

Auxiliary Interface Definition

module: dynamic_loader
synopsis: Import magic that enables aliased auxiliary loading in TestCases

class pykiso.test_setup.dynamic_loader.DynamicImportLinker[source]

Public Interface of Import Magic.

initialises the Loaders, Finders and Caches, implements interfaces to install the magic and register the auxiliaries and connectors.

Initialize attributes.

install()[source]: Install the import hooks with the system.

provide_auxiliary(name, module, aux_cons=None, **config_params)[source]

Provide a auxiliary.

Parameters

name (str) – the auxiliary alias
module (str) – either ‘python-file-path:Class’ or ‘module:Class’ of the class we want to provide
aux_cons – list of connectors this auxiliary has

provide_connector(name, module, **config_params)[source]

Provide a connector.

Parameters

name (str) – the connector alias
module (str) – either ‘python-file-path:Class’ or ‘module:Class’ of the class we want to provide

uninstall()[source]: Deregister the import hooks, close all running threads, delete all instances.

Config Registry

module: config_registry
synopsis: register auxiliaries and connectors to provide them for import.

class pykiso.test_setup.config_registry.ConfigRegistry[source]

Register auxiliaries with connectors to provide systemwide import statements.

classmethod delete_aux_con()[source]

deregister the import hooks, close all running threads, delete all instances.

Return type: None

classmethod get_all_auxes()[source]

Return all auxiliaires instances and alias

Return type: dict
Returns: dictionary with alias as keys and instances as values

classmethod get_auxes_alias()[source]

return all created auxiliaries alias.

Return type: list
Returns: list containing all auxiliaries alias

classmethod get_auxes_by_type(aux_type)[source]

Return all auxiliaries who match a specific type.

Parameters: aux_type (Any) – auxiliary class type (DUTAuxiliary, CommunicationAuxiliary…)
Return type: dict
Returns: dictionary with alias as keys and instances as values

classmethod register_aux_con(config)[source]

Create import hooks. Register auxiliaries and connectors.

Parameters: config (dict) – dictionary containing yaml configuration content
Return type: None

Test Suites

Test Suite

module: test_suite
synopsis: Create a generic test-suite based on the connected modules

class pykiso.test_coordinator.test_suite.BaseTestSuite(test_suite_id, test_case_id, aux_list, setup_timeout, run_timeout, teardown_timeout, test_ids, args, kwargs)[source]

Initialize generic test-case.

Parameters

test_suite_id (int) – test suite identification number
test_case_id (int) – test case identification number
aux_list (Optional[List[AuxiliaryInterface]]) – list of used auxiliaries
setup_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during setup execution
run_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during test_run execution
teardown_timeout (Optional[int]) – the maximum time (in seconds) used to wait for a report during teardown execution
test_ids (Optional[dict]) – jama references to get the coverage eg: {“Component1”: [“Req1”, “Req2”], “Component2 [“Req3”]}

base_function(current_fixture, step_name, test_command, timeout_resp)[source]

Base function used for test suite setup and teardown.

Parameters

current_fixture (Callable) – fixture instance teardown or setup
step_name (str) – name of the current step
test_command (Message) – A message you want to print while cleaning up the test
timeout_resp (int) – maximum amount of time in seconds to wait for a response

cleanup_and_skip(aux, info_to_print)[source]

Cleanup auxiliary and log reasons.

Parameters

aux (AuxiliaryInterface) – corresponding auxiliary to abort
info_to_print (str) – A message you want to print while cleaning up the test

class pykiso.test_coordinator.test_suite.BasicTestSuite(modules_to_add_dir, test_filter_pattern, test_suite_id, args, kwargs)[source]

Inherit from the unittest framework test-suite but build it for our integration tests.

Initialize our custom unittest-test-suite.

class pykiso.test_coordinator.test_suite.BasicTestSuiteSetup(test_suite_id, test_case_id, aux_list, setup_timeout, run_timeout, teardown_timeout, test_ids, args, kwargs)[source]

Inherit from unittest testCase and represent setup fixture.

Initialize generic test-case.

Parameters

test_suite_id (int) – test suite identification number
test_case_id (int) – test case identification number
aux_list (Optional[List[AuxiliaryInterface]]) – list of used auxiliaries
setup_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during setup execution
run_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during test_run execution
teardown_timeout (Optional[int]) – the maximum time (in seconds) used to wait for a report during teardown execution
test_ids (Optional[dict]) – jama references to get the coverage eg: {“Component1”: [“Req1”, “Req2”], “Component2”: [“Req3”]}

test_suite_setUp()[source]: Test method for constructing the actual test suite.

class pykiso.test_coordinator.test_suite.BasicTestSuiteTeardown(test_suite_id, test_case_id, aux_list, setup_timeout, run_timeout, teardown_timeout, test_ids, args, kwargs)[source]

Inherit from unittest testCase and represent teardown fixture.

Initialize generic test-case.

Parameters

test_suite_id (int) – test suite identification number
test_case_id (int) – test case identification number
aux_list (Optional[List[AuxiliaryInterface]]) – list of used auxiliaries
setup_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during setup execution
run_timeout (Optional[int]) – maximum time (in seconds) used to wait for a report during test_run execution
teardown_timeout (Optional[int]) – the maximum time (in seconds) used to wait for a report during teardown execution
test_ids (Optional[dict]) – jama references to get the coverage eg: {“Component1”: [“Req1”, “Req2”], “Component2”: [“Req3”]}

test_suite_tearDown()[source]: Test method for deconstructing the actual test suite after testing it.

class pykiso.test_coordinator.test_suite.TestSuiteMsgHandler[source]

Encapsulate all test suite communication mechanisms.

Warning

This class is speculative code and not currently used. it should eventually replace the handle_basic_interaction call in the BaseTestSuite

classmethod setup(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test suite setup. Based on actual implementation, this context manager calls handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

classmethod teardown(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test suite teardown. Based on actual implementation, this context manager call handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

Test Execution

module: test_execution
synopsis: Execute a test environment based on the supplied configuration.

Note

Glob a list of test-suite folders
Generate a list of test-suites with a list of test-cases
Loop per suite
Gather result

class pykiso.test_coordinator.test_execution.ExitCode(value)[source]: List of possible exit codes

pykiso.test_coordinator.test_execution.create_test_suite(test_suite_dict)[source]

create a test suite based on the config dict

Parameters: test_suite_dict (Dict) – dict created from config with keys ‘suite_dir’, ‘test_filter_pattern’, ‘test_suite_id’
Return type: BasicTestSuite

pykiso.test_coordinator.test_execution.execute(config, report_type='text')[source]

create test environment base on config

Parameters

config (Dict) – dict from converted YAML config file
report_type (str) – str to set the type of report wanted, i.e. test or junit

pykiso.test_coordinator.test_execution.failure_and_error_handling(result)[source]

provide necessary information to Jenkins if an error occur during tests execution

Parameters: result – a unittest.TestResult object
Returns: an ExitCode object

Test-Message Handling

Handle common communication with device under test

By default, the integration test framework handles internal messaging and control flow using a message format defined in pykiso.Message. pykiso.test_message_handler defines the default messaging protocol from a behavioral point of view.

The general procedure is described in handle_basic_interaction context manager, but specific _MsgHandler_ classes are provided with TestCaseMsgHandler and TestSuiteMsgHandler to provide shorthands for the specialised communication from pykiso.test_case.BasicTest and pykiso.test_suite.BasicTestSuite.

module: test_message_handler
synopsis: default communication between TestManagement and DUT.

class pykiso.test_coordinator.test_message_handler.TestCaseMsgHandler[source]

Encapsulate all test case communication mechanism.

classmethod run(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test case run. Based on actual implementation, this context manager call handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

classmethod setup(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test case setup. Based on actual implementation, this context manager call handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

classmethod teardown(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test case teardown. Based on actual implementation, this context manager call handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

class pykiso.test_coordinator.test_message_handler.TestSuiteMsgHandler[source]

Encapsulate all test suite communication mechanisms.

Warning

This class is speculative code and not currently used. it should eventually replace the handle_basic_interaction call in the BaseTestSuite

classmethod setup(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test suite setup. Based on actual implementation, this context manager calls handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

classmethod teardown(cls, test_entity, timeout_cmd, timeout_resp)[source]

Handle communication mechanism during test suite teardown. Based on actual implementation, this context manager call handle_basic_interaction.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
timeout_cmd (int) – timeout in second apply on auxiliary run_command
timeout_resp (int) – timeout in second apply on auxiliary wait_and_get_report

Returns

namedtuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

pykiso.test_coordinator.test_message_handler.handle_basic_interaction(test_entity, cmd_sub_type, timeout_cmd, timeout_resp)[source]

Handle default communication mechanism between test manager and device under test as follow:

TM |   COMMAND ---------->        | DUT
TM |           <---------- ACK    | DUT
TM |           <---------- LOG    | DUT
TM |   ACK     ---------->        | DUT
...
TM |           <---------- LOG    | DUT
TM |   ACK     ---------->        | DUT
TM |           <---------- REPORT | DUT
TM |   ACK     ---------->        | DUT

This behaviour is implemented here.

Logs can be sent to TM while waiting for report.

Parameters

test_entity (Callable) – test instance in use (BaseTestSuite, BasicTest,…)
cmd_sub_type (MessageCommandType) – message command sub-type (Test case run, setup,….)
timeout_cmd (int) – timeout in seconds for auxiliary run_command
timeout_resp (int) – timeout in seconds for auxiliary wait_and_get_report

Return type

List[report_analysis]

Returns

tuple containing current auxiliary, reported message, logging method to use, and pre-defined log message.

class pykiso.test_coordinator.test_message_handler.report_analysis(current_auxiliary, report_message, logging_method, log_message)

Create new instance of report_analysis(current_auxiliary, report_message, logging_method, log_message)

property current_auxiliary: Alias for field number 0

property log_message: Alias for field number 3

property logging_method: Alias for field number 2

property report_message: Alias for field number 1

test xml result

test_xml_result

module: test_xml_result
synopsis: overwrite xmlrunner.result to be able to add additional data into the xml report.

class pykiso.test_coordinator.test_xml_result.TestInfo(test_result, test_method, outcome=0, err=None, subTest=None, filename=None, lineno=None, doc=None)[source]

This class keeps useful information about the execution of a test method. Used by XmlTestResult

Initialize the TestInfo class and append additional tag: that have to be stored for each test

Parameters

test_result (_XMLTestResult) – test result class
test_method – test method (dynamically created eg: test_case.MyTest2-1-2)
outcome (int) – result of the test (SUCCESS, FAILURE, ERROR, SKIP)
err – error cached during test
subTest – optional, refer the test id and the test description
filename (Optional[str]) – name of the file
lineno (Optional[bool]) – store the test line number
doc (Optional[str]) – additional documentation to store

class pykiso.test_coordinator.test_xml_result.XmlTestResult(stream=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, descriptions=True, verbosity=1, elapsed_times=True, properties=None, infoclass=<class 'pykiso.test_coordinator.test_xml_result.TestInfo'>)[source]

Test result class that can express test results in a XML report. Used by XMLTestRunner

Initialize the _XMLTestResult class.

Parameters

stream (TextIOWrapper) – buffered text interface to a buffered raw stream
descriptions (bool) – include description of the test
verbosity (int) – print output into the console
elapsed_times (bool) – include the time spend on the test
properties – junit testsuite properties
infoclass (_TestInfo) – class containing the test information

report_testcase(xml_testsuite, xml_document): Appends a testcase section to the XML document.