pyats.easypy package

pyats.easypy.Task(*args, **kwargs)
pyats.easypy.main()

command line entry point

command-line entry point. Uses the default runtime and checks for whether a jobfile is parsed from command line, if not, exist with parser error.

strictly used for setuptools.load_entry_point/console_script.

pyats.easypy.run(*args, max_runtime=None, **kwargs)

run api

Shortcut function to start a Task(), wait for it to finish, and return the result to the caller. This api avoids the overhead of having to deal with the task objects, and provides a black-box method of creating tasks sequentially.

Parameters:

  • (int) (max_runtime) – finish, before terminating it forcifully and raising errors. If not provided, waits forever.

  • (tuple) (args) –

  • (dict) (kwargs) –

Return type:

the task’s result code

Subpackages

Submodules

class pyats.easypy.base.Component(runtime)

Bases: object

classmethod configure_parser(parser, legacy_cli=True)

configures the main runtime parser with this component’s own parser group.

Note

  • to be implemented by the component owner/developer.

  • should be adding argument groups and args to the provided parser

property runtime
class pyats.easypy.commands.RunEasypy(prog)

Bases: Command

Run Easypy Job file

description = '\nRuns a pyATS job file with the provided arguments, generating & report result.\n'
help = 'runs the provided pyATS job file'
main(argv)

internal entrypoint for commands

name = 'job'
usage = '{prog} [file] [options]\n\nExample\n-------\n  {prog} /path/to/jobfile.py\n  {prog} /path/to/jobfile.py --testbed-file /path/to/testbed.yaml\n'
class pyats.easypy.email.AbstractEmailReport

Bases: object

AbstractEmailReport class

Base class for all report emails to inherit from and follow.

create_email(from_addrs, to_addrs, mailhtml, default_domain=None)
Parameters:

(bool) (mailhtml) –

returns EmailMessage class instance with subject and body filled. This class can be overwritten to allow for MIME rich contents

abstract format_content()

abstract method, returns the content of email report in str

abstract format_subject()

abstract method, returns the subject of email report in str

save(filename)

saves the report content to file. can be overwritten to do more.

class pyats.easypy.email.ExceptionEmailReport(runtime, subject='pyATS Report - job: {runtime.job.name} by: {runtime.env.user}, !!! an exception occurred !!!', contents=None, attachment=None)

Bases: AbstractEmailReport

ExceptionEmailReport class

Used by MailBot to convert exceptions and tracebacks into an informative email message. This class is only used when the test-infrastructure crashes.

format_content()

abstract method, returns the content of email report in str

format_subject()

abstract method, returns the subject of email report in str

get_attachment()
class pyats.easypy.email.MailBot(*args, **kwargs)

Bases: Component

Mail Bot Class

Allows auto-generation of report emails when Easypy is finished running, pulling the reports from runtime.mail_report. This class is a context manager: when codes are executed within it, any exceptions caught by __exit__ will be added to the email report and sent out.

mailbot constructor

Parameters:

  • (list/str) (from_addrs) – in the generated email’s “From:” field.

  • to_addrs(list/str) (list or string-list of addresses to be used) – in the generated email’s “To:” field.

  • (str) (subject) –

  • (bool) (mailhtml) –

  • (bool) –

classmethod configure_parser(parser, legacy_cli=True)

configures the main runtime parser with this component’s own parser group.

Note

  • to be implemented by the component owner/developer.

  • should be adding argument groups and args to the provided parser

property mailto

mailto property getter

Returns the to_addrs Mailbot object.

class pyats.easypy.email.TextEmailReport(runtime, subject='pyATS Report - job: {runtime.job.name} by: {runtime.env.user}, total: {runtime.job.results[total]} (P:{runtime.job.results[passed]}, PX:{runtime.job.results[passx]}, F:{runtime.job.results[failed]} ...)', contents=None, attachment=None)

Bases: AbstractEmailReport

TextEmailReport class

Used as the standard email report generator. The reporting content is based off basic string-formating using templates and “runtime” as the input.

contains a “contents” OrderableDict, allowing users to add, remove and re-order report content. Each key in this dict formats into a section’s banner title, and each value is sent through the formatter to generate section body.

format_content()

abstract method, returns the content of email report in str

format_subject()

abstract method, returns the subject of email report in str

get_attachment()
class pyats.easypy.job.Job(*args, **kwargs)

Bases: Component

Job class

Representing a currently running jobfile within easypy. Each easypy run is only associated with a single Job class instance, and is accessible as runtime.job.

Job objects performs the actual job environment setup, run, finalization and cleanup. Refer to corresponding method for details.

