moler.util package

Submodules

moler.util.cmds_events_doc module

Perform command autotest for selected command(s).

moler.util.cmds_events_doc.check_cmd_or_event(path2cmds)
moler.util.cmds_events_doc.check_if_documentation_exists(path2cmds)

Check if documentation exists and has proper structure.

Parameters:path2cmds (str) – relative path to comands directory
Returns:True if all checks passed
Return type:bool

moler.util.connection_observer module

Utilities related to connection-observers

class moler.util.connection_observer.exception_stored_if_not_main_thread(connection_observer, logger=None)

Bases: object

Context manager storing exception inside connection-observer for non-main threads

moler.util.connection_observer.inside_main_thread()

moler.util.connection_observer_life_status module

class moler.util.connection_observer_life_status.ConnectionObserverLifeStatus

Bases: object

moler.util.converterhelper module

Units converter

class moler.util.converterhelper.ConverterHelper

Bases: object

static get_converter_helper()
to_bytes(str_bytes, binary_multipliers=True)

Method to convert size with unit to size in bytes :param str_bytes: String with bytes and optional unit :param binary_multipliers: If True then binary multipliers will be used, if False then decimal :return: 3 values: int value in bytes, float value in units (parsed form string), String unit

to_number(value, raise_exception=True)

Convert number to int or float.

Parameters:
  • value – string with number inside
  • raise_exception – if True then raise exception if cannot convert to number, If False then return 0.
Returns:

int or float with value.

to_seconds(value, unit)

Method to convert number of :param value: numeric value in units :param unit: Unit of time, h for hour :return: number of seconds

to_seconds_str(str_time)

Method to convert string time with unit to seconds. :param str_time: String with time and unit. :return: 3 values: float value in seconds, float value in units (parsed form string), String unit

moler.util.devices_SM module

Perform devices SM autotest.

class moler.util.devices_SM.RemoteConnection(moler_connection, echo=True, name=None, logger_name='')

Bases: moler.io.raw.memory.ThreadedFifoBuffer

remote_inject_response(input_strings)

Simulate remote endpoint that sends response. Response is given as strings.

send(input_bytes)

What is written to connection comes back on read() only if we simulate echo service of remote end.

set_device(device)

Need to get actual state of device when sending cmds response.

write(input_bytes)

What is written to connection comes back on read() only if we simulate echo service of remote end.

moler.util.devices_SM.get_cloned_device(src_device, new_name, new_connection)
moler.util.devices_SM.get_device(name, connection, device_output, test_file_path)
moler.util.devices_SM.get_memory_device_connection()
moler.util.devices_SM.iterate_over_device_states(device, max_time=None, tests_per_device=300, max_no_of_threads=11, rerun=0, timeout_multiply=3.0)

Check all states in device under test. :param device: device :param max_time: maximum time of check. None for infinity. If execution time is greater then max_time then test is

interrupted.
Parameters:
  • tests_per_device – how many tests should be performed in one thread.
  • max_no_of_threads – maximum number of threads that can be started.
  • rerun – rerun for goto_state
  • timeout_multiply – timeout_multiply for goto_state
Returns:

None

moler.util.loghelper module

Utilities related to logging

moler.util.loghelper.debug_into_logger(logger, msg, extra=None, levels_to_go_up=0)
moler.util.loghelper.disabled_logging(from_level_and_below=20)
moler.util.loghelper.error_into_logger(logger, msg, extra=None, levels_to_go_up=0)
moler.util.loghelper.find_caller(levels_to_go_up=0)

Find the stack frame of the caller so that we can note the source file name, line number and function name.

Based on findCaller() from logging module but allows to go higher back on stack

Parameters:levels_to_go_up – 0 - info about ‘calling location’ of caller of findCaller(); 1 - ‘calling -1 location’
Returns:
moler.util.loghelper.info_into_logger(logger, msg, extra=None, levels_to_go_up=0)
moler.util.loghelper.log_into_logger(logger, level, msg, extra=None, levels_to_go_up=0)

Log into specific logger

No support for logging exceptions

Parameters:
  • logger – logger to send log record into
  • level – logging level
  • msg – message to be logged
  • levels_to_go_up – 0 - info about function using log_into_logger(), 1 - caller of that function, …
Returns:

None

moler.util.loghelper.warning_into_logger(logger, msg, extra=None, levels_to_go_up=0)

moler.util.moler_serial_proxy module

class moler.util.moler_serial_proxy.AtConsoleProxy(port, verbose=False, at_echo=False)

Bases: object

Class to proxy AT commands console into stdin/stdout

close()

Close underlying serial connection.

open()

Open underlying serial connection.

Return context manager to allow for: with connection.open() as conn:

send(cmd)

Send data over underlying serial connection

class moler.util.moler_serial_proxy.AtToStdout(prefix, verbose=False)

Bases: serial.threaded.LineReader

ATserial->stdout

await_response_event(timeout=4)

Await till reading thread sets found event

connection_lost(exc)

Forget transport

connection_made(transport)

Store transport

handle_line(line)

Process one line - to be overridden by subclassing

send(line)
send_and_await_response(line, response, timeout=4)

Await till data comes

class moler.util.moler_serial_proxy.IOSerial(port, baudrate=115200, stopbits=1, parity='N', timeout=2, xonxoff=1)

Bases: object

Serial-IO connection.

close()

Close established connection.

open()

Take ‘how to establish connection’ info from constructor and open that connection.

Return context manager to allow for: with connection.open() as conn:

read()

Returns data read from serial connection

send(cmd)

Sends data over serial connection

class moler.util.moler_serial_proxy.IoThreadedSerial(port, protocol_factory, baudrate=115200, stopbits=1, parity='N', timeout=2, xonxoff=1)

Bases: moler.util.moler_serial_proxy.IOSerial

Threaded Serial-IO connection.

close()

Close the serial port and exit reader thread

open()

Take ‘how to establish connection’ info from constructor and open that connection.

Return context manager to allow for: with connection.open() as conn:

send(cmd)

Sends data over serial connection

send_and_await_response(line, response, timeout=4)

moler.util.moler_test module

Utility/common code of library.

class moler.util.moler_test.MolerTest

Bases: object

classmethod error(msg, raise_exception=False, dump=None)

Makes an error (fail the test) and (optional) continue the test flow. :param msg: Message to show. :param raise_exception: If True then raise an exception (if not in try except block then test will be

terminated), if False then only show msg and mark error in logs.
Parameters:dump – If defined then dump object.
Returns:None.
classmethod info(msg, dump=None)

Shows the message :param msg: Message to show. :param dump: If defined then dump object. :return: None.

classmethod raise_background_exceptions(decorated='function', check_steps_end=False)

Decorates the function, method or class. :param decorated: Function, method or class to decorate. :param check_steps_end: If True then check if steps_end was called before return the method, if False then do

not check
Returns:Decorated callable
classmethod sleep(seconds, quiet=False)

Add sleep functionality TODO: add support to asyncio when runner ready :param seconds: Time to sleep (in seconds) :param quiet: If True then no info to log about sleeping, if False then sleep info will be logged :return:

classmethod steps_end()

You should call this function at the end of your test code with Moler. :return: None

classmethod stop_python(force=False)

Stops current Python. :return: None

classmethod warning(msg, dump=None)

Shows the message as warning. :param msg: Message to show. :param dump: If defined then dump object. :return: None

Module contents