taf.plugins

Submodules

taf.plugins.loggers

loggers.py

logging functionality for TAF

class taf.plugins.loggers.ClassLogger(log_level='INFO', log_file=None, log_stream=True, for_exception=False, introspection=True)

Bases: object

Class logger descriptor.

__get__(instance, owner)

This method is called from class.

Parameters:owner (owner) – class instance. Returns:logger adaptor. Return type:logging.LoggerAdapter

Note

In case using logger for module level use get() method. __get__() won’t be called from module level.

__init__(log_level='INFO', log_file=None, log_stream=True, for_exception=False, introspection=True)

Initialize instance of ClassLogger.

Parameters:

• log_level (str) – Log level value.
• log_file (str) – Path to log file.
• log_stream (bool) – Log stream value.
• for_exception (bool) – True for exception information.
• introspection (bool) – True for extended information.

_get_logger(modulename, classname='', caller_func='')

Configure and return loggerAdapter instance.

Parameters:

• modulename (str) – module name.
• classname (str) – class name.
• caller_func (str) – function name.

Returns:

logger adaptor.

Return type:

logging.LoggerAdapter

get(modulename, classname='')

Return logerAdapter instance for module level logging.

Parameters:

• modulename (str) – module name.
• classname (str) – class name.

Returns:

logger adaptor.

Return type:

logging.LoggerAdapter

class taf.plugins.loggers.LoggerWrapper(logger, level)

Bases: threading.Thread

Read text message from a pipe and redirect them to a logger.

Note

The object itself is able to supply a file descriptor to be used for writing. fdWrite ==> fdRead ==> pipeReader.

__init__(logger, level)

Setup the object with a logger and a loglevel and start the thread.

_write(message)

Utility method to send the message to the logger with the correct loglevel.

fileno()

Return the write file descriptor of the pipe.

run()

This is the method executed by the thread, it simply read from the pipe (using a file-like wrapper) and write the text to log.

Note

NB the trailing newline character of the string read from the pipe is removed.

class taf.plugins.loggers.NoErrArgumentParser(*args, **kwargs)

Bases: argparse.ArgumentParser

ArgumentParser class that handle only predefined for an instance options.

Note

The original ArgumentParser class raises an error if handle unknown option. But py.test have it’s own options and it’s own custom parser and if ArgumentParser find them it raises an error. Using this class allows not to define all possible options in each module that uses ArgumentParser.

__init__(*args, **kwargs)

Initialize NoErrArgumentParser class.

add_argument(*args, **kwargs)

Add arguments and save regexps of valid for the instance options in valid_args_cre_list.

parse_args(*args, **kwargs)

Filter out invalid options and parse only predefined ones for the instance.

taf.plugins.loggers.mkdir_p(path)

taf.plugins.loggers.module_logger(name='', clsname='')

Return LoggerAdapter for module level logging.

taf.plugins.loggers.parse_options()

Parse additional cli logging options.

taf.plugins.loggers.pipe_loggers(name, log_file)

Return LoggerWrapper for pipe logging.

taf.plugins.pytest_caselogger

pytest_caselogger.py

Plugin is getting logs for each test case

class taf.plugins.pytest_caselogger.CaseLoggerExecutor(env)

Bases: object

Base class for TAF caselogger functionality.

__init__(env)

Initialize CaseLoggerExecutor instance.

Parameters:env (testlib.common3.Environment) – Environment instance from config

case_setup()

Add message into device logs on test case setup.

case_teardown()

Add message into device logs on test case teardown. Copy test case logs to the log host.

suite_teardown()

Copy core logs to the log host on test suite teardown.

class taf.plugins.pytest_caselogger.CaseLoggerPlugin

Bases: object

Base class for caselogger plugin functionality.

caselogger(request, suitelogger, env)

Call caselogger on test case setup/teardown

Parameters:

• request (pytest.request) – pytest request instance
• suitelogger – pytest fixture
• env (testlib.common3.Environment) – ‘env’ pytest fixture from pytest_onsenv.py

pytest_runtest_makereport(item, call)

Add information about test case execution results.

suitelogger(request, env_main)

Call caselogger on test suite teardown.

Parameters:

• request (pytest.request) – pytest request instance
• env_main (testlib.common3.Environment) – ‘env_main’ pytest fixture from pytest_onsenv.py

Returns:

instance of CaseLoggerExecutor class

Return type:

CaseLoggerExecutor