Job constructor

Instantiates a job object and sets all sorts of internal states.

Parameters:

  • (str) (jobfile) –

  • (EasypyRuntime) (runtime) –

classmethod configure_parser(parser, legacy_cli=True)

configures the main runtime parser with this component’s own parser group.

Note

  • to be implemented by the component owner/developer.

  • should be adding argument groups and args to the provided parser

details()
static generate_uid(jobfile)
property report
run()

Job run

Runs the provided jobfile. Steps:

  1. runs all pre-job plugins

  2. run main() function entry of provided jobfile

  3. cleanup any dangling tasks

  4. add any child task messages to report email

  5. run post-job plugins

update_job()
class pyats.easypy.main.EasypyRuntime(parser=None, legacy_cli=True)

Bases: object

Built-in __init__

Initializes Easypy object with default values required for the parser.

CLI_DESCRIPTION = 'Easypy command line arguments.\nNon-recognized args will pass-through to the user job file.\n\nExample\n-------\n  %(prog)s /path/to/jobfile.py\n  %(prog)s /path/to/jobfile.py -testbed_file /path/to/testbed.yaml\n\n--------------------------------------------------------------------------------\n'
class ReturnCodes(value)

Bases: IntEnum

Standard return codes for Easypy Runtime

ARGPARSE_ERROR = 2
NOT_100_PERCENT = 1
OK = 0
PLUGIN_ERROR = 3
UNKNOWN_ERROR = 4
property archive

archive

shortcut redirect to job.runinfo for backwards compatbility

Type:

property

cleanup()

Cleanup runtime environment

Stage to cleanup the whole runtime environment. Best-attempt: should not be raising exceptions. Since this stage is performed after archive is created, all messages are logged to screen only.

  1. kill all children processes (if any)

  2. detach all logging handlers (for graceful exit)

  3. delete the runinfo.runinfo_dir

configure_logging()
configure_parser()

configure main program parser

details()

Return the current TestSuite information from the Reporter.

property directory

directory

shortcut redirect to job.runinfo for backwards compatbility

Type:

property

generate_email_reports()
main(argv=None, configuration=None)

run

Business logic, runs everything.

Parameters:

  • argv (command line arguments to parse, default to sys.argv) –

  • configuration (additional configuration pertaining to this run) –

parse_args(*args, **kwargs)

parse and return cli arguments

print_help()
property results_summary
property testbed

testbed

shortcut redirect to job.testbed for backwards compatbility

Type:

property

pyats.easypy.main.build_report_tree(section, parent_node, level=0)

Builds a tree for displaying items mimicking linux tree command.

pyats.easypy.main.main()

command line entry point

command-line entry point. Uses the default runtime and checks for whether a jobfile is parsed from command line, if not, exist with parser error.

strictly used for setuptools.load_entry_point/console_script.

class pyats.easypy.tasks.Task(testscript, taskid=None, test_harness='pyats.aetest', runtime=None, **kwargs)

Bases: ForkProcess

Task class

Task class represents the actual task/testscript being executed through a child process. All tasks within easypy are encapsulated in its own forked process from easypy main program.

Comes with all the necessary apis to allow users to control task runs from a jobfile, supporting both synchronous and asynchronous execution of tasks.

task constructor

instantiates a new task object. This does not fork/start the new process yet. Task results are piped back to the calling process using TaskPipe instance.

Parameters:

  • (path/file) (testscript) – the test-harness in the task child process

  • (str) (test_harness) –

  • (str) – testscript. Default to ats.aetest. (very advanced developers only)

  • (TaskManager) (manager) –

  • (dict) (kwargs) – harness.

property error

error

returns the task error from shared namespace.

Type:

property getter

error_handler(e)
property kwargs

kwargs

Return type:

returns the keyword argument kwargs provided to class constructor

Type:

property

property result

result

returns the task result from shared namespace. Default to Passed

Type:

property getter

run()

run task

main entry point of all task processes. This api is run in the context of the forked child process. The following is performed here:

  1. configure environment/process for test harness execution

  2. attempt to run the test-harness with the provided arguments, catch any errors that may occur and report to parent using task messaging system

  3. wrap up.

start()

start method

starts this task. Nuff said.

property taskid

taskid

Returns:

  • returns the unique task identifer. this is the same as the process

  • name.

Type:

property

terminate()

Terminate process; sends SIGTERM signal or uses TerminateProcess()

property uid

taskid

Return type:

returns the unique task identifer in a TestableId format.

Type:

property

wait(max_runtime=None)

wait method

