Reference

Config

class testsystem.config.Config

Configuration class for the RTOS Test System. Members can be overwritten via environment variables or a JSON config file. Details on how configuration works can be found in section Configuration. To set a list parameter with an environment variable, concatenate the values with ;.

The following labels describe how a property can be set or modified:

env: Configuration via environment variable is possible.
file: Configuration via JSON config file is possible. This usually also implies env.
dyn: Dynamically changing the configuration will affect the running system.

conf_enable_env: bool = True: Flag to enable configuration through environment variables.

conf_enable_file: bool = True: env dyn

Flag to enable configuration through the config file.

conf_file: str = '/host/config.json': env dyn

Path to the JSON config file.

db_database: str = 'testsystem': env dyn

Database name.

db_file: str = '/host/testsystem.db': env dyn

Path to the Database file.

db_password: str = 'Password1': env dyn

Database password.

db_server: str = '127.0.0.1': env dyn

Database server/host.

db_type: str = 'sqlite': env file

Set the database type used for persisting data. Available options are: sqlite, mysql .

db_user: str = 'testsystem': env dyn

Database user name.

enable_timing_tests: bool = True: env file dyn

Enables timing test cases. Timing tests won’t be included in test runs if this is set to False.

exercise_nr: int = 0: env file dyn

The current exercise number. Test runs will also include all test cases from previous exercises.

force_test_tags: list[str] = ['ex1', 'ex2', 'ex3', 'ex4', 'ex5', 'ex6']: env file dyn

This parameter provides a list of tags that must be tested anyways. This guarantees a test report for special commits that might be skipped by regular scheduling. Tag values are case-insensitive.

git_primary_branch_name: str = 'main': env file dyn

Name for primary git branch. Usually this is master or main.

git_public_path: str = 'eas/teaching/RTOS_SS23': env file dyn

Relative path on git server root where to find public repos.

git_server: str = 'ssh://git@iti-gitlab.tugraz.at': env file dyn

Git server with all repos. You can use absolute paths to use the filesystem as remote.

git_student_path: str = 'eas/teaching/RTOS_SS23': env file dyn

Relative path on git server root where to find student repos.

git_system_path: str = 'eas/teaching/RTOS_SS23/Testsystem_Reports_SS23': env file dyn

Relative path to the system repository. This repository is used to push test reports, to configure the testsystem and for monitoring.

group_ids: list[int] = [1]: env file dyn

A list of group ids. This is used to determine which groups actively participate and should get tested. You can remove or add existing groups during runtime.

log_file: str = '/host/testsystem.log': env file

Option to define where to wirte testsystem logs.

log_level: str = 'DEBUG': env file

Specify the log level to use for testsystem logs. Available options are:

DEBUG, INFO, WARN, ERROR, CRITICAL.

prio_reset_h = 24: env file dyn

Parameter to specify when to reset group priorities. The value is given in hours after which the priorities are reset to default.

run_hello_testsystem: bool = False: env file

If set to True, the test system runs its hello world equivalent, displaying some general information.

start_delay: int = 0: env file

System start delay in seconds. This might be used to wait for the database container.

stop: bool = False: env file dyn

This flag is used to gracefully shut down the testsystem.

tc_root_path = '/testcases': env file

Path to the test case directory.

term: str = 'SS23': env file

Term name.

tu_connections: list[str] = ['P6.0-D7']: env file

Defines which connections must be available between an MSP430 and a PicoScope to form a TestUnit. The syntax for the MSP430 pin is P<Port>.<Pin>, and for the PicoScope channel, this would be D<Channel>, where D stands for digital channels. The pin and channel strings are combined with a -.

P1.2-D3 would be a valid configuration for a required connection between port 1.2 and digital channel 3.

testsystem.config.get_config() → Config: Function to request current configuration.

Device Discovery

testsystem.device_discovery.discover_msp_boards() → list[MSP430]

Discover MSP boards connected to serial ports.

Returns:: Returns a list of available MSP430 boards.

testsystem.device_discovery.discover_pico_scopes() → list[PicoScope]

Discover PicoScopes connected via USB.

Returns:: Returns a list of available PicoScopes.

testsystem.device_discovery.discover_pico_scopes_by_sn() → list[str]

Discover PicoScopes connected via USB.

Returns:: Returns a list of serial numbers for the available PicoScopes.

Exceptions

exception testsystem.exceptions.BuildError(msg: str, output: str | None = None, error: str | None = None, *args: object)

exception testsystem.exceptions.ConfigError(parameter, value, msg: str | None = None)