taf.plugins.pytest_caselogger.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_caselogger.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_caselogger.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_heat_checker

pytest_heat_checker.py

Plugin is checking temperature

class taf.plugins.pytest_heat_checker.HeatChecker(env, item)

Bases: object

Base class for HeatChecker functionality.

__init__(env, item)

Initialize HeatChecker instance.

Parameters:

• env (testlib.common3.Environment) – Environment instance.
• item (pytest.Item) – Test case instance.

check_and_log_temperature()

Get temperature of device and log this info.

class taf.plugins.pytest_heat_checker.HeatCheckerPlugin

Bases: object

Log device’s temperature.

heatcheck(request, env)

Call heat checker on test case setup/teardown.

Parameters:

• request (pytest.request) – Pytest request instance.
• env (testlib.common3.Environment) – ‘env’ pytest fixture from pytest_onsenv.py.

taf.plugins.pytest_heat_checker.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_heat_checker.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_heat_checker.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_helpers

pytest_helpers.py

pytest specific helpers functions

taf.plugins.pytest_helpers.get_brief(item, tc_name)

Parse doc-string and return brief.

Parameters:

• item (pytest.Item) – pytest test case item
• tc_name (str) – test case name

Returns:

test case docstring’s brief

Return type:

str

taf.plugins.pytest_helpers.get_doc_from_item(item)

taf.plugins.pytest_helpers.get_failure_reason(data)

Return test case failure reason from report string.

Parameters:data (str) – test case report Returns:failure reason or None Return type:str

taf.plugins.pytest_helpers.get_html_xml_path(path, build_name)

Parse and replace $BUILD_NAME variable in the path.

Parameters:

• path (str) – path to html report
• build_name (str) – software build number

Returns:

modified path to html report

Return type:

str

taf.plugins.pytest_helpers.get_skipped_reason(data)

Return test case skip reason from report string.

Parameters:data (str) – test case report Returns:skip reason or None Return type:str

taf.plugins.pytest_helpers.get_steps(item, tc_name)

Parse and return test steps.

Parameters:

• item (pytest.Item) – pytest test case item
• tc_name (str) – test case name

Returns:

test case’s steps

Return type:

str

taf.plugins.pytest_helpers.get_suite_name(nodeid)

Return suitename from nodeid string.

Parameters:nodeid (str) – pytest.Item nodeid string Returns:test suite name Return type:str

taf.plugins.pytest_helpers.get_tcname(report)

Return TC name from pytest report object.

Parameters:report (pytest.report) – pytest report object Returns:test case name Return type:str

Note

This function allows to get proper name without parameters string from normal and parametrized TCs.

taf.plugins.pytest_helpers.get_test_keys(data)

Return case keys from report string.

Parameters:data (str) – test case results Returns:list of report keys Return type:list[str]

taf.plugins.pytest_loganalyzer

pytest_loganalyzer.py

Plugin is checking device’s log after each test case is executed

class taf.plugins.pytest_loganalyzer.LogAnalyzer(switch)

Bases: object

IGNORE_RES = [re.compile('^Runtime journal is using ')]

__init__(switch)

Initialize LogAnalyzer instance.

Parameters:switch (dict{SwitchGeneral}) – switch dictionary

_check_log_duplicates(log_output)

Check if duplicated records are present in device log.

Parameters:log_output (str) – output of device log

_check_log_errors(log_output)

Check if errors with specific priority are present in device log.

Parameters:log_output (str) – output of device log

class_logger = <logging.LoggerAdapter object>

command = "date '+%Y-%m-%d %T'"

error_priority = 3

ignore(entry)

Check if duplicate can be ignored.

Parameters:entry (dict) – journalctl json decoded dict Returns:if we should ignore this message when checking duplicates Return type:bool

runtest_setup_flag = {}

setup()

Initialize start time of test run.

start_time = {}

teardown()

Check for errors and duplicated records in device log (journalctl).

class taf.plugins.pytest_loganalyzer.LogAnalyzerPlugin

Bases: object

log_analyzer(request, env, log_analyzer_setup)

Validate device’s logs on test teardown.

Parameters:

• request (pytest.request) – pytest request
• log_analyzer_setup (LogAnalyzer) – LogAnalyzer instance

log_analyzer_setup(env_main)

Setup LogAnalyzer plugin after devices had been started.

