Skip to content

Test API

Test scripts import their API from can_explorer.testing:

from can_explorer.testing import TestModule, testcase

See Test automation for how test scripts are written and run.

Test modules and test cases

TestModule

Base class of test modules. Subclass it in a test script and mark its test methods with @testcase.

The docstring describes the module in the results. setup runs before the first test case of the module and teardown after the last one; a failing setup marks all test cases of the module as errors.

Attributes:

Name Type Description
requirements Union[str, Iterable[str]]

Requirement identifiers all test cases of the module verify (besides their own).

uds_request_id Optional[int]

CAN identifier of the UDS requests (t.uds), e.g. 0x7E0.

uds_response_id Optional[int]

CAN identifier of the UDS responses, e.g. 0x7E8.

uds_extended bool

Whether the UDS identifiers are extended (29 bit).

setup(t) async

Runs before the first test case of the module.

teardown(t) async

Runs after the last test case of the module, also when test cases failed.

testcase(function=None, *, title=None, timeout=30.0, params=None, ids=None, requirements=())

Mark a method of a TestModule as test case. Test cases run in the order they are defined.

Examples:

@testcase
async def engine_speed_is_sent(self, t): ...

@testcase(title='Brake lights follow the pedal', timeout=10, requirements='REQ-BODY-012')
async def brake_lights(self, t): ...

@testcase(params=[800, 2000, 4000])                      # One test case per value
async def idle_and_cruise(self, t, speed): ...

@testcase(params=[{'gear': 1, 'limit': 50}, {'gear': 5, 'limit': 180}], ids=['first', 'fifth'])
async def speed_limit(self, t, gear, limit): ...

Parameters:

Name Type Description Default
title Optional[str]

Title in the results; defaults to the method name with spaces.

None
timeout float

The test case ends with an error when it runs longer (seconds).

30.0
params Optional[Iterable[Any]]

Run the test case once per item: a dict is passed as keyword arguments, a tuple as positional arguments, anything else as one argument after t. Each run is a test case of its own, named method[id].

None
ids Optional[Iterable[str]]

Names of the parameter sets in the results; by default the values, e.g. gear=1,limit=50.

None
requirements Union[str, Iterable[str]]

Requirement identifiers the test case verifies, e.g. 'REQ-ENG-001' or a list of them. The reports list the verdict per requirement.

()

Test context

TestContext(runner, module)

The t of a test case: checks, sending, waiting for frames and signals, timing and diagnostics.

Checks record a step and continue the test case; require and fail end it. Waits see the frames that arrive after they start, and, for the first wait after send, also those that arrived since the send (responses can come before the wait starts). Frames the test sent itself are never seen.

elapsed property

Seconds since the test case (or setup/teardown) started.

variables property

Shared variables of the simulation of the connection (e.g. t.variables['ignition'] = True).

uds property

UDS client (can_explorer.core.uds.UdsClient) on the module's uds_request_id/uds_response_id.

Examples:

vin = await t.uds.read_data_by_identifier(0xF190)
await t.uds.diagnostic_session_control(3)
result = await t.uds.request(bytes([0x22, 0xF1, 0x90]))   # Negative responses as result

log(*values)

Add a line to the test case's log.

check(condition, description)

Record a check. A failed check fails the test case but lets it continue. Returns the result.

check_equal(actual, expected, description='', tolerance=0.0)

Check that actual equals expected (numbers within tolerance).

check_range(value, minimum, maximum, description='')

Check that minimum <= value <= maximum.

require(condition, description)

Like check, but a failure ends the test case.

fail(description)

End the test case as failed.

skip(reason)

End the test case as skipped, e.g. when the device under test lacks a feature.

sleep(seconds) async

message(name)

A buffer with the signals of a database message, to fill and send.

send(message, data=None, *, extended=None, fd=False, remote=False, **signals)

Send a frame: t.send('BrakeCommand', Pressure=50), t.send(buffer) or t.send(0x123, b'\x01').

wait_for_message(selector=None, timeout=1.0, condition=None) async

The next frame matching selector (and condition), or None after timeout seconds.

Examples:

msg = await t.wait_for_message('EngineData')
msg = await t.wait_for_message(0x7E8, timeout=0.5)
msg = await t.wait_for_message('EngineData', condition=lambda m: m.signals['EngineSpeed'] > 2000)

expect_message(selector, timeout=1.0, condition=None, description='') async

Wait for a frame and record whether it came in time.

expect_no_message(selector, duration=1.0, description='') async

Check that no matching frame arrives for duration seconds.

wait_for_signal(key, expected=None, timeout=1.0) async

Wait until signal key ("Message.Signal") has a value: any value, expected, or one for which the function expected returns True. Returns the value, or None after timeout seconds.

Value table texts can be compared too: await t.wait_for_signal('EngineData.GearPosition', 'First').

expect_signal(key, expected=None, timeout=1.0, description='') async

Wait for a signal value (see wait_for_signal) and record whether it came in time.

collect(selector=None, duration=1.0) async

All matching frames that arrive within duration seconds.

measure_cycle_time(selector, duration=1.0) async

Measure the cycle time of a message for duration seconds.

expect_cycle_time(selector, expected_ms, tolerance=0.1, duration=1.0) async

Check that the mean cycle time is within tolerance (a fraction, 0.1 = ±10 %) of expected_ms.

Results

CycleTimes(count, mean=math.nan, minimum=math.nan, maximum=math.nan, deviation=math.nan) dataclass

Cycle time statistics of a message in milliseconds; NaN with fewer than two frames.

Verdict

Received frames are ReceivedMessage objects, as in simulation scripts.