reporting

Submodules

reporting.connectors.SYNAPSERT

SYNAPSERT.py

SYNAPSERT class

class reporting.connectors.SYNAPSERT.SYNAPSERT(config_file='synapsert_client.json')

Bases: object

Class for TCMS interaction.

__init__(config_file='synapsert_client.json')

Class initialization.

Parameters:config_file (str) – SYNAPSERT configuration file

_check_prefix(issue, prefix, platform)

Check if summary of Test Case containts prefix. Return True or prefix is None or prefix exists.

Parameters:

• issue (JIRA.Issue) – Jira issue
• prefix (str) – Prefix of Test Plan
• platform (str) – Platform of failed issue

Returns:

True if platform or prefix in issue

Return type:

bool

_create_test_suite(ts_name)

Create test suite.

Parameters:ts_name (str) – Jira Test Suite name

_do_transition(item, value)

Do transition of JIRA issue.

Parameters:

• item (JIRA.Issue) – Instance of issue
• value (str) – Name of button

Returns:

True in case has been pushed successfully

Return type:

bool

Note

Push buttons on issue: Pass, Fail, Cant Test, etc…

_generate_servlet(key, issue_type=None)

Generate OptionServlet issue or suite to OptionField.

Parameters:

• key (str) – Jira Issue key
• issue_type (str) – Type of Jira issues

_get_all_issues(jql)

Get all issue because one request can get only 1,000-2,000 issues.

Parameters:jql (str) – jql request Returns:List of Jira issues Return type:list[JIRA.Issue]

_get_all_statuses()

Get all available statuses.

Returns:Get all available Jira statuses Return type:list[str]

_get_linked_defects(value, auto_tc_name, platform, prefix)

Get linked defects.

Parameters:

• value (str) – Value of failure reason
• auto_tc_name (str) – Automated TC Name customfield.
• platform (str) – Platform of failed issue
• prefix (str) – Prefix of Test Plan

Returns:

list of linked defetcs

Return type:

list[JIRA.Issue]

Note

Get list of linked defects of SubTests by auto_tc_name, platform and/or prefix.

static _get_suite_value(suite)

Get suite value without _test_.

Parameters:suite (str) – Full test suite name Returns:Suite name without _test_ Return type:str

_init_jira_client()

Return Jira client.

Returns:jira-instance Return type:JIRA

_link_issues(test_case, defect_keys)

Relink defects to subtest.

Parameters:

• test_case (JIRA.Issue) – Instance of Test Case
• defect_keys (list[str]) – List of defect keys

Returns:

True if defect list updated successfully

Return type:

bool

_set_allowed_suites(test_case)

Set allowed suites.

Parameters:test_case (JIRA.Issue) – Instance of Test Case

Note

Get allowed values from ‘Test Suite’ custom field and store them.

_set_default_custom_fields()

Find and set custom fields.

Note

Serve for operating with custom fields.

_set_suite(test_case, suite)

Set Test Suite.

Parameters:

• test_case (JIRA.Issue) – Jira test case
• suite (str) – Test suite name

_update_cf(test_case, custom_field=None, data=None)

Update Custom Field.

Parameters:

• test_case (JIRA.Issue) – Jira test case
• custom_field (str) – Name of custon field
• data (str) – Custom field value

Note

Get customfield_id from allowed values.

class_logger = <logging.LoggerAdapter object>

create_subtest(test_plan, test_case, client=None, build_info=None)

Create subtest.

Parameters:

• test_plan (JIRA.Issue) – Jira test plan
• test_case (JIRA.Issue) – Jira test case
• client (dict) – Client info
• build_info (dict) – Build info

Returns:

Jira SubTest instance

Return type:

JIRA.Issue

create_tc(name, description, auto_tc_name, suite=None)

Create test case.

Parameters:

• name (str) – Summary of Test Case
• description (str) – Description of Test Case
• auto_tc_name (str) – Automated TC Name customfield.
• suite (str) – Test Suite name.

Returns:

instance of Test Case

Return type:

JIRA.Issue

create_tp(tp_name)

Create test plan.

Parameters:tp_name (str) – Summary of Test Plan Returns:Instance of Jira Test Plan Return type:JIRA.Issue

get_available_platforms()

Get platform field values.

Returns:List with available platforms Return type:list

get_cf_value(issue, name)

Get custom field value.

Parameters:

• issue (JIRA.Issue) – Jira Issue
• name (str) – Name of custom field