Waits for this task to finish executing. If max_runtime is provided and exceeded, the task child process will be terminated (SIGTERM), and a TimeoutException will be raised to the caller.

Parameters:

(int) (max_runtime) – finish, before terminating it forcifully and raising errors. If not provided, waits forever.

class pyats.easypy.tasks.TaskManager(*args, **kwargs)

Bases: Component

TaskManager class

Each jobfile contains a single TaskManager, and all jobfile tasks are automatically registered to it. This allows global-management of all current running tasks & etc.

constructor

sets internal variables

Task(*args, **kwargs)
property active

active

returns the list of currently active and running task objects.

Type:

property

property count

count

number of currently managed tasks

Type:

property

property error_str

error_str

A string specifically for displaying the number of non-run tasks in the email report.

Type:

property

property errors

Returns an ordered dictionary of task names & their corresponding errors

register(task)
property result

returns roll-up of all task result codes

run(*args, max_runtime=None, **kwargs)

run api

Shortcut function to start a Task(), wait for it to finish, and return the result to the caller. This api avoids the overhead of having to deal with the task objects, and provides a black-box method of creating tasks sequentially.

Parameters:

  • (int) (max_runtime) – finish, before terminating it forcifully and raising errors. If not provided, waits forever.

  • (tuple) (args) –

  • (dict) (kwargs) –

Return type:

the task’s result code

terminate()

terminate all tasks

This api terminates all currently active tasks. If a task is parent to other children processes, its these child processes are sent the SIGTERM signal before the parent task is terminated.

pyats.easypy.tasks.run(*args, max_runtime=None, **kwargs)

run api

Shortcut function to start a Task(), wait for it to finish, and return the result to the caller. This api avoids the overhead of having to deal with the task objects, and provides a black-box method of creating tasks sequentially.

Parameters:

  • (int) (max_runtime) – finish, before terminating it forcifully and raising errors. If not provided, waits forever.

  • (tuple) (args) –

  • (dict) (kwargs) –

Return type:

the task’s result code

Module provides utility functions for easypy package.

pyats.easypy.utils.download_file_contents(url: str, filename: str, auth_token: str | None = None) → int

Downloads contents from a url into a specified file

Parameters:

  • url (str) – Address to download file from

  • filename (str) – Location to save file

  • auth_token (str, optional) – Auth header token. Defaults to None.

Raises:

Exception – Failure to download file

Returns:

Bytes downloaded

Return type:

int

pyats.easypy.utils.filter_exception(exc_type, exc_value, tb)

Filters an exception’s traceback stack and removes Easypy stack frames from it to make it more apparent that the error came from a script. Should be only used on user-script errors, and must not be used when an error is caught from pyats.easypy infra itself.

Any frame with __easypy_infra__ flag set is considered easypy infra stack.

Returns:

  • properly formatted exception message with stack trace, with easypy

  • stacks removed

pyats.easypy.utils.get_backend_group_name()

Get the group name used by backend. Checks for <BACKENDNAME>_GROUP_NAME environment variable :returns: (str) group_name

pyats.easypy.utils.get_backend_job_name()

Get the job name used by the backend system. Checks for <BACKENDNAME>_JOB_NAME environment variable :returns: (str) job_name

pyats.easypy.utils.get_backend_name()

Get the backend name by checking for known environment variables. :returns: (str) backed_name or None

pyats.easypy.utils.get_manifest_profile()

Get the manifest profile name by checking for known environment variables. :returns: (str) manifest profile or None

pyats.easypy.utils.get_request_id()

get the request ID and backend from environment variable. Tries to get the request ID and backend values from environment variable with ‘XPRESSO_REQUEST_ID’ or ‘EARMS_REQUEST_ID. :returns: (str, str) (request_id, backend)

pyats.easypy.utils.get_suite_id(jobfile, runtime)

get the unique suite ID from EARMS or XPRESSO environment variables or try to identify git info for jobfile. :returns: suite_id (str)

pyats.easypy.utils.makedirs(*paths, mode=None, exist_ok=False)

make dirs

needed because makedirs mode=x is combined with the user/process’s UMASK, requiring explicit call to os.chmod, so this api supports both creating the dirs and setting the mode in one shot.

Only creates the dirs if it doesn’t exist already. Double wammy.

Parameters:

  • (list) (paths) –

  • (oct) (mode) –

  • (bool) (exist_ok) –

pyats.easypy.utils.tozip(runinfo, archive)

Add all files in the folder to archive file.

pyats.easypy.utils.validate_suite_name(suite_name)

Check the suite name for conformance of site/repo:path syntax.

Returns:

(str or None) returns suite name if valid, otherwise None