exception testsystem.exceptions.DecodeError(msg: str, *args: object)

exception testsystem.exceptions.DetectError(*args: object)

exception testsystem.exceptions.FirmewareError(msg: str, *args: object)

exception testsystem.exceptions.FlashError(msg: str, output: str | None = None, error: str | None = None, *args: object)

exception testsystem.exceptions.GitError(msg: str, *args: object)

exception testsystem.exceptions.IdentificationError(msg)

exception testsystem.exceptions.InvalidSignalError(*args: object)

exception testsystem.exceptions.InvalidTXPauseError(*args: object)

exception testsystem.exceptions.MSPConnectionError(msg: str, *args: object)

exception testsystem.exceptions.MSPError(msg: str, *args: object)

exception testsystem.exceptions.NoSignalError(*args: object)

exception testsystem.exceptions.NotConnectedError(*args: object)

exception testsystem.exceptions.ParsingError(msg)

exception testsystem.exceptions.PicoError(code)

exception testsystem.exceptions.ProcessError(msg: str, *args: object)

exception testsystem.exceptions.StartBitError(*args: object)

exception testsystem.exceptions.StopBitError(*args: object)

exception testsystem.exceptions.TestCaseError(msg: str, *args: object)

Filesystem

class testsystem.filesystem.TestEnv(group_name: str, commit: str)

Secure test environment for a group on a specific commit.

Parameters:

directory – The directory of the git repository.
state – The state in whicht to checkout the repository. This can be a tag or commit hash.

cleanup(): Cleanup the test environment and remove directories used for testing.

property commit_hash: str: Returns the hash of the tested commit.

property commit_hash_short: str: Returns the short hash (first 8 chars) of the tested commit.

property commit_msg: str: Returns the commit message of the tested commit.

property commit_time: int: Returns the commit timestamp as unix millis of the tested commit.

export(dest: str)

Copies the test environment to a given destination.

Parameters:: dest – Path to destination.

property group_name: str: Returns the group name that is loaded in this environment.

property group_path: str: Returns the path to the group directory of this environment.

property path: str: Returns the path to this environment directory.

property public_path: str: Returns the path to the public directory of this environment.

testsystem.filesystem.create_msp430_identifier_program(device_id: int) → str

Create a new MSP430 identification program.

Parameters:: device_id – The device id to use for the new program.
Returns:: Path to the program file.

testsystem.filesystem.get_commit_message(group_name: str, commit: str) → str

Get the message from a specific commit.

Parameters:

group_name – Group name for repository.
commit – The commit hash from where to get the message.

Returns:

Returns the commit message.

testsystem.filesystem.get_commit_timestamp(group_name: str, commit: str) → int

Get the timestamp from a specific commit.

Parameters:

group_name – Group name for repository.
commit – The commit hash from where to get the timestamp.

Returns:

Returns the commit timestamp as unix millis.

testsystem.filesystem.get_latest_commit(group_name: str) → str

Get the latest commit on the primary branch.

Parameters:: group_name – Group name for repository.
Returns:: Returns the hash of the commit.

testsystem.filesystem.get_next_commit(group_name: str, commit: str | None = None) → str

Get the next child commit on the primary branch after the provided commit. If no commit is provided or the commit is not found in the primary branch, the first one will be used.

Parameters:

group_name – Group name for repository.
state – The commit hash of the parent.

Returns:

Returns the hash of the child commit on the primary branch.

testsystem.filesystem.get_tagged_group_commit(group_name: str, tags: list[str]) → list[str]

Get all commits tagged with one of the provided tags.

Parameters:

group_name – The name of the group where to look for tagged commits.
tags – A list of tags.

Returns:

Returns a list of commit hashes.

testsystem.filesystem.init_fs(): Cleans and initializes local directories.

testsystem.filesystem.load_group(group_name: str) → str | None

Initializes and updates a local clone for the group.

Raises:: GitError – If cloning or pulling the group repository failed.
Parameters:: group_name – Repo name.
Returns:: Hash of latest commit on primary branch.

testsystem.filesystem.load_public() → str

Initializes and updates the local RTOS repository.

Raises:: GitError – If cloning or pulling the public repository failed.
Returns:: Hash of lates commit on primary branch.

testsystem.filesystem.load_repo(local_path: str, rel_remote_path: str) → str

Initializes and updates a local clone.

Parameters:

local_path – Local git repository path.
rel_remote_path – Relative path on git server.

Returns:

Hash of latest commit on primary branch.

testsystem.filesystem.load_test_case(test_env: TestEnv, test_case_name: str) → str