Returns:

Custom field value

Return type:

str

get_config()

Return config fields.

Returns:host, project name, build number, test case state, fail reason, automated test case name, platform Return type:tuple(str)

get_custom_fields(issue_type)

Get custom fields.

Parameters:issue_type (str) – Jira issue type Returns:Dictionary with custom field name, value pairs Return type:dict{}

get_defects_list(tc_key)

Get defects of Issue.

Parameters:tc_key (str) – Jira issue key Returns:List of linked defects Return type:list[str]

get_issue_by_name(name, issue_type='Test Case')

Get Jira issue by name.

Parameters:

• name (str) – Summary of Test Plan
• issue_type (str) – Jira issue type

Returns:

JIRA issue

Return type:

JIRA.Issue

get_issue_key(test_case)

Get Issue key.

Parameters:test_case (JIRA.Issue) – Jira issue Returns:Jira issue key Return type:str

get_issue_link(tc_key)

Get issue links of Test Case.

Parameters:tc_key (str) – Jira issue key Returns:Link to Jira issue Return type:str

get_issue_status(issue)

Get issue status

Parameters:issue (JIRA.Issue) – Instance of Test Case Returns:Test case status Return type:str

get_issue_type_id(issue_type)

Get Jira issue type ID.

Parameters:issue_type (str) – Jira issue type Returns:Jira issue type ID Return type:int

get_last_failed_issues_from_tps(tp_names, linked_defects)

Get last failed issues by test plan name.

Parameters:

• tp_names (str) – Jira Test Plan name
• linked_defects (bool) – Look for linked defects

Raises:

Exception – Test Plan does not exist

Returns:

List of failed test cases

Return type:

list[str]

get_last_open_tp(tp_name)

Return last open Test Plan.

Parameters:tp_name (str) – Jira Test Plan name Returns:Last open Test Plan in JIRA Return type:JIRA.Issue

get_last_tp_by_platform(platform)

Get latest Test Plan by platform.

Parameters:platform (str) – Platform name Returns:Latest Test Plan by platform Return type:JIRA.Issue

get_previous_subtask(auto_tc_name, current_key, platform)

Get previous subtask.

Parameters:

• auto_tc_name (str) – Automated Test Case name
• current_key (str) – Jira issue key
• platform (str) – Platform of Jira issue

Returns:

Previous Jira issue

Return type:

JIRA.Issue

get_st_history(auto_tc_name, tp_names)

Get history of SubTests.

Parameters:

• auto_tc_name (str) – Jira issue key
• tp_names (str) – Jira Test Plans

Returns:

SubTest history

Return type:

dict

static get_summary(issue)

Get summary of Jira issue.

Parameters:issue (JIRA.Issue) – Instance of Test Case Returns:Summary of Jira issue Return type:str

get_tc_by_auto_tc_name(auto_tc_name)

Get Test Case by auto_tc_name.

Parameters:auto_tc_name (str) – Automated TC Name of Test Case Returns:Instance of Test Case in JIRA Return type:JIRA.Issue

get_tc_by_key(tc_key)

Get test case by key.

Parameters:tc_key (str) – Jira issue key Returns:Jira Test Case issue Return type:JIRA.Issue

get_tracker()

Check if tracker exists and sleep on 30 seconds.

Returns:True if connect to Jira host Return type:bool

get_transitions(issue)

Get available transitions of issue.

Parameters:issue (JIRA.Issue) – Jira issue Returns:List of available transitions Return type:list[str]

jira = None

name = 'SynapseRT'

set_failure_reason(test_case, value, platform=None, prefix=None)

Set Failure Reason.

Parameters:

• test_case (JIRA.Issue) – Instance of Test Case
• value (str) – Value of failure reason
• platform (str) – Platform of failed issue
• prefix (str) – Prefix of Test Plan

update_tc_status(test_case, status)

Update status of test case.

Parameters:

• test_case (JIRA.Issue) – Instance of Test Case
• status (str) – Test case status

Raises:

ValueError – unknown status

update_test_case(test_case, info, suite)

Update summary, description, suite of Test Case.

Parameters:

• test_case (JIRA.Issue) – Jira issue
• info (dict) – Jira issue info
• suite (str) – Test suite name

reporting.loggers

loggers.py

logging functionality for TAF

class reporting.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.

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 ClassLogger class.

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

Configure and return loggerAdapter instance.