Parameters:env_main (common3.Env) – pytest fixture Returns:LogAnalyzer instance Return type:LogAnalyzer

taf.plugins.pytest_loganalyzer.pairwise(iterable)

s -> (s0,s1), (s1,s2), (s2, s3), ...

taf.plugins.pytest_loganalyzer.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_loganalyzer.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_loganalyzer.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_multiple_run

pytest_multiple_run.py

Re-run test case n times

class taf.plugins.pytest_multiple_run.MultipleRun(config)

Bases: object

pytest_collection_modifyitems(items)

pytest_generate_tests(metafunc)

pytest_runtest_call(item)

taf.plugins.pytest_multiple_run.pytest_addoption(parser)

Plugin specific options.

taf.plugins.pytest_multiple_run.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_multiple_run.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_multipletg

pytest_multipletg.py

Work with several TGs like with one instance

Note

If setup JSON file contains several TG devices they will be handled as one instance.

Use @pytest.mark.multiple_tgs class level marker in order to disable this functionality for particular test suites.

class taf.plugins.pytest_multipletg.MultipleTGClass(env)

Bases: object

change_cross_section()

class_logger = <logging.LoggerAdapter object>

create_multiple_tg()

setup()

teardown()

class taf.plugins.pytest_multipletg.MultipleTGPlugin

Bases: object

one_tg(request, env_init)

taf.plugins.pytest_multipletg.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_multipletg.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_onsenv

pytest_onsenv.py

Creates env fixture for ons test cases

class taf.plugins.pytest_onsenv.Env(request, env)

Bases: object

create()

destroy()

Destroy testing environment.

class taf.plugins.pytest_onsenv.EnvTest(request, env)

Bases: object

Cleanup/Check testing environment.

setup()

Cleanup/Check testing environment on test case setup.

teardown()

Cleanup/Check testing environment on test case teardown.

class taf.plugins.pytest_onsenv.OnsEnvPlugin

Bases: object

env(request, env_main)

Clear devices from environment.

Parameters:request (pytest fixture) – pytest.request Returns:Environment instance Return type:testlib.common3.Environment

env_init(request)

Validate command line options.

Parameters:request (pytest.request) – pytest request Returns:Environment instance Return type:testlib.common3.Environment

env_main(request, env_init)

Start/stop devices from environment.

Parameters:request (pytest.request) – pytest request Returns:Environment instance Return type:testlib.common3.Environment

pytest_sessionstart(session)

taf.plugins.pytest_onsenv._check_pytest_version(version, max_version)

Check if version is less or equal to the max_version.

Parameters:

• version (str) – product version
• max_version (str) – max product version

Returns:

True/False

Return type:

bool

taf.plugins.pytest_onsenv.finish(self)

taf.plugins.pytest_onsenv.pytest_addoption(parser)

TAF specific options.

taf.plugins.pytest_onsenv.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_onsenv.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_onsenv.setup_scope()

Return setup_scope option value in global namespace.

taf.plugins.pytest_pidchecker

pytest_pidchecker.py

Plugin is checking processes for being restarted

class taf.plugins.pytest_pidchecker.PidChecker(env, flag, item)

Bases: object

Base class for TAF pidchecker functionality.

__init__(env, flag, item)

Create PidChecker instance.

Parameters:

• env (testlib.common3.Environment) – Environment instance from ‘env’ fixture.
• flag (bool) – Flag to know if plugin should skip process validation.
• item (pytest.Item) – Test case instance.

_check_processes_setup(switch, tc_name)

Check processes on setup.

Parameters:

• switch (testlib.switch_general.SwitchGeneral) – switch instance from test environment.
• tc_name (str) – test case’s name.

_check_processes_teardown(switch, tc_name)

Check process on teardown.

Parameters:

• switch (testlib.switch_general.SwitchGeneral) – switch instance from test environment.
• tc_name (str) – test case’s name.

setup()

Get ONS processes PIDs on test case setup.

teardown()

Get ONS processes PIDs on test case teardown.

class taf.plugins.pytest_pidchecker.PidCheckerPlugin

Bases: object

PidCheckerPlugin implementation.

pidcheck(request, env)

Call ONS processes verification on test case setup/teardown.

Parameters:

• request (pytest.request) – pytest request instance.
• env (testlib.common3.Environment) – ‘env’ pytest fixture from pytest_onsenv.py.

taf.plugins.pytest_pidchecker.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_pidchecker.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_pidchecker.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_random_collection

