Third party modules

Arpings checking helpers

stepler.third_party.arping.arping(ip, iface, remote, count=None, latency=2)

Non-blocking context manager for run arping on background.

It yields ping results (dict) and update it with ‘sent’ and ‘received’ values after CM will be exited.

Parameters:

• ip (str) – ip to arping
• iface (string) – name of interface, like ‘eth0’
• remote (obj) – instance of stepler.third_party.ssh.SshClient
• count (int, optional) – Count of packets to send. By default, arping will send packets until termination
• latency (int, optional) – time to wait before arping will be terminated

Yields:

dict – arping results

Chunk serializer

Nova instance metadata has restriction - keys and values of it can contains not more than 255 symbols. This serializer dumps passed metadata to json, split to to small chunks and makes a dict with this chunks.

stepler.third_party.chunk_serializer.dump(obj, prefix)

Transform object to dict with small chunks of jsoned object.

Example

>>> dump({'keypair': 'a' * 260}, prefix='some_prefix_')
{
    'some_prefix_0':
        '{"keypair": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
        # cut
        aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
    'some_prefix_1': 'aaaaaaaaaaaaaaaaaa"}'
}

Parameters:

• obj (object) – object to serialize. Should be json-serializable
• prefix (str) – prefix to result dict keys

Returns:

dict with small chunks of json object representation

Return type:

dict

stepler.third_party.chunk_serializer.load(meta, prefix)

Restore object from dict, created with dump function.

Parameters:

• meta (dict) – dict with serialized object
• prefix (str) – serialized records keys prefix

Returns:

deserialized object

Return type:

object

Custom context manager generator

stepler.third_party.context.context(func)

Decorator to make context manager from generator with guaranteed finalization.

Note

contextlib.contextmanager doesn’t guarantee context manager finalization and requires usage of try-finally for that. But in fixtures it needs to rid of try-finally and to guarantee context manager finalization after yield. This decorator makes that.

Example

@pytest.fixture
def create_server_context(server_steps):

    @context
    def _create_server_context(server_name, *args, **kwgs):
        server = server_steps.create_server(server_name,
                                            *args, **kwgs)
        yield server
        server_steps.delete_server(server)

    return _create_server_context

See also

1. Exception inside context manager:

   @context
   def x():
       yield
   
   with x():
       raise Exception('error')
   

   Exception: error will be raised.

2. Exception in context manager finalization:

   @context
   def x():
       yield
       raise Exception('final')
   
   with x():
       pass
   

   Exception: final will be raised.

3. Exceptions inside context manager and in finalization:

   @context
   def x():
       yield
       raise Exception('final')
   
   with x():
       raise Exception('error')
   

   Exception: error will be raise as root cause.

Pytest plugin to add mark @pytest.mark.idempotent_id(<id>)

stepler.third_party.idempotent_id.get_item_id(item)

Return item (test) idempotent id.

stepler.third_party.idempotent_id.pytest_addoption(parser)

Add option to pytest.

stepler.third_party.idempotent_id.pytest_collection_modifyitems(session, items)

Add marker to test name, if test marked with idempotent_id marker.

If optional kwargs passed - test parameters should be a superset of this kwargs to mark be applied. Also kwargs can be passed as params argument.

stepler.third_party.idempotent_id.pytest_runtestloop(session)

Check idempotent id presence.

Logger for steps

stepler.third_party.logger.log(func)

Decorator to log function with arguments and execution time.

Network checks

stepler.third_party.network_checks.check_tcp_connect(ip, port=22, timeout=1)

Check whether TCP connection to ip with port.

Parameters:

• ip (str) – ip to establish connect to
• port (int, optional) – TCP port
• timeout (int, optional) – socket timeout to wait connection.

Returns:

is connection can be established or not

Return type:

bool

Pytest plugin to show output error message if no tests are found

It shows error message in stdout if no tests are found according to input parameters. For example:

No tests are found matching input parameters: keyword expression
'test1 and not test2', mark expression 'destructive'.

stepler.third_party.no_tests_found.pytest_collection_modifyitems(config, items)

Hook to show error message if no tests are found.

Collection of utilities for parsing CLI clients output

stepler.third_party.output_parser.listing(output_lines)

Return list of dicts with basic item info parsed from cli output.

stepler.third_party.output_parser.table(output_lines)