get(modulename, classname='')

Return logerAdapter instance for module level loging.

class reporting.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 reporting.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.

reporting.loggers.mkdir_p(path)

Create a directory.

reporting.loggers.module_logger(name='', clsname='')

Return LoggerAdapter for module level logging.

reporting.loggers.parse_options()

Parse additional cli logging options.

reporting.loggers.pipe_loggers(name, log_file)

Return LoggerWrapper for pipe logging.

reporting.reporting_server

reporting_server.py

Standalone loggingserver for pytest plugin

class reporting.reporting_server.ClientCollector

Bases: object

Thread safe object for collecting clients dictionaries.

STATUSES = ['Active', 'Inactive']

__init__()

Initialize ClientCollector class.

active()

List of active clients.

Returns:list of clients. Return type:list[dict]

Note

There are can be commands from client in queue when client is disconnected (see inprocess() method).

addreport(client, report_type)

Add and open report type.

Parameters:

• client (tuple(str)) – Tuple of ClientID and build.
• report_type (str) – “html”, “xml” or “wrtm”.

Raises:

KeyError – client not in self.clients

Returns:

None

Examples:

client = ("py.test-user-1234", "v.0.0.0.0.0.1")
report_type = "html"
client_collector.addreport(client, report_type)

all()

All connected/disconnected client.

Returns:@return: list of clients. Return type:list[dict]

class_logger = <logging.LoggerAdapter object>

closereport(client, report_type)

Close report type (set False for existing opened report).

Parameters:

• client (tuple(str)) – Tuple of ClientID and build
• report_type (str) – “html”, “xml” or “wrtm”.

Raises:

KeyError – client not in self.clients

Returns:

None

delreport(client, report_type)

Remove report type (disabling collecting info for report).

Parameters:

• client (tuple(str)) – Tuple of ClientID and build.
• report_type (str) – “html”, “xml” or “wrtm”.

Raises:

KeyError – client not in self.clients

Returns:

None

get(client, attr)

Return client’s attribute.

Parameters:

• client (tuple(str)) – Tuple of ClientID and build.
• attr (str) – Client attribute - key of client dict.

Returns:

Client’s attribute (return type depends on attribute type).

Examples:

client = ("py.test-user-1234", "v.0.0.0.0.0.1")
client_collector.get(client, "connect_time")
client_collector.get(client, "reports")

getall(client)

Return all client attributes.

Parameters:client (str) – Tuple of ClientID and build. Returns:client dict. Return type:dict

inprocess()

List of clients with unprocessed close command.

Returns:list of clients. Return type:list[dict]

Note

On close command server should close (and dump) client’s reports.

update(client, status)

Update status of the client.

Parameters:

• client (tuple(str)) – Tuple of ClientID and build.
• status (str) – Available values are listed in STATUSES constant.

Raises:

ValueError – status not in self.STATUSES

Returns:

None

Examples:

client_id = "py.test-user-1234"
build = "0.0.0.0.pre-alfa"
client_collector.update((client_id, build), ClientCollector.STATUSES[0])

class reporting.reporting_server.CommandCollector

Bases: object

Thread safe collector for server command queue.

__init__()

Initialize CommandCollector class.

add(cmd)

Add command to collector.

Parameters:cmd (dict) – Command Returns:Status Return type:bool

Examples:

command = {'client': "client-name-with-pid-1111", 'build': 'cool-software-v.0.0.0.1-pre-alfa', 'close': True, 'duration': 15}
command_collector.add(command)

class_logger = <logging.LoggerAdapter object>

drop(index)

Remove item by index.

Parameters:index (int) – Index of item in queue list. Returns:Command or Error message. Return type:dict|str

get()

Same as pop but does not remove item.

Returns:command Return type:dict

Note

See pop() method description.

len()

Return command queue length.

Returns:command queue length. Return type:int

list()

Return list of all items.

Returns:All commands list. Return type:list[dict]

pop()

Pop the first item.

Returns:command Return type:dict

Note

See also get() method.

class reporting.reporting_server.XMLReportingServer(allowNone=False, useDateTime=False)

Bases: twisted.web.xmlrpc.XMLRPC

Root reporting server handler.

Note

This handler receive all test execution information and with command_processor process it to reports.

NAME = 'XMLReportingServer'

check_report_instance(r_type, client_id)

Create report instance if any for given client_id and build.

Parameters:

• r_type (str) – report type.
• client_id (str) – Unique client name.