Load a given test case into the test environment.

Raises:

TestCaseError – If loading the test case failed.

Parameters:

env_id – A test environment.
test_case_name – Name of test case that should be loaded.

Returns:

Path to the loaded test case.

testsystem.filesystem.publish_group_report(group_name: str, commit: str, content: str, file_type: str = 'md')

Publish test results for a specific group.

Parameters:

group_name – Group for which to publish the report.
content – The content of the report.
file_type – The file type of the content.

testsystem.filesystem.publish_system_status_report(sys_report: str)

Publish the global system status report.

Parameters:: sys_report – The content of the report.

testsystem.filesystem.publish_test_run_report(group_name: str, commit: str, content: str, file_type: str = 'md')

Publish a test run report for a specific group.

Parameters:

group_name – Group name for repository.
content – The content of the report.
file_type – The file type of the content.

testsystem.filesystem.setup_test_env(group_name: str, commit: str) → TestEnv: Creates a save environment for the test run. This is outside of the local git repositories.

Models

class testsystem.models.ChannelReader(pin: int, smpls_per_bit: int)

Class to capture long UART signals on a PicoScope channel.

get_data() → list[int]

Get a list of bytes that were decoded from the recorded UART signal.

Returns:: List of bytes.

record(data)

Record signal data.

Parameters:: data – New data to add.

class testsystem.models.ConnectionDetector(msps: list[MSP430], picoscopes: list[PicoScope])

This class contains logic to identify connections between MSPs and PicoScopes and build test units out of them.

Parameters:

msps – A list of available MSPs. These will identify themselves to PicoScopes through a unique id.
picoscopes – A list of available PicoScopes. These will listen for MSP ids on their channels.

get_test_units() → list[TestUnit]

Start the identification process and get valid combinations of MSPs and PicoScopes as test units.

Returns:: Returns a list of all identified test units.

class testsystem.models.ConnectionInfo(port: int, pin: int, channel: str, device_id: int | None = None, msp: MSP430 | None = None, pico: PicoScope | None = None)

This class stores information about a connection between an MSP430 board an a PicoScope.

Parameters:

port – The port on the MSP430. Port and pin are required to identify the connection on the MSP side uniquely.
pin – The pin on the MSP430. Port and pin are required to identify the connection on the MSP side uniquely.
channdel – The channel on the PicoScope. The channel string starts with ‘A’ for analog or ‘D’ for a digital channel, followed by its number.
device_id – The id which the PicoScope decoded from the signal it received.
msp – The msp object on one side of the connection.
pico – The PicoScope object on one side of the connection.

class testsystem.models.Group(**kwargs)

The group class. This class is also a database object.

add_queue_time(waiting_time: float)

Add time to this groups queue time.

Parameters:: waiting_time – The time to add in seconds.

classmethod get(group_nr: int, term: str) → Group | None

Get a specific group by its number and the term.

Parameters:

group_nr – The group number.
term – The term. E.g: SS22

Returns:

The group if it exists, None otherwise.

classmethod get_by_id(id: int) → Group | None

Get a group by its id.

Parameters:: id – The group id.
Returns:: The group if it exists, None otherwise.

classmethod get_by_term(term: str, active=True) → list[Group]

Get all groups from a specific term.

Parameters:: term – The term. E.g: SS22
Returns:: Returns a list of groups.

get_latest_finished_test_set() → TestSet | None

Get the latest test set which is already finished for this group.

Returns:: The latest finished test set if it exists, None otherwise.

get_latest_test_set() → TestSet | None

Get the latest test set for this group.

Returns:: The latest test set if it exists, None otherwise.

static get_max_queue_time() → float

Get the max queue time of all groups.

Returns:: The max queue time in seconds.

classmethod get_name(group_nr, term) → str

Returns the group name based on the group number and term. Use this method every time you need to construct the group name from the group number.

Parameters:

group_nr – The group number.
term – The term. E.g: SS22

Returns:

The group name. E.g: RTOS_SS22_Group01

get_priority() → float

Get the priority of this group. Change this method if you want to modify the group priority calculation. The priority is between 10 and 20 to leave some levels for higher priority tasks that may come in the future.

Returns:: The priority of this group.

get_queue_time() → float

Get the queue time of the group.

Returns:: The queue time in seconds.

get_test_set(commit) → TestSet | None

Get a specific test set from this group.

Returns:: The test set if it exists, None otherwise.

reset_queue_time(): Reset this groups queue time to zero.

class testsystem.models.MSP430(**kwargs)

Class for MSP430 boards. This class is also a database object.