Parse single table from cli output. Return dict with list of column names in ‘headers’ key and rows in ‘values’ key.

stepler.third_party.output_parser.tables(output_lines)

Find all ascii-tables in output and parse them.

Return list of tables parsed from cli output as dicts. (see OutputParser.table()) And, if found, label key (separated line preceding the table) is added to each tables dict.

Pings checking helpers

class stepler.third_party.ping.FixedIDPinger(ip_to_ping, remote, icmp_id, command_path='ping')

Pinger class to work with modified ping implementation.

Custom implementation allows to set ID for ICMP requests.

class stepler.third_party.ping.PingResult

Ping result class.

Useful for object-oriented access to results of ping (such as transmitted, received, loss counts)

class stepler.third_party.ping.Pinger(ip_to_ping, remote=None)

Pinger class to call ping and return result.

Can be used directly (as Pinger.ping) and as non-blocking context manager.

Example

>>> with Pinger('10.109.8.2') as result:
...     some_action()
>>> print(result.loss)
0

>>> result = Pinger('10.0.0.1', remote=remote).ping(count=3)
>>> print(result.loss)
0

ping(count=1)

Start ping command and return result.

Parameters:count (int) – count of pings to send Returns:instance of PingResult Return type:object

Pytest plugin to mark and choose platform-specific tests

Usage example:

from stepler.third_party.supported_platforms import platform

@platform.mos10
def test_something():
    pass

Launching example:

py.test stepler --platform mos10

Supported platforms are mos10, mk2x, mcp.

stepler.third_party.supported_platforms.pytest_addoption(parser)

Add option --platform to choose platform-specific tests.

stepler.third_party.supported_platforms.pytest_collection_modifyitems(session, items)

Skip tests for unspecified platforms.

Notes

• If platform isn’t specified, only common tests will be executed.
• If platform is specified, common tests with platform-specific tests will be executed.

Interprocess locker

class stepler.third_party.process_mutex.Lock(filename)

Process mutex.

acquire()

Acquire lock.

release()

Release lock.

Pytest plugin to clean test reports

It ensures two things:

• Remove test reports folder before tests launching
• Remove test report folder if test is passed

stepler.third_party.reports_cleaner.pytest_configure(config)

Pytest hook to remove test reports before tests launching.

stepler.third_party.reports_cleaner.pytest_runtest_makereport(item, call)

Pytest hook to remove test report if test is passed.

SSH client

class stepler.third_party.ssh.SshClient(host, port=22, username=None, password=None, pkey=None, timeout=None, proxy_cmd=None)

SSH client.

background_call(command, stdout='/dev/null', stderr='&1')

Start long-running command in background and return it’s pid.

Parameters:

• command (str) – command to execute
• stdout (str) – path to file to redirect command stdout to
• stderr (str, optional) – path to file to redirect command stderr to. By default stderr combines with stdout.

Returns:

pid of running command

Return type:

str

Raises:

AssertionError – if command is not running in background

check()

Check SSH connection.

check_call(command, verbose=False)

Call command and check that exit_code is 0.

Parameters:

• command (str) – command to execute
• verbose (bool) – make log records or not

Returns:

CommandResult instance

Return type:

object

Raises:

Exception – if command exit_code is not 0

check_process_present(name)

Check that name is present in ps aux output.

Parameters:name (str) – command name to search Raises:Exception – if name is not found.

check_stderr(command, verbose=False)

Call command and check that stderr is empty.

Parameters:

• command (str) – command to execute
• verbose (bool) – make log records or not

Returns:

CommandResult instance

Return type:

object

Raises:

Exception – if command stderr is not empty

close()

Close ssh connection.

connect()

Connect to ssh server.

execute(command, merge_stderr=False, verbose=False, timeout=None)

Execute command and returns CommandResult instance.

Parameters:

• command (str) – command to execute
• merge_stderr (bool) – merge stderr to stdout
• verbose (bool) – make log records or not
• timeout (int, optional) – maximum command executing time in seconds

Returns:

CommandResult instance

Return type:

object

Raises:

ExecutionTimeout – if command executing more than timeout

execute_async(command, merge_stderr=False, verbose=False)

Start executing command async.

Parameters:

• command (str) – command to execute
• merge_stderr (bool) – merge stderr to stdout
• verbose (bool) – make log records or not

Returns:

SSH session, file-like stdin, stdout, stderr