pytest_random_collection.py

Choose one test case from test suite randomly if random marker is set

class taf.plugins.pytest_random_collection.RandomChoice(seed)

Bases: object

Choose one test case from test suite randomly if random marker is set.

pytest_collection_modifyitems(session, config, items)

taf.plugins.pytest_random_collection.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_random_collection.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_random_collection.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_reportingserver

pytest_reportingserver.py

XML-RPC reporting server plugin

class taf.plugins.pytest_reportingserver.ReportingServer(opts)

Bases: object

Logging xmlrpc server class.

REPORTINGSRV_PATH = 'reporting_server.py'

REPORTINGSRV_TIMEOUT = 30

UNDEFINED_BUILD = 'Undefined'

__init__(opts)

Initialize ReportingServer class.

_get_build_info(item=None)

Get info about build.

Parameters:item (pytest.Item) – test case item Returns:str, “build”: str}: build info Return type:dict{“platform”

_send_post_queue(item=None, buildname=None, sanity=False)

Send info about test execution to the Reporting Server.

Parameters:

• item (pytest.Item) – test case item
• buildname (str) – buildname
• sanity (bool) – True if sanity test

_send_post_request(item)

Send post request to reporting server or add it to queue.

Parameters:item (pytest.Item) – test case item

_sessionstart(item)

Tell to XMLRPC Server that we are going to interact with it.

Parameters:item (pytest.Item) – test case item

buildname(env_prop=None)

Return buildname for current session.

Parameters:env_prop (dict) – environment information e.g. build, name, etc. Returns:buildname Return type:str

check_server(timeout=1)

Check if xmlrpc server is alive.

Parameters:timeout (int) – timeout for server verification

class_logger = <logging.LoggerAdapter object>

launch_server(port=None)

Launch xmlrpc server.

Parameters:port (int) – port to launch xmlrpc server

pytest_keyboard_interrupt(excinfo)

Handle KeyboardInterrupt.

Parameters:excinfo (py.code.ExceptionInfo) – exception info

pytest_runtest_call(item)

Create TC instance and send it to the Reporting Server.

Parameters:item (pytest.Item) – test case item

pytest_runtest_logreport(report)

Send update TC run status to the Reporting Server.

Parameters:report (pytest.BaseReport) – pytets report

pytest_runtest_setup(item)

Add info about test case start time.

Parameters:item (pytest.Item) – test case item

pytest_sessionfinish(session)

Tell to XMLRPC Server that we have finished interaction.

Parameters:session (pytest.Session) – test session

server_cmd(cmd, args, retry=3)

Send XML query to server with retry and exception handling.

Parameters:

• cmd (str) – command name
• args (list) – command arguments
• retry (int) – retry count

shutdown_server()

Send xmlrpc request to shutdown xmlrpc server.

class taf.plugins.pytest_reportingserver.ReportingServerConfigBase

Bases: object

Reporting Server configuration.

_additional_option()

Defining options for Reporting Server.

taf.plugins.pytest_reportingserver.imp_plugins(dest)

Import all py modules from <dest> subfolder.

taf.plugins.pytest_reportingserver.pytest_addoption(parser)

Plugin specific options.

taf.plugins.pytest_reportingserver.pytest_configure(config)

Registering plugin.

Raises:Exception – not able to connect to the reporting server.

taf.plugins.pytest_reportingserver.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_returns

pytest_returns.py

Collect return results from testcases instead of ‘PASSED’ message

class taf.plugins.pytest_returns.ReturnsPlugin

Bases: object

PYTEST_REPORT_STATUSES = {'failed': {'letter': 'F', 'outcome': 'failed', 'msg': 'FAILED'}, 'skipped': {'letter': 'S', 'outcome': 'skipped', 'msg': 'SKIPPED'}, 'passed': {'letter': 'P', 'outcome': 'passed', 'msg': 'PASSED'}}

REPORT_RESULT_GETTER = operator.itemgetter('outcome', 'letter', 'msg')

RESULT_FAILED = 'failed'

RESULT_PASSED = 'passed'

RESULT_SKIPPED = 'skipped'

classmethod class_init()

classmethod get_result(result)

pytest_pyfunc_call(pyfuncitem)

pytest_report_teststatus(report)

pytest_runtest_makereport(item, call)

taf.plugins.pytest_returns.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_returns.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_skip_filter

pytest_skip_filter.py

Verify skipif condition on collect stage and remove skipped tests from test run

class taf.plugins.pytest_skip_filter.SkipFilterPlugin

Bases: object

Verify skipif reason and remove skipped tests from test run.

pytest_collection_modifyitems(session, config, items)

Handle skipif condition and remove skipped tests from test run.

pytest_terminal_summary(terminalreporter)

Add info in summary about removed tests.

taf.plugins.pytest_skip_filter.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_skip_filter.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_smartrerun

pytest_smartrerun.py

Plugin is collecting failed test cases from specific Test Plan in Jira and run them again

class taf.plugins.pytest_smartrerun.SmartRerun(config)

Bases: object

Re-run failed test cases.

_set_switch_log_level(env, loglevel)

Change loglevel on switches.

Parameters:loglevel (str) – logging level

log_level(request, env_main)

pytest_collection_modifyitems(config, items)

pytest_sessionfinish(session, exitstatus)

pytest_terminal_summary(terminalreporter)

taf.plugins.pytest_smartrerun.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_smartrerun.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_smartrerun.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_start_from_case

pytest_start_from_case.py

Starting tests execution from the specified test case not from the beginning

Examples:

# complete test-name
$ py.test PATH_TO_TESTS --start_from_case test_my_func

# not complete test-name
$ py.test PATH_TO_TESTS --start_from_case test*func

# or
$ py.test PATH_TO_TESTS --start_from_case *func

# parameterized tests
$ py.test PATH_TO_TESTS --start_from_case test*[1]

# or
$ py.test PATH_TO_TESTS --start_from_case test*1

class taf.plugins.pytest_start_from_case.StartFromCase(config)

Bases: object

Execute test run starting from specified test item.

pytest_collection_modifyitems(session, config, items)

Leave only necessary items in collected list.

taf.plugins.pytest_start_from_case.pytest_addoption(parser)

Plugin specific options.

taf.plugins.pytest_start_from_case.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_start_from_case.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_sut_monitor

pytest_sut_monitor.py

Collect monitoring info based on Colletcd RRDs

Note

For correct functioning collectd should be properly configured:

1. Collectd server should be configured in the lab
2. Collectd client should be configured on device
3. On device should be configured hostname equal to the ‘name’ value in JSON

class taf.plugins.pytest_sut_monitor.SutMonitor(env)

Bases: object

Main functionality for collectd client/server manipulation.

__init__(env)

Initialize SutMonitor object instance.

Parameters:env (testlib.common3.Environment) – TAF environment instance

class_logger = <logging.LoggerAdapter object>

configure()

Configure monitoring tool.

configure_server()

Create SSH connection to Collectd server host.

copy_file(remote_file, local_file)

Copy file from Collect server host to TAF host.

Parameters:

• remote_file (str) – path to file on remote Collectd server host
• local_file (str) – path to file on local TAF host

create_dir(dir_name, clear=True)

Create folder on TAF host.

Parameters:

• dir_name (str) – folder path
• clear (bool) – flag to delete folder if exists

exec_command(command)

Execute shell command on Collectd server host.

Parameters:command (str) – command to execute Returns:command execution output Return type:str

is_not_empty(folder, start, end, gtype)

Check if RRD folder contains non-empty data.

Parameters:

• folder (str) – RRD folder to check for
• start (str) – time period start value
• end (int) – time period end value
• gtype (str) – RRD folder CF

Returns:

True if folder contains non-empty data

Return type:

bool

item_teardown()

Create RRD graphs on test teardown.

list_rrd_folders()

List device’s RRD folders on Collectd server host.

multicall(commands)

Execute a list of commands on Collectd server host.

Parameters:commands (list[str]) – list of commands to be executed Returns:list of commands results Return type:list[str]

start_on_nodes()

Start Collectd service on devices.

teardown()

Close SSH connection to the Collectd server host.

class taf.plugins.pytest_sut_monitor.SutMonitorPlugin

Bases: object

SutMonitorPlugin implementation.

monitor(request, env_main, monitor_init)

Start Collectd service on devices.

monitor_init(env_init)

Initialize SutMonitor on session start.

Parameters:env_init (testlib.common3.Environment) – ‘env_init’ pytest fixture from pytest_onsenv.py

pytest_runtest_makereport(item, call)

Add generated graphs to the pytest report in order to access from reporting plugin.

test_monitor(request, env, monitor)

Gather collectd info for certain test case.

Parameters:

• request (pytest.request) – pytest request object
• env (testlib.common3.Environment) – env fixture
• monitor_start (SutMonitor) – monitor_start fixture

taf.plugins.pytest_sut_monitor.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_sut_monitor.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_sut_monitor.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_syslog

pytest_syslog.py

Plugin writes messages to SysLogHandler

Note

For correct functioning syslog should be properly configured, e.g.

1. On device should be configured hostname seacliff9.

2. The following lines should be added to syslog config:

   # This is criteria for seacliff messages
   if $hostname contains 'seacliff9' then /var/log/switches/seacliff09.log
   # This s criteria for our log messages
   if $msg contains 'seacliff9' and $programname == 'pytest_syslog' then /var/log/switches/seacliff9.log
   

Also it is good to configure logrotate (e.g. /etc/logrotate.d/switches) file of :

Examples:

/var/log/switches/*.log
{
    rotate 730
    weekly
    missingok
    notifempty
    delaycompress
    compress
    postrotate
        reload rsyslog >/dev/null 2>&1 || true
    endscript
}

Known issues:

• rsyslog could miss log messages from remote host if it cannot resolve host FQDN 10 times. To disable dns resolving add -x option to /etc/default/rsyslog (for debian based distros).

class taf.plugins.pytest_syslog.SyslogNotifier

Bases: object

Send syslog messages.

_create_header(env_prop, stage)

Send syslog header/footer message for each device in config.

Parameters:

• env_prop (dict) – Environment properties for test run identification
• stage (str) – “SessionStart” or “SessionFinish”

Returns:

None

Examples:

self._create_header(env_object.config, {"py.test PID": "5216", "chipName": "8086", "cpuArchitecture": "8bit"}, "SessionStart")

_update_tc_status(tcname, status)

Send syslog TC status message for each device in config.

Parameters:

• tcname (str) – Name of TC
• status (str) – TC status - started or finished

Returns:

None

Examples:

self._update_tc_status(env_object.config, "test_some_feature_1", "started")

get_last_record_from_log(log, item)

Return last log records for TC.

Parameters:

• log (str) – log file
• item (pytest.Item) – test case item

Returns:

Log related to specified test item

Return type:

str

get_log_path(switch_name)

Return log path from environment.json for switch.

Parameters:switch_name (str) – switch name Returns:path to log for specified device Return type:str

is_test_completed(report)

Return True if make_report hook called after TC and TC failed.

Parameters:report (pytest.Report) – pytest report Returns:True if test is complited without errors Return type:bool

pytest_runtest_call(item)

Send syslog message on TC start.

pytest_runtest_makereport(item, call)

If TC failed add system log (rsyslog) for TC to report object.

pytest_runtest_teardown(item, nextitem)

Send syslog message on TC end.

pytest_sessionfinish(session)

Send syslog message with session footer.

pytest_sessionstart(session)

Send syslog message with session header.

setup_logger(session)

Setup logger for each device.

Parameters:session (pytest.Session) – pytest session

taf.plugins.pytest_syslog.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_syslog.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_syslog.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_tempest

pytest_tempest.py

Prepares sys path for tempest

taf.plugins.pytest_tempest.prepend_path(path)

taf.plugins.pytest_tempest.pytest_addoption(parser)

Tempest specific options.

taf.plugins.pytest_tempest.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_tempest.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_test_duration

pytest_test_duration.py

Plugin is controlling test execution by time or count of iteration

class taf.plugins.pytest_test_duration.Duration(option=None)

Bases: object

Main functionality for test duration manipulation.

__init__(option=None)

Initialize Duration object instance.

Parameters:option (str, e.g "30s", "2.5H") – time to interrupt test(cmd option)

_parse_and_define_delta_time(timing)

Parse time string.

Parameters:timing (str, e.g "30s", "2.5H") – time to interrupt test

control_duration(timing=None, count=None)

Control duration test by timing, option or count.

Parameters:

• timing (str, e.g "30s", "2.5H") – time to interrupt test
• count (int|float) – interrupt test after iterated count

class taf.plugins.pytest_test_duration.TestDurationPlugin

Bases: object

TestDurationPlugin implementation.

duration(request)

Initialize Duration fixture.

taf.plugins.pytest_test_duration.pytest_addoption(parser)

Plugin specific options.

taf.plugins.pytest_test_duration.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_test_duration.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.pytest_workload

pytest_workload.py

Add workload on SUTs using stress tool

class taf.plugins.pytest_workload.ARGS(class_name, config, opts, shared, workers)

Bases: tuple

__getnewargs__()

Return self as a plain tuple. Used by copy and pickle.

static __new__(_cls, class_name, config, opts, shared, workers)

Create new instance of ARGS(class_name, config, opts, shared, workers)

__repr__()

Return a nicely formatted representation string

_asdict()

Return a new OrderedDict which maps field names to their values.

classmethod _make(iterable, new=<built-in method __new__ of type object at 0xa395c0>, len=<built-in function len>)

Make a new ARGS object from a sequence or iterable

_replace(_self, **kwds)

Return a new ARGS object replacing specified fields with new values

class_name

Alias for field number 0

config

Alias for field number 1

opts

Alias for field number 2

shared

Alias for field number 3

workers

Alias for field number 4

class taf.plugins.pytest_workload.WorkloadContinuous(env, workers)

Bases: object

Main functionality for workload manipulation.

__init__(env, workers)

Initialize WorkloadContinuous object instance.

Parameters:env (testlib.common3.Environment) – TAF environment instance

class_logger = <logging.LoggerAdapter object>

item_setup()

Start the workload if no active.

item_teardown()

Stop the workload and get the results.

start_on_nodes()

Start workload on devices.

teardown()

Stop the workload.

class taf.plugins.pytest_workload.WorkloadInterrupted(env, workers)

Bases: object

Main functionality for workload manipulation.

__init__(env, workers)

Initialize WorkloadInterrupted object instance.

Parameters:env (testlib.common3.Environment) – TAF environment instance

class_logger = <logging.LoggerAdapter object>

item_setup()

Start the workload if no active.

item_teardown()

Stop the workload and get the results.

start_on_nodes()

Start workload on devices.

teardown()

Stop the workload.

class taf.plugins.pytest_workload.WorkloadPlugin(workload_type, workers)

Bases: object

WorkloadPlugin implementation.

test_workload(request, env, workload)

Gather collectd info for certain test case.

Parameters:

• request (pytest.request) – pytest request object
• env (testlib.common3.Environment) – env fixture
• monitor_start (SutMonitor) – monitor_start fixture

workload(request, env_main, workload_init)

Start stress tool on devices.

workload_init(env_init)

Initialize WorkloadPlugin on session start.

Parameters:env_init (testlib.common3.Environment) – ‘env_init’ pytest fixture from pytest_onsenv.py

taf.plugins.pytest_workload.device_workload(args)

Start workload on device.

taf.plugins.pytest_workload.get_workers(line)

taf.plugins.pytest_workload.pytest_addoption(parser)

Describe plugin specified options.

taf.plugins.pytest_workload.pytest_configure(config)

Registering plugin.

taf.plugins.pytest_workload.pytest_unconfigure(config)

Unregistering plugin.

taf.plugins.reports_conf

taf.plugins.reports_conf

taf.plugins.reports_conf.SYNAPSERT

SYNAPSERT.py

SYNAPSERT class

class taf.plugins.reports_conf.SYNAPSERT.ReportingServerConfig

Bases: object

Reporting Server configuration.

static _additional_option(parser)

Plugin specific options.

static _configure(config)

Checking tm_update option.

Parameters:config (bool) – get options

static _get_build_name(options)

Return spesified buildname.

Parameters:options (str) – get options

static _sessionstart(log_class, item, name, buildname)

Tell to XMLRPC Server that we are going to interact with it.

Parameters:

• item (pytest.Item) – test case item
• name (str) – name for current session
• buildname (str) – buildname for current session

class_logger = <logging.LoggerAdapter object>

taf.plugins.reports_conf.XML

XML.py

XML and HTML report classes

class taf.plugins.reports_conf.XML.ReportingServerConfig

Bases: object

Reporting Server configuration.

static _additional_option(parser)

Plugin specific options.

static _configure(config)

Checking XML/HTML options.

static _get_build_name(options)

Return specified buildname.

static _sessionstart(log_class, item, name, buildname)

Tell to XMLRPC Server that we are going to interact with it.

Parameters:

• item (pytest.Item) – test case item
• name (str) – name for current session
• buildname (str) – buildname for current session

class_logger = <logging.LoggerAdapter object>

taf.plugins.reports_conf.XML.str2dict(dictstr)

Convert string to dictionary.