classmethod get_all() → list[MSP430]

Get all MSPs ever connected to the test system.

Returns:: A list of all MSPs.

classmethod get_by_sn(serial_number: str) → MSP430 | None

Get an MSP by its serial number.

Parameters:: serial_number – The serial number of the MSP.
Returns:: The MSP if it exists, None otherwise.

classmethod get_connected() → list[MSP430]

Get all connected MSPs.

Returns:: A list of all connected MSPs.

get_identifier() → str

Get a string representation of this MSP.

Returns:: Representative name.

increment_flash_counter(): Increment the flash counter of this MSP.

is_connected() → bool

Check if this MSP is currently connected.

Returns:: True if it is connected, False otherwise.

set_defective(): Mark this device as defective. Defective devices are checked if they can be recovered.

set_name(name: str)

Set the name of an MSP board. The name parameter is only for easier identification in the system reports and serves no functional purpose.

Parameters:: name – The name to set for this MSP.

class testsystem.models.PicoMeasure(picoscope: PicoScope)

Class used for timing tests to measure the timing signal on a PicoScope.

Parameters:: picoscope – PicoScope to be used for measuring.

measure() → float

Read and decode the captured timing signal.

Returns:: Returns the timing signal in microseconds.

start(): Start the measuring task.

class testsystem.models.PicoReader(picoscope: PicoScope)

This class is used as an interface to the PicoScopes during the discovery process. It handles the setup and configuration for the PicoScope, reads the streamed data from the scope, and decodes the UART signal into a list of bytes for identification.

Parameters:: picoscope – PicoScope instance to be used for capturing.

get_channel_data(channel: int) → list[int]

This method returns a list of bytes captured on a given channel. To get some results you must first call the read() method.

Parameters:: channel – The channel number from where to get the data.

read(baud_rate: int, smpls_per_bit: int, reading_time_ms: float)

This method first connects to the PicoScope and configures it for capturing. The call will fail if there is already an open connection to the scope. The reader will close the connection to the PicoScope once capturing, and decoding are finished. This call blocks until the described tasks are completed.

Exceptions PicoError:

Is raised when something went wrong when communicating with the PicoScope.

Parameters:

baud_rate – The baud rate used by the msp430-identifier program flashed to the MSP430 boards.
smpls_per_bit – The number of samples we want to capture for each bit. The sample, which is half-time through the bit, will be used for decoding.
reading_time_ms – The capturing time in milliseconds. This value should be high enough to include at least one complete id. A reasonable capture would contain 2 - 5 ids.

class testsystem.models.PicoScope(**kwargs)

Class for PicoScopes. This class is also a database object.

classmethod from_serial(serial_number: str) → PicoScope

Get or create a new PicoScope instance base on a serial number.

Parameters:: serial – The serial number of the PicoScope.
Returns:: A PicoScope instance.

classmethod get_all() → list[PicoScope]

Get all PicoScopes ever connected to the test system.

Returns:: A list of all PicoScopes.

classmethod get_by_sn(serial_number: str) → PicoScope | None

Get a PicoScope by its serial number.

Parameters:: serial_number – The serial number of the PicoScope.
Returns:: The PicoScope if it exists, None otherwise.

set_name(name: str)

Set the name of a PicoScope. The name parameter is only for easier identification in the system reports and serves no functional purpose.

Parameters:: name – The name to set for this PicoScope.

class testsystem.models.TaskWorker(test_unit: TestUnit, name: str)

Worker class for a test unit. This class executes task on a test unit.

Parameters:

test_unit – The test unit which should be used by this worker.
name – A name for this worker. This is only for easier log inspection and has no functional purpose.

start(): Start the task worker.

stop(): Stop the task worker. This call can block for a while because the test unit might need some time to stop its current task.

class testsystem.models.Task(priority: float = 100, test_unit: TestUnit | None = None, tag: str | None = None, callback=None, error_callback=None)

The base class for all tasks.

Parameters:

priority – The priority for the task. Lower value means higher priority.
test_unit – If specified, the task may only run on this test unit.
tag – If specified, the task my only run on a test unit with the specific tag.
callback – A function of type func(Task), which is called on successful completion of the task.
error_callback – A function of type func(Task, Exception), which is called when the task fails.

run(): Override this method to implement the tasks functionallity.

property runtime: float: Time between start and finish of the task.

property use_specific_test_unit: bool: Flag if this task must use a specific test unit.

property use_tagged_test_unit: bool: Flag if this task must use a test unit with a specific tag.

property wait_time: float: Time between scheduling and start in seconds.