Return type:

tuple

kill_process(name)

Kill all processes by killall <name>

Parameters:name (str) – command name to search Returns:CommandResult for execute command Return type:object

open(path, mode='r')

Open remote file with SFTP.

Parameters:

• path (str) – path to remote file
• mode (str, optional) – file open mode. The arguments are the same as for Python’s built-in open. Read-only by default.

Yields:

obj – an SFTPFile object representing the open file

sudo()

Context manager to run command with sudo.

wait_process_done(pid, timeout=0)

Wait until command with pid will be done.

Parameters:

• pid (int|str) – pid
• timeout (int, optional) – time to wait for process to be done

Raises:

ExecutionTimeout – if process executing after timeout.

Pytest plugin to check consistency

• steps and fixtures only are called inside a test
• steps must have format described in STEPS architecture

Checker can be disabled via py.test key --disable-steps-checker.

Checker can be disabled with comments:

def test_inline():
    foo = 'bar'
    baz = non_permitted_call(
        foo
    )  # checker: disable


def test_block():
    # checker: disable
    baz = non_permitted_call_1()
    baz = non_permitted_call_2()
    # checker: enable

stepler.third_party.steps_checker.pytest_addoption(parser)

Hook to register checker options.

stepler.third_party.steps_checker.pytest_collection_modifyitems(config, items)

Hook to detect forbidden calls inside test.

stepler.third_party.steps_checker.pytest_configure(config)

Hook to check steps consistency.

stepler.third_party.steps_checker.step(func)

Decorator to append step method name to permitted calls.

Parameters:func (function) – function to add its name to list with permitted calls Returns:the same function wrapped to log Return type:function

tcpdump helpers

stepler.third_party.tcpdump.filter_icmp(packet)

Returns True if packet contains ICMP layer.

stepler.third_party.tcpdump.filter_vxlan(packet)

Returns True if packet contains VxLAN layer.

stepler.third_party.tcpdump.get_last_ping_reply_ts(path)

Returns last ICMP echo response timestamp.

If there are no replies in packets - it returns None.

Parameters:packets (list) – list packets Returns:last ICMP reply timestamp or None Return type:float|None

stepler.third_party.tcpdump.read_pcap(path, lfilter=None)

Read pcap file and yields packets.

Parameters:

• path (str) – path to pcap file
• lfilter (function, optional) – function to filter returned packets. By default all packets will be returned.

Yields:

obj – packet

stepler.third_party.tcpdump.tcpdump(remote, args='', prefix=None, latency=2, lfilter=None)

Non-blocking context manager for run tcpdump on backgroud.

It yields path to pcap file.

Parameters:

• remote (SshClient) – instance of ssh client
• args (str, optional) – additional tcpdump args
• prefix (str, optional) – prefix for command. It can be useful for executing tcpdump on ip namespace.
• latency (int, optional) – time to wait after tcpdump’s starting and before tcpdump’s termination to guarantee that all packets will be captured
• lfilter (function, optional) – function to filter returned packets. By default all packets will be returned.

Yields:

str – path to pcap file

Utils

class stepler.third_party.utils.AttrDict(*args, **kwgs)

Wrapper over attrdict to provide context manager to update fields.

put(**kwgs)

Put fields to update in buffer.

stepler.third_party.utils.generate_ids(prefix=None, postfix=None, count=1, length=None, use_unicode=False, _stepler_prefix=None)

Generate unique identificators, based on UUID.

Parameters:

• prefix (str|None) – prefix of unique ids
• postfix (str|None) – postfix of unique ids
• count (int) – count of unique ids
• length (int|None) – length of unique ids
• use_unicode (boolean|False) – generate str with unicode or not
• _stepler_prefix (str, optional) – Resources prefix is used to call generate_ids inside stepler.config and avoid cross imports problem. By default it has value stepler.config.STEPLER_PREFIX.

Returns:

unique ids

Return type:

generator

stepler.third_party.utils.generate_files(prefix=None, postfix=None, folder=None, count=1, size=1024)

Generate files with unique names.

Parameters:

• prefix (str|None) – prefix of unique ids.
• postfix (str|None) – postfix of unique ids.
• folder (str|None) – folder to create unique files.
• count (int) – count of unique ids.
• size (int) – size of unique files.

Returns:

files with unique names.

Return type:

generator

