openpectus.test.engine.utility_methods

Attributes

`logger`
`UodFactory`
`RunCondition`
`EventName`	Defines the awaitable events of the test engine runner
`InstructionName`	Defines the awaitable instructions of the test engine runner
`FindInstructionState`	Defines the awaitable instruction states of the test engine runner
`last_tick_time`
`interval`

Classes

`EngineTestRunner`	Expose an interface of Engine similar to what is available in the frontend to tests.
`EngineTestInstance`	Defines the listener interface and base class for engine life-time events.

Functions

`build_program`(pcode)
`run_engine`(engine, pcode[, max_ticks])
`continue_engine`(engine[, max_ticks])
`print_runlog`(e[, description])
`print_runtime_records`(e[, description])
`clear_log_config`()	Remove any existing logging setup, eg. from logging.basicConfig()
`configure_test_logger`()
`set_engine_debug_logging`()
`set_interpreter_debug_logging`([include_events, ...])

Module Contents

openpectus.test.engine.utility_methods.logger

openpectus.test.engine.utility_methods.build_program(pcode)

Parameters:: pcode (str)
Return type:: openpectus.lang.model.ast.ProgramNode

openpectus.test.engine.utility_methods.UodFactory

openpectus.test.engine.utility_methods.RunCondition

openpectus.test.engine.utility_methods.EventName: Defines the awaitable events of the test engine runner

openpectus.test.engine.utility_methods.InstructionName: Defines the awaitable instructions of the test engine runner

openpectus.test.engine.utility_methods.FindInstructionState: Defines the awaitable instruction states of the test engine runner

class openpectus.test.engine.utility_methods.EngineTestRunner(uod_factory, method='', interval=0.1, speed=1.0)

Expose an interface of Engine similar to what is available in the frontend to tests.

Parameters:

uod_factory (UodFactory)
method (str | openpectus.protocol.models.Method)
interval (float)
speed (float)

uod_factory

method: str | openpectus.protocol.models.Method = ''

interval = 0.1

speed = 1.0

run()

Return type:: Generator[EngineTestInstance, None, None]

class openpectus.test.engine.utility_methods.EngineTestInstance(engine, method, timing)

Bases: openpectus.lang.exec.events.EventListener

Defines the listener interface and base class for engine life-time events.

Parameters:

engine (openpectus.engine.engine.Engine)
method (str | openpectus.protocol.models.Method)
timing (openpectus.engine.engine.EngineTiming)

engine

timing

_search_index = 0

_last_event: EventName | None = None

start(): Schedules the Start command. run_* is required to actually run it

property marks: list[str]

Return type:: list[str]

property method_manager: openpectus.engine.method_manager.MethodManager

Return type:: openpectus.engine.method_manager.MethodManager

run_until_condition(condition, max_ticks=30)

Continue program until condition occurs. Return the number of ticks spent.

Raises TimeoutError if the condition is not met before max_ticks is reached.

Parameters:: condition (RunCondition)
Return type:: int

run_until_instruction(instruction_name, state='any', arguments=None, max_ticks=30, increment_index=True)

Continue program execution and wait until the given instruction is run.

Use the state argument to specify how to wait. The default is ‘any’ which waits for any record for the given instruction. Note that for some instructions this happens much earlier than the start state, e.g. for Watch. This can be changed to e.g. started to wait until the instruction is started.

Return the number of ticks spent.

Raises TimeoutError if the instruction is not found before max_ticks is reached.

Use increment_index to control searching within states of the same instruction as last search. The default value of True skips to the next instruction which is normally what you want. If set to False, search also includes the record of the previous match. This is useful if searching for different states for the same record, such as “started” and “completed”. See also index_step_back().

Parameters:

instruction_name (InstructionName)
state (FindInstructionState)
arguments (str | None)

Return type:

int

index_step_back(steps=1)

_clear_last_event()

run_until_event(event_name, max_ticks=100)

Continue program execution and wait until the given event fires.

Return the number of ticks spent.

Raises TimeoutError if the instruction is not found before max_ticks is reached.

Parameters:: event_name (EventName)
Return type:: int

restart_and_run_until_started()

Restart the method and wait a few ticks for engine to come back up.

Return type:: None

run_ticks(ticks)

Continue program execution until te specified number of ticks.

Parameters:: ticks (int)
Return type:: None

get_runtime_table(description='')

Return a text view of the runtime table contents.

Parameters:: description (str)
Return type:: str

on_engine_configured(): Invoked once on engine startup, after configuration and after the connection to hardware has been established.

on_start(run_id)

Is invoked by the Start command when method is started.

Parameters:: run_id (str)

on_block_start(block_info)

Invoked just after a new block is started, before on_tick, in the same engine tick.

Parameters:: block_info (openpectus.lang.exec.events.BlockInfo)

on_block_end(block_info, new_block_info)

Invoked just after a block is completed, before on_tick, in the same engine tick. If the new block is the root/program scope, the name is the empty string.

Parameters:

block_info (openpectus.lang.exec.events.BlockInfo)
new_block_info (openpectus.lang.exec.events.BlockInfo | None)

on_tick(tick_time, increment_time)

Is invoked on each tick.

Intended for NA (calculated/derived) tags to calculate the value for the tick and apply it to the value property.

Parameters:

tick_time (float)
increment_time (float)

on_method_end(): Is invoked when interpretation of the last method line is complete. Note: This does not necessarily mean that interpretation ends: - Any alarms still have their condition evaluated and may execute - The method may be modified by the user and the new lines will start executing

on_stop(): Is invoked by the Stop command when method is stopped.

on_engine_shutdown(): Invoked once when engine shuts down

openpectus.test.engine.utility_methods.last_tick_time = 0.0

openpectus.test.engine.utility_methods.interval = 0.1

openpectus.test.engine.utility_methods.run_engine(engine, pcode, max_ticks=-1)

Parameters:

engine (openpectus.engine.engine.Engine)
pcode (str)
max_ticks (int)

Return type:

int

openpectus.test.engine.utility_methods.continue_engine(engine, max_ticks=-1)

Parameters:

engine (openpectus.engine.engine.Engine)
max_ticks (int)

Return type:

int

openpectus.test.engine.utility_methods.print_runlog(e, description='')

Parameters:: e (openpectus.engine.engine.Engine)

openpectus.test.engine.utility_methods.print_runtime_records(e, description='')

Parameters:

e (openpectus.engine.engine.Engine)
description (str)

openpectus.test.engine.utility_methods.clear_log_config(): Remove any existing logging setup, eg. from logging.basicConfig()

openpectus.test.engine.utility_methods.configure_test_logger()

openpectus.test.engine.utility_methods.set_engine_debug_logging()

openpectus.test.engine.utility_methods.set_interpreter_debug_logging(include_events=False, include_runlog=False)