class testsystem.models.TestCaseDef(id: int = 0, exercise_nr: int = 0, runtime: int = 0, description: str = '', timing: bool = False, panic: bool = False, size: bool = False, ranking: bool = False)

RTOS test case definition class. Instances of this class are derived from the testcases.txt file.

classmethod get() → list[TestCaseDef]

Load active test case definitions. This method accounts for config settings.

Parameters:: disable_cache – Flag to disable caching.
Returns:: A list of test cases that should be used for test runs.

classmethod get_all() → list[TestCaseDef]

Load all parsable test case definitions. This does not account for configuration (exersice, timing, size, …).

Parameters:: disable_cache – Flag to disable caching.
Returns:: A list of test case definitions.

classmethod get_by_id(id: int) → TestCaseDef | None

Get a test case by its test case number.

Parameters:: id – Test case id.
Returns:: Returns the test case definition or None if it does not exist.

classmethod get_max_exercise_nr() → int

Get the highest exercise number found in test case definitions.

Returns:: The highest exercise number.

classmethod get_ranking_tests() → list[TestCaseDef]

Get a list of active ranking test cases.

Returns:: A list of all ranking tests.

classmethod parse(line: str) → TestCaseDef

Parses a line from the test case definitions into a testsystem.models.TestCaseDef object.

Exceptions ParsingError:: Is raised if the config could not be parsed.
Parameters:: line – A line from the test case definition file.
Returns:: A new test case definition instance.

class testsystem.models.TestCaseTask(group: Group, priority: float, test_case_def: TestCaseDef, test_env: fs.TestEnv, test_unit: TestUnit | None = None, test_unit_tag: str | None = None, callback=None, error_callback=None)

Test case task instance, which is scheduled and executed by the test system.

run(): Override this method to implement the tasks functionallity.

property test_case: TestCase: The test case used by this task. This property is only valid after the task was executed.

class testsystem.models.TestCase(definition: TestCaseDef, group: Group, test_unit: TestUnit)

Instance of a test case definition which holds information for a specific test run.

compare_output(output) → bool

Method to compare the output of a test case with the expected output.

Returns:: True if the output is as expected, False otherwise.

class testsystem.models.TestResult(**kwargs)

Class for a specific test result produced by a test case. This is also a database object.

property contribute_to_score: bool: Returns true if this result should be considered in the result score.

property score: int: Returns the value this testcase contributes to the total score.

property tc_def: TestCaseDef | None: The test case definition for this result. This parameter would be none if the test case definition was deleted after the test result was created.

class testsystem.models.TestSet(**kwargs)

Data container for a test run and collection of test results. This is also a database object.

delete(): Delete this test set.

classmethod delete_unfinished_test_sets(): Delete all unfinished test sets. This should only be used during start-up and never be called during regular test system operation.

classmethod exists(group_id: int, commit_hash: str) → bool

Checks if a test set exists for a specific commit.

Parameters:

group_id – The group to check for the test set.
commit_hash – The commit to check for an existing test set.

Returns:

True if a test set exists and false otherwise.

classmethod get_or_create(group_id: int, commit_hash: str) → TestSet

Get or create a test set. If a test set for this group and commit already exists, it will be returned. Otherwise, a new instance is created.

Parameters:

group_id – The group for which this set should hold results.
commit_hash – The commit hash for which this set should hold results.

Returns:

A test set instance.

get_results() → list[TestResult]

Get a list of all results stored in this test set.

Returns:: A list of all results.

set_finished(state: bool = True)

Set the finished state of this test set.

Parameters:: state – If True, this set is marked finished.

update(commit_time: int, commit_msg: str | None = None)

Update this test set with additional information.

Parameters:

commit_time – The timestamp of the commit in unix millis.
commit_msg – The message of the commit used in this set.

class testsystem.models.TestUnit(msp: MSP430, pico: PicoScope | None = None, connections: list[ConnectionInfo] = [])

A Test Unit consists of an MSP430 board and a PicoScope with some predefined connections.

Parameters:

msp – The MSP430 used for this Test Unit.
pico – The PicoScope used for this Test Unit.
connections – A list of all available connections between the MSP430 and the PicoScope. This list must contain at least the required connections to construct a valid Test Unit.

static get() → list[TestUnit]: Get all active test units. This is a list of test units that were discovered during start-up.

static get_connections() → list[ConnectionInfo]

Get all available connections.

Returns:: All connections discovered during start-up.

classmethod get_required_connections() → list[ConnectionInfo]