stepler.third_party.utils.generate_file_context(prefix=None, postfix=None, folder=None, size=1024)

Context manager to generate file with unique name and delete it later.

Useful for large files.

Parameters:

• prefix (str|None) – prefix of unique id
• postfix (str|None) – postfix of unique id
• folder (str|None) – folder to create unique file
• size (int) – size of unique file (in bytes)

Yields:

str – file path.

stepler.third_party.utils.generate_ips(ip_start=1, ip_end=254, count=1)

Generate random IP v4 addresses.

Examples

173.217.169.131, 207.105.178.224, 193.121.141.217

Parameters:

• ip_start (int) – Start range of the generated sequence
• ip_end (int) – End range of the generated sequence
• count (int) – count of unique ids.

Returns:

Random IP addresses

Return type:

generator

stepler.third_party.utils.get_file_path(url, name=None)

Download file by URL to local cached storage.

Parameters:

• url (str) – URL of file location.
• name (str|None) – file name.

Returns:

file path of downloaded file.

Return type:

str

stepler.third_party.utils.get_size(value, to)

Get size of value with specified type.

stepler.third_party.utils.get_unwrapped_func(func)

Get original function under decorator.

Decorator hides original function inside itself. But in some cases it’s important to get access to original function, for ex: for documentation.

Parameters:func (function) – function that can be potentially a decorator which hides original function Returns:unwrapped function or the same function Return type:function

stepler.third_party.utils.slugify(string)

Replace non-alphanumeric symbols to underscore in string.

Parameters:string (str) – string to replace Returns:replace string Return type:str

stepler.third_party.utils.background(f, *args, **kwargs)

Run function in separate thread.

Parameters:

• f (function) – function to call in background
• *args – function args
• **kwargs – function kwargs

Returns:

started process instance

Return type:

Process

stepler.third_party.utils.join_process(process)

Wait until process to be done.

Parameters:process (Process) – process

stepler.third_party.utils.check_ssh_connection_establishment(server_ssh, must_work=True, timeout=0)

Function to check that ssh connection can be established.

Parameters:

• server_ssh (ssh.SshClient) – ssh connection
• must_work (bool, optional) – flag whether ‘server_ssh’ should be able to connect or not
• timeout (int, optional) – seconds to wait a result of check

Raises:

• RuntimeError – if server_ssh is not closed
• TimeoutExpired – if check failed after timeout

Video capture of display

class stepler.third_party.video_recorder.VideoRecorder(file_path, frame_rate=30)

Video capture of display.

clear()

Remove video file.

start()

Start video capture.

stop()

Stop video capture.

Waiter

exception stepler.third_party.waiter.ExpectationError(base_ex)

Expectation error class.

exception stepler.third_party.waiter.PredicateTimeout

Predicate timeout exception class.

exception stepler.third_party.waiter.TimeoutExpired(base_ex)

Timeout expired exception.

stepler.third_party.waiter.wait(predicate, args=None, kwargs=None, expected_exceptions=(), predicate_timeout=None, **wait_kwargs)

Wait that predicate execution returns non-false result.

It catches all raised ExpectationError and uses last exception to construct TimeoutException. It also can pass arguments to predicate.

Example

>>> def predicate(foo, bar='baz'):
...    expect_that(foo, equal_to(bar))
...    return bar
>>> wait(predicate, args=(1,), kwargs={'bar': 2}, timeout_seconds=1)

TimeoutExpired: Timeout of 1 seconds expired waiting for <function predicate at 0x7f7798622c08>

Expected: <2>

but: was <1>

>>> wait(lambda: False, timeout_seconds=0.5)

TimeoutExpired: Timeout of 0.5 seconds expired waiting for <function <lambda> at 0x7f2b54360848> No exception raised during predicate executing

Parameters:

• predicate (function) – predicate to wait execution result
• timeout_seconds (int) – seconds to wait result
• sleep_seconds (float) – polling time between predicate executions
• multiplier (int) – coefficient to multiply polling time
• expected_exceptions (tuple) – predicate exceptions which will be omitted during waiting.
• predicate_timeout (int) – max predicate execution timeout. Equals to timeout_seconds by default.
• waiting_for (str) – custom waiting message.

Returns:

result of predicate execution in format:

(predicate result, “Error message” or None)

Return type:

tuple

Raises:

TimeoutExpired – if predicate execution has false value after timeout