Returns:

None

class_logger = <logging.LoggerAdapter object>

clients = None

command_processor(cmd)

Processing command from queue.

Parameters:cmd (dict) – Command. Returns:None

init_connectors()

Initializing connectors.

is_idle()

Command processor idle status.

Returns:True if command queue is empty and all clients are in closed state Return type:bool

keep_alive_interval = 3600

keep_alive_last = 1561419714.326169

last_cmdprocess_status = None

last_cmdprocess_time = None

last_operation_time = None

queue = None

queue_watchdog()

Send commands from queue to command processor in loop.

Raises:Exception – exception while processing command.

setup(opts)

Create WRTM proxy instance and set necessary attributes.

Parameters:opts (OptionParse) – cli options (OptionParse parsed options). Returns:None

shutdown()

Shutdown xmlrpc server.

start_queue_watchdog()

Start watchdog in thread.

watchdog = None

watchdog_thr = None

xmlrpc_checklocks()

Get CommandCollector and ClientCollector lock statuses.

Returns:CommandCollector and ClientCollector lock statuses. Return type:dict{bool}

xmlrpc_clientfullinfo(client)

Return full client info.

Parameters:client (tuple(str)) – Tuple of ClientID and build. Returns:Full client dictionary with all attributes. Return type:dict

xmlrpc_clientlist(ltype='active')

Get list of active clients.

Parameters:ltype (str) – ‘all’|’active’ Returns:list of active clients Return type:list

xmlrpc_close(**kwargs)

Function wrapper.

xmlrpc_cmdproccheck()

Return queue command processing status.

Returns:Query execution status. Return type:str

xmlrpc_cmdprocdisable()

Disabling and stopping queue command processing.

Returns:Query execution status. Return type:bool

xmlrpc_cmdprocenable()

Enabling queue command processing.

Returns:Query execution status. Return type:bool

xmlrpc_cmdprocstart()

Starting queue command processing.

Returns:Query execution status. Return type:bool

xmlrpc_idletime()

Return idle time.

Returns:time elapsed from last processed xmlrpc query (int). Return type:int

xmlrpc_open(**kwargs)

Function wrapper.

xmlrpc_ping()

Return self.NAME.

xmlrpc_post(**kwargs)

Function wrapper.

xmlrpc_queuedropcmd(index)

Remove command from queue by index.

Parameters:index (int) – Index of command in queue. Returns:Command. Return type:dict

Note

It’s recommended to stop queue processing first (see cmdprocdisable method) and check queuelist.

xmlrpc_queueidletime()

Return queue idle time.

Returns:time elapsed from last processed command (int). Return type:int

xmlrpc_queuelastcmdstatus()

Get last queue command process status.

Returns:last queue command process status Return type:bool

xmlrpc_queuelen()

Get command queue length.

Returns:command queue length Return type:int

xmlrpc_queuelist()

Get command queue list.

Returns:command queue list Return type:list[dict]

xmlrpc_reportadd(**kwargs)

Function wrapper.

xmlrpc_reportconfig(**kwargs)

Function wrapper.

xmlrpc_reportdel(**kwargs)

Function wrapper.

xmlrpc_reportmethod(report_type, client, method, params=None)

Call report class method.

Parameters:

• report_type (str) – Report type.
• client (tuple(str)) – Tuple of ClientID and build.
• method (str) – method name.
• params (list) – method params.

Returns:

Query execution status if one.

xmlrpc_reportslist(r_type=None)

List of all report instances.

Parameters:r_type (str) – Report type to list. Returns:dict of dict {<type>: {<build>: {<client>: <value>}}} Values are dicts of report specific values. (dict of dict) Return type:dict

xmlrpc_shutdown(trycount=0, lasttry=0)

Store shutdown server command.

Parameters:

• trycount (int) – attempts count.
• lasttry (int) – time of last attempt.

reporting.reporting_server.get_local_port()

Return free local port.

reporting.reporting_server.imp_plugins(dest)

Import all py modules from <dest> subfolder.

reporting.reporting_server.log_exception(msg, exc_type=(<class 'Exception'>, ))

reporting.reporting_server.main(ppid)

Start standalone server.

reporting.reporting_server.parse_options()

Parsing options.

reporting.reporting_server.update_timestamp(function)

Decorator: update last operation timestamp.

reporting.reports._reporter

reporting.reports.SYNAPSERT

reporting.reports.XML