Parses the configuration and returns a list of required connections with a port, a pin, and a channel, to form a valid Test Unit.

Exceptions ConfigError:: Is raised if the configuration is invalid.
Returns:: A list of required connections.

classmethod get_timing_channel() → int

Get the digital channel number for timing tests. This is the channel of the first testsystem.config.Config.tu_connections configuration.

Returns:: The digital channel number.

is_available() → bool

Check if this test unit is currently availabe.

Returns:: True if it is available, False otherwise.

static is_config_valid() → bool

Checks whether the required configuration for Test Units is valid.

Returns:: Returns True if the configuration is valid, False otherwise.

static set_connections(connections: list[ConnectionInfo])

Set globally available connections.

Parameters:: connections – All discoverd connections.

static validate_setup(msp: MSP430, picoscope: PicoScope | None, connections: list[ConnectionInfo]) → tuple[bool, list[ConnectionInfo]]

Checks if a specific MSP430 board and a PicoScope can be combined into a Test Unit.

Parameters:

msp – MSP430 board
picoscope – PicoScope
connections – All available connections between any devices. They are not restricted to the specific MSP430 or PicoScope provided as the previous parameters.

Returns:

The first parameter indicates if the MSP430 and the PicoScope can be combined into a Test Unit. The second parameter is a subset of connections between the two devices used to build a valid Test Unit.

class testsystem.models.UARTCapture(smpls_per_bit: int)

Container class that stores and decodes a single UART byte signal.

capture(data) → int

Push new signal data.

Parameters:: data – UART data.
Returns:: The length of the consumed data.

property complete: bool: Check if capturing a full byte is completed.

get_byte() → int | None

Get the byte value of the UART signal recorded by this capture.

Returns:: Byte or None for an invalid signal.

is_valid() → bool

Check if this capture is valid.

Returns:: True if it is valid, False otherwise.

Reporting

testsystem.reporting.create_md_group_report(test_set: TestSet | None = None, group: Group | None = None, group_nr: int | None = None) → str

This function creates a markdown group report. It contains an anonymized table of all groups and their scores and a detailed test result section for a specific group. This function requires at most one parameter.

Parameters:

test_set – A specific test set for which to create the report.
group – A specific group for which to create the report.
group_nr – The group number for which to create the report.

Returns:

Returns the markdown report as string.

Scheduling

testsystem.scheduling.get_next_task(worker: task_worker.TaskWorker) → Task | None

Get the next task from the schedule queue. If a task is returned it is also removed from the queue.

Parameters:: worker – The worker which will run the task.
Returns:: Returns the next task in queue for the specific task worker or None if nothing is to be done.

testsystem.scheduling.queue_size() → int

Get the current queue size. The result is not guaranteed to be still valid when read by the caller.

Returns:: Queue length.

testsystem.scheduling.schedule_task(task: Task) → int

Add a new task the the queue.

Parameters:: task – The task to schedule.
Returns:: Returns the current new size of the queue.

testsystem.scheduling.start(): Starts the scheduling thread.

testsystem.scheduling.stop(): Stops the scheduling thread. This call blocks until the scheduler stopped.

System

testsystem.system.idle(): Start the test system in idle mode. The test system won’t do anything in this mode.

testsystem.system.list_devices(): Get a list of all MSPs and PicoScopes that have ever been connected to the test system.

testsystem.system.startup_routine(): Execute only the startup routine. The test system will exit after startup.

Test Framework

Test System

class test_framework.test_system_model.TestSystemModel(name: str = 'unnamed', exercise_nr: int = 1, keep_repos: bool = False)

Setup a test system model for a specif test case.

Parameters:

name – An arbitrary name for the test environemt. You could use the name of the testcase.
exercise_nr – The exercise number which the test system should use.
kee_repos – Flag to disable deletion of repos when the test case finished.

get_config(property: str) → str

Get the value of a specific config property.

Parameters:: property – The property name from which to get the value.
Returns:: Returns the property value as string.

register_group(group_nr: int) → GroupModel

Register a group on the test system model. The test system will test the commits made by this group.

Parameters:: group_nr – The group number.
Returns:: A newly registered group model. Use its convenience methods to add actions for the group.

run(): Start and run the test system and execute group actions. This call blocks until the test system is shut down.

Update a config property for the test system.

Parameters:

property – The name of the property.
value – The new value for the property.

test_framework.test_system_model.setup_test_system(name: str = 'unnamed', exercise_nr: int = 1, keep_repos: bool = False) → TestSystemModel

Setup a test system model for a specif test case.

Parameters:

