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

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

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

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', 'msg': 'FAILED', 'outcome': 'failed'}, 'passed': {'letter': 'P', 'msg': 'PASSED', 'outcome': 'passed'}, 'skipped': {'letter': 'S', 'msg': 'SKIPPED', 'outcome': 'skipped'}}

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

taf.plugins.pytest_smartrerun

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

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

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.