name – An arbitrary name for the test environemt. You could use the name of the testcase.
exercise_nr – The exercise number which the test system should use.
kee_repos – Flag to disable deletion of repos when the test case finished.

Returns:

A test system model instance. Use its register_group() method to add groups.

Group Module

class test_framework.group_model.GroupModel(group_nr: int, test_system: TestSystemModel)

A model to simulate a group’s behavior and verify the results returned by the test system. A group model starts with an initial git repository equal to the one provided by the contents of tests/integration_tests/git/RTOS_SS00_GroupInit.tar.gz. The model then performs some actions based on its configuration.

Parameters:

group_nr – An arbitrary group number.
test_system – The test system model where to register this group.

add_action(action: Action)

Add an action to the group model. Once the model is started, it performs the actions in the order in which they were added.

Parameters:: action – The action to add.

commit(tc_name: str, commit_msg: str | None = None) → CommitAction

Adds a commit action. The data to commit is specified by the parameter tc_name, which is the name of a directory in data/test_cases.

Parameters:

tc_name – The name of the directory containing the data to be commited.
commit_msg – An optional commit message. IF not set, the commit message will be the tc_name.

Returns:

Commit action.

initial_commit(tc_name: str, commit_msg: str | None = None) → InitialCommitAction

Adds a commit that is already present when the test system starts. The data to commit is specified by the parameter tc_name, which is the name of a directory in data/test_cases.

Parameters:

tc_name – The name of the directory containing the data to be commited.
commit_msg – An optional commit message. IF not set, the commit message will be the tc_name.

Returns:

Commit action.

start(): Starts the group model.

stop(): Stops the group model.

stop_test_system() → StopTestsystemAction: Adds an action to stop and shutdown the test system. You may not add actions after a stop testsystem action. They are not guaranteed to be executed.

verify(report: WaitReportAction | None = None) → VerifyAction

Adds a verification action. This action checks if a report fulfills the expected results. If no report parameter is specified, the previous report is used. The results for a commit and its expected report are configured in the result.json file in the test case directory data/test_cases.

Parameters:: report – Optional report to be verified.
Returns:: Verify action.

verify_any_report() → VerifyNoReportAction

Adds an action that will verify that no report is published.

Returns:: Verify no report action.

verify_no_report(commit: CommitAction) → VerifyNoReportAction

Adds an action that will verify that no report is published for a specific test case.

Parameters:: commit – Commit for which a report should not exist.
Returns:: Verify no report action.

wait(wait_time_s: float) → WaitAction

Adds a wait action.

Parameters:: wait_time_s – The time to wait in seconds.
Returns:: Wait action.

wait_report(commit: CommitAction | None = None) → WaitReportAction

Adds a wait for report action. This action waits unitl the test system published a report for a commit. If no commit parameter is specified, the previous commit is used.

Parameters:: commit – Optional commit of whose report is to be waited for.
Returns:: Wait report action

class test_framework.parsing.ExpectedResult(**kwargs)

Data class that contains information about what to expect from the test system. Add a JSON object of this class to each test case directory with the facts to verify in reports. A valid starting point for this JSON object could look like this:

{
    "total": 5,                     // All test cases from ex1 should pass.
    "test_cases": [                 // But only one detailed result is asserted
        {
            "id": 101,              // *sleep 100 ms* test case
            "successful": true,
            "result": 100000,
            "tolerance": 0.0001     // Acceptable results: [99990, 100010]
        }
    ]
}

Note

All parameters in the JSON file (except the test case id) are optional. You may only use the once that are useful in a specific scenario.

Parameters:

total (int) – The expected total result. Default value: 0
test_cases (list[ExpectedTCResult]) – A list of detailed RTOS test case results. This list does not require each RTOS test case but only the ones that are interesting for this specific test case. Default value: []

class test_framework.parsing.ExpectedTCResult(id: int, **kwargs)

Data class that contains information about what to expect from a single RTOS test case.

Parameters:

id – The test case id as defined in the testcases.txt file.
successful (bool | None) – Flag if the test case must be successful. Default value: None
result (int | float | None) – The expected result value for this test case. Default value: None
tolerance (float | None) – A tolerance margin for the result value between 0 and 1. This parameter could be helpful if you expect a timing result of ideally 100ms but want the test case to pass for values between 95ms and 105ms. In this case, you can set the tolerance property to 0.05. Default value: None

Actions

class test_framework.Action

This is the base class for all actions executed by a group model. It has some convenience methods to chain action methods for a group.

Parameters:: group – Reference to this actions group model.

class test_framework.CommitAction(tc_name: str, commit_msg: str | None = None)

Action to create a new commit. The new commit is based on the data of a test case found in the data/test_cases subdirectory of the integration tests directory. The directory structure should look as follows:

/                                           : Reporitory root directory
└── tests/integration_tests/data/test_cases : Directory that contains test cases
    └── <tc_name>                           : A specific test case
        ├── data                            : This repository usually contains an
        |   ├── [apps]                      :   RTOS implementation but could include
        |   ├── [middleware]                :   anything a group should commit.
        |   └── ...                         :
        └── result.json                     : The expected results of this test case.

The result.json file should be a JSON object of type ExpectedResult.

Parameters:

tc_name – The name of the test case directory containing the data to be commited.
commit_msg – The commit message for the new commit.

class test_framework.InitialCommitAction(tc_name: str, group: tf.GroupModel, commit_msg: str | None = None)

Action to create a new commit that gets pushed immediately. This guarantees that the commit is already present when the test system starts.

Parameters:

tc_name – The name of the test case directory containing the data to be commited.
commit_msg – The message of the commit.

class test_framework.StopTestsystemAction: This action stops the test system. It uses the stop property in the configuration to shut down the test system instance. You may not add actions after a stop testsystem action. They are not guaranteed to be executed.

class test_framework.VerifyAction(report_action: WaitReportAction)

This action verifies that the results from a report match the expected values. To see what the expected results are and what data this action verifies, look at the ExpectedResult class.

Parameters:: report_action – The report that should be verified.

class test_framework.VerifyNoReportAction(commit: tf.CommitAction | None = None)

Verifies that specific reports do not exist. It will assert that no report is published for a specific test case if the commit parameter is set. If no parameter is given, this action will verify that any report is published.

Parameters:: commit – Optional commit for which a report should not exist.

class test_framework.WaitAction(wait_time_s: float)

class test_framework.WaitReportAction(commit: CommitAction, stop_event: threading.Event | None = None)

Testing

testsystem.testing.measure_size(tc: TestCase): Measure the size for a size test case.

testsystem.testing.measure_timing(tc: TestCase): Measure the timing for a timing test case.

testsystem.testing.run_compare_test(tc: TestCase, retries: int = 3): Run a standard (compare) test case.

testsystem.testing.run_size_test(tc: TestCase): Run a size test case.

testsystem.testing.run_test(tc: TestCase, test_env: TestEnv): Run a test case.

testsystem.testing.run_timing_test(tc: TestCase): Run a timing test case.

Tool Chain

testsystem.tool_chain.build(src_dir: str, args: list[str]) → str

Build program with make..

Raises:: BuildError – If building failed.
Parameters:: src_dir – The directory of the makefile.
Returns:: The output from the build task.

testsystem.tool_chain.build_test_case(tc: TestCase)

Build test case from source.

Raises:: BuildError – If building failed.
Parameters:: tc – Test case.

testsystem.tool_chain.flash(msp: MSP430, file: str) → str

Flash a hex file onto an MSP430.

Raises:

FlashError – If flashing failed.
MSPConnectionError – If connection to MSP430 failed.
FirmewareError – If the msp requires a firmeware update.

Parameters:

file – Hex file to flash to MSP.

Returns:

The stdout from the flash task.

testsystem.tool_chain.flash_test_case(tc: TestCase)

Flash a test case onto an MSP430.

Raises:

FlashError – If flashing failed.
MSPConnectionError – If connection to MSP430 failed.
FirmewareError – If the msp requires a firmeware update.

Parameters:

tc – Test case to flash.

Utils

testsystem.utils.run_external_task(args, input=None, timeout=10) → tuple[int, str, str]

Interface to safely run an external task. This function should be used any time the test system requires an additional tool.

Parameters:

args – Arguments to the external task.
input – An optional input after the task is stared.
timeout – A timeout when to kill the external task and return to the test system.

Returns:

The first parameter is the return code from the process. The second parameter is the standard output of the process and the last parameter is the error output of the process.

testsystem.utils.to_bool(value: int | str) → bool

Convert an integer or string to bool.

Parameters:: value – Bool variable as integer or string type.
Returns:: Actual bool value.

testsystem.utils.to_local_time_str(timestamp: float | int, strftime: str = '%d.%m.%Y %H:%M:%S') → str

Converts a timestamp to the local time format. This should be used any time a conversion from a timestamp to a readable date time is required.

Parameters:

timestamp – Timestamp in unix millis.
strftime – A format string for the date time convertion.

Returns:

The timestamp as date time string.