taf.testlib.TRex

Submodules

taf.testlib.afs

afs.py

AFS-specific functionality

class taf.testlib.afs.AFS(config)

Bases: object

Basic interact with AFS.

__del__()

Disconnect all telnet connections to AFS on destroy.

__init__(config)

Initialize AFS class.

Parameters:config (dict) – Configuration

_del_port_config(port, port_map=None)

Delete maps mapped to port.

Parameters:

• port (int) – Port ID
• port_map (dict) – Port map

_get_maps_id()

Return list of map’s ids founded in running config.

Raises:AFSException – AFS running config is not available Returns:list of map’s ids founded in running config Return type:list

_get_port(connection_element)

Get AFS port number from portmap.

Parameters:connection_element (list) – Link information Raises:AFSException – invalid port number in connection element Returns:Port number Return type:int

_get_ports_from_config(connection=None)

Return AFS port number from given connection.

Parameters:connection (list) – Link information Returns:Port numbers Return type:tuple

class_logger = <logging.LoggerAdapter object>

clear_connection_pool()

Close all connections from connection pool.

get_run_config(**kwargs)

Get afs_instance for class methods.

get_sys_info(**kwargs)

Get afs_instance for class methods.

reenable_port(port, dev_id)

Set Down and than Up status for AFS port connected to given port and device.

Parameters:

• port (int) – Port ID
• dev_id (int) – Device ID

Returns:

True if workaround performed and False if skipped.

Return type:

bool

xconnect(connection=None)

Make cross connection device <-> device.

Parameters:connection (list[int]) – Link info - list [dev1Id, dev1portId, dev2Id, dev2portId] ([0, 1, 1, 24])

xdisconnect(connection=None)

Destroy cross connection device <-> device.

Parameters:connection (list[int]) – Link info - list [dev1Id, dev1portId, dev2Id, dev2portId] ([0, 1, 1, 24])

taf.testlib.afs.afs_conf_mode(function)

Decorator: get afs_instance with enabled config mode for class methods.

taf.testlib.afs.afs_normal_mode(function)

Decorator: get afs_instance for class methods.

taf.testlib.afs.get_unused_values(used_values, min_value, max_value)

Return two first unused values from range min_value - max_value.

Parameters:

• used_values (list[int]) – List of used integers
• min_value (int) – Min value
• max_value (int) – Max value

Raises:

AFSException – not enough free values

Returns:

Two first unused values from range min_value - max_value

Return type:

list[int]

taf.testlib.afs.set_vty_timeout(afs_instance, config_prompt)

Set AFS vty timeout.

Parameters:

• afs_instance (CLIGenericMixin) – Instance of AFS device
• config_prompt (str) – Prompt message

Notes

By default AFS has timeout = 1800 seconds. Such long timeout could cause excided number of allowed connection in case any error.

taf.testlib.afscross

afscross.py

xconnect-specific functionality

class taf.testlib.afscross.AFS(config, opts, env)

Bases: taf.testlib.dev_basecross.GenericXConnectMixin

Basic interact with ASF.

__init__(config, opts, env)

Reading/updating config, initialize afs object instance.

Parameters:

• config (dict) – Configuration
• opts (OptionParser) – Options
• env (Environment) – Environment object

check()

Obligatory class for entry_type = cross.

class_logger = <logging.LoggerAdapter object>

cross_clear()

Clear all connections between switches. (Not supporter for AFS environment).

Raises:CrossException – not supported method

cross_connect(conn_list)

Make connections between switches.

Parameters:conn_list (list[list[int]]) – List of connections in format: [[sw1, port1, sw2, port2], … ]

Examples:

cross_connect([[0, 1, 1, 1], [0, 2, 1, 2]])

cross_disconnect(disconn_list)

Destroy connections between switches.

Parameters:disconn_list (list[list[int]]) – List of connections in format: [[sw1, port1, sw2, port2], … ]

Examples:

cross_disconnect([[0, 1, 1, 1], [0, 2, 1, 2]])

start()

Obligatory class for entry_type = cross.

stop()

Obligatory class for entry_type = cross.

taf.testlib.caselogger

taf.testlib.cli_template

cli_template.py

Abstract class for any CLI classes

class taf.testlib.cli_template.CLIGenericMixin

Bases: object

Base class for CLI configuration.

class Raw

Bases: str

This class represents raw commands for cli object.

__init__()

Entry __init__ method defines class variable.

action_on_connect(obj, alternatives, timeout=60, interval=0.1, command_timeout=600, is_shell=False)

Performs actions on found prompts. Returns command output data. This is only for CLI connect. See action_on_expect for details.

Parameters:

• obj (Channel) – CLI object of different types.
• alternatives (tuple) –

  Tuples of (“expected line”, “action if line is found”, <Exit execution? (bool)>, <Use ones? (bool)>). action can be:

  • str - in case this is just command;
  • function - callable object to execute without parameters;

• timeout (int) – Expecting timeout.
• interval (int) – Interval between read data cycles.
• command_timeout (int) – Command execution timeout.
• is_shell (bool) – Indicates shell command

Raises:

CLIException – sudoprompt is not defined

Returns:

Full output

Return type:

str

action_on_expect(obj, alternatives, timeout=60, interval=0.1, command_timeout=600, is_shell=False)

Performs actions on found prompts. Returns command output data.

Parameters:

• obj (Channel) – CLI object of different types.
• alternatives (tuple | list) –

  Tuples of (“expected line”, “action if line is found”, <Exit execution? (bool)>, <Use ones? (bool)>). action can be:

  • str - in case this is just command;
  • function - callable object to execute without parameters;

• timeout (int) – Expecting timeout.
• interval (int | float) – Interval between read data cycles.
• command_timeout (int) – Command execution timeout.
• is_shell (bool) – Indicates shell command

Raises:

CLIException – timeout exceeded for command execution

Returns:

Full output

Return type:

str

check_shell()

Check if CLI connection is alive.

class_logger = <logging.LoggerAdapter object>

close()

Close CLI object connection.

close_shell()

Close interactive CLI shell on existing connection.

cmd_output_log(so, se)

log message normalizer.

Parameters:

• so (str) – StdOut
• se (str) – StdErr

Returns:

Normalized output

Return type:

str

exec_command()

Execute command without shell (tty).

Parameters:

• command (str) – Command to be executed.
• timeout (int) – Timeout for command execution.

Returns:

tuple of stdout, stderr, rc

Return type:

tuple(str, str, int)

expect(obj, expect_list, timeout=60, interval=0.1, remove_cmd=True, is_shell=False)

Expecting prompts and return prompt index in expect_list.

Parameters:

• obj (Channel) – CLI obj of different types.
• expect_list (list) – List of compiled re objects to find prompt in data.
• timeout (int) – Expecting timeout.
• interval (int) – Interval between read data cycles.
• remove_cmd (bool) – Flag whether to remove command from output during searching prompt in data.
• is_shell (bool) – Indicates shell command

Returns:

Found position from expect_list and full command output

Return type:

tuple

get_file()

Put file to remote host.

Parameters:

• src (str) – File’s source.
• dst (str) – File’s destination.

login()

Do CLI object login procedure.

Parameters:

• username (str) – Host login (string).
• password (str) – Host password(string).
• timeout (int) – Time to execute login procedure (integer).
• wait_login (int) – time to wait login prompt before sending <Enter>. <Enter> is necessary if login prompt has been already appeared before connection is established.
• alternatives (list of tuples) – list of alternative prompts and actions.
• connect (bool) – Flag if connection should be established before login procedure (bool).

Returns:

None

normalize_output(data, command, ret_code, end_pattern=None)

Removes command and command’s end flag from output data. Extracts return code of the command.

Parameters:

• data (str) – Output data of command.
• command (str) – Executed command.
• ret_code (bool) – Flag which shows if return code command was added to main command.
• end_pattern (str) – pattern which is used to find end command flag.

Returns:

data and return code

Return type:

tuple

open_shell()

Create interactive CLI shell on existing connection.

prepare_alter(command, alternatives=None, sudo=False, ret_code=True, page_break=None)

Adds specified alternatives to list, updates command with end command and sudo if needed.

Parameters:

• command (str) – Command to be executed.
• alternatives (tuple) –

  Tuples of (“expected line”, “action if line is found”, <Exit execution? (bool)>, <Use ones? (bool)>). action can be:

  • str - in case this is just command;
  • function - callable object to execute without parameters;

• sudo (bool) – Flag if sudo should be added to the list of alternatives .
• ret_code (bool) – Flag if return code should be added to the list of alternatives.
• page_break (bool) – Flag if page break should be added to the list of alternatives.

Returns:

command, alternatives, end_pattern

Return type:

tuple

prepare_pexpect_obj(pexp_obj)

Add read(), write() and expect() methods to emulate object IO methods.

Parameters:pexp_obj (pexpect) – pexpect object. Returns:pexpect object. Return type:pexpect

prepare_ssh_shell_obj(shell)

Add read(), write() and expect() methods to emulate object IO methods.

Parameters:shell (paramiko.Channel) – paramiko.Channel object. Returns:paramiko.Channel object. Return type:paramiko.Channel

prepare_telnet_obj(telnet_obj)

Add read() and expect() methods to object to emulate object IO methods.

Parameters:telnet_obj (telnetlib) – telnetlib object. Returns:telnetlib object Return type:telnetlib

put_file()

Transfer file from/to remote host.

Parameters:

• src (str) – File’s source.
• dst (str) – File’s destination.

randstr(length)

Return random string with required length.

Parameters:length (int) – Required length

send_command()

Run command without waiting response.

Parameters:command (str) – Command to be executed.

shell_command()

Run interactive command on previously created shell (tty).

Parameters:

• command (str) – Command to be executed.
• alternatives (tuple) –

  Tuples of (“expected line”, “action if line is found”, <Exit execution? (bool)>, <Use ones? (bool)>). action can be:

  • str - in case this is just command;
  • function - callable object to execute without parameters;

• timeout (int) – Expecting timeout.
• sudo (bool) – Flag if sudo should be added to the list of alternatives.
• ret_code (bool) – Flag if return code should be added to the list of alternatives.
• expected_rc (int) – Sets return code and verifies if return code of executed command the same as expected return code (int or str).
• quiet (bool) – Flag to verify if expected return equals expected.
• raw_output (bool) – Flag whether to return ‘pure’ output.

shell_read()

Read data from output buffer.

Parameters:

• timeout (int) – Increases time to read data from output buffer.
• interval (int) – Time delay between attempts to read data from output buffer.

class taf.testlib.cli_template.CmdStatus(stdout, stderr, rc)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, stdout, stderr, rc)

Create new instance of CmdStatus(stdout, stderr, rc)

__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>, len=<built-in function len>)

Make a new CmdStatus object from a sequence or iterable

_replace(**kwds)

Return a new CmdStatus object replacing specified fields with new values

rc

Alias for field number 2

stderr

Alias for field number 1

stdout

Alias for field number 0

class taf.testlib.cli_template.Raw

Bases: str

This class represents raw commands for cli object.

taf.testlib.clicmd_iss

clicmd_iss.py

Module for CLI specific functionality

class taf.testlib.clicmd_iss.CLICmd(ipaddr, port, login, passw, prompt, devtype, delay=None, build_path=None, img_path=None, page_break='<?> - help.', xmlrpcport=None)

Bases: object

Class for CLI specific functionality.

Parameters:

• config – environment config.
• switches – switches list.

__init__(ipaddr, port, login, passw, prompt, devtype, delay=None, build_path=None, img_path=None, page_break='<?> - help.', xmlrpcport=None)

Initialize CLICmd class

_connect_to_switch(prompt, timeout=20)

SSH connect to switch and wait untill prompt string appeared.

Parameters:

• prompt (str) – expected CLI prompt.
• timeout (int) – connection timeout.

Returns:

None

Examples:

self._connect_to_switches(sw_keys=1, prompt="Switch ")

cli_connect(prompt=None)

SSH connect to switch and wait untill prompt string appeared.

Parameters:prompt (str) – expected CLI prompt. Returns:None

Examples:

env.switch[1].cli.cli_connect(prompt="Switch ")

cli_disconnect()

Close ssh connection to switch.

Raises:CLICMDException – error on disconnect Returns:None

Examples:

env.switch[1].cli.cli_disconnect()

cli_get(arguments_list, prompt=None, show=True, timeout=25)

Getting values by CLI.

Parameters:

• arguments_list (list) – list of arguments to get values.
• prompt (str) – expected promt or message, takes from cli_set_result.
• show (bool) – execute command with show prefix
• timeout (int) –

Raises:

Exception – error on command execution

Returns:

List of CLI-GET command results.

Return type:

list

Examples:

env.switch[1].cli.cli_get(['enable, none 0 none', 'statistics, Port 1, RX Discards']])

cli_get_all(arguments_list, prompt=None, timeout=25, interval=0.1)

Getting values by CLI.

Parameters:

• arguments_list (list) – list of arguments to get values.
• prompt (str) – expected promt or message, takes from cli_set_result.
• timeout (int) – command execution timeout
• interval (int) – time interval between read attempts

Raises:

Exception – error on command execution

Returns:

List of CLI-GET command results.

Return type:

list

cli_set(commands_list, timeout=5, prompt=None, connect=True)

Setting values by CLI.

Parameters:

• commands_list (list) – list of commands.
• (str) – expected promt or message, takes from cli_set_result.
• timeout (int) – commnad execution timeout
• connect (bool) – Flag if connection should be established before login procedure.

Raises:

Exception – error on command execution

Returns:

List of CLI-SET command results.

Return type:

list

Examples:

env.switch[1].cli.cli_set([["enable"], ["vlan-database"], ["vlan 10"]])

suite_logger = <logging.LoggerAdapter object>

update_prompt(prompt)

Updating prompt in both clissh and clicmd_ons objects.

Parameters:prompt (str) – Prompt to be updated

taf.testlib.clicmd_iss.get_column_names(table_data, column_ranges)

Get column name.

Parameters:

• table_data (str) – String fetched from SSH CLI with column names
• column_ranges (list) – List with columns width string indexes used to extract column names

Returns:

column_names_list - List of strings with columns names

Return type:

list[str]

taf.testlib.clicmd_iss.get_column_ranges(line)

Get column ranges.

Parameters:line (str) – string with “–” column delimiters Returns:column_indexes - list of lists with string indexes of columns Return type:list[list[str]]

taf.testlib.clicmd_iss.get_dotted_table(table_lines)

Get table data.

Parameters:table_lines (list) – list of table rows Returns:table_list - list of lists with row names and values Return type:list[list]

taf.testlib.clicmd_iss.get_table_title(line)

Get table name.

Parameters:line (str) – output

taf.testlib.clicmd_iss.get_table_value(table_data, identifier=None, checker=None)

Gets necessary field value from the table.

Parameters:

• table_data (str) – console output data (string),
• identifier (list[]) – Column name and row number[‘column’, ‘value’],
• checker (str) – column name to check value (string).

Raises:

CLICMDException – invalid row length

Returns:

Field value.

Return type:

str

taf.testlib.clicmd_ons

taf.testlib.clinns

clinns.py

Module contains classes for managing device using SSH connection or SSH connection emulation for Linux Network

class taf.testlib.clinns.CLISSHNetNS(nsname, port=None, username=None, password=None, page_break=None, prompt=None, pass_prompt=None, sudo_prompt=None, login_prompt=None, page_break_lines=None, exit_cmd=None, timeout=10, quiet=False)

Bases: taf.testlib.cli_template.CLIGenericMixin

Class for executing command inside of namespace. Unused parameters added to support the same interface for other CLI classes.

Examples:

client = CLISSHNetNS("some_nns", timeout=10)
client.exec_command("some_command")

__init__(nsname, port=None, username=None, password=None, page_break=None, prompt=None, pass_prompt=None, sudo_prompt=None, login_prompt=None, page_break_lines=None, exit_cmd=None, timeout=10, quiet=False)

Initialize CLISSHNetNS class.

Parameters:

• nsname (str) – NNS name.
• port (int) – Host port.
• username (str) – Host user
• password (str) – Host password
• page_break (str) – Page brake marker.
• prompt (str) – Shell prompt.
• pass_prompt (str) – Login password prompt.
• sudo_prompt (str) – Sudo password prompt.
• login_prompt (str) – Login prompt.
• page_break_lines (int) – Number of page brake lines.
• exit_cmd (str) – Command to perform telnet exit.
• timeout (int) – Default timeout for commands.
• quiet (bool) – Flag for return code verification.

check_shell()

Shell is always existed for PopenPexpect.

class_logger = <logging.LoggerAdapter object>

close()

PopenPexpect object is always opened and doesn’t have ‘close’ method.

close_shell()

Shell is always opened for PopenPexpect.

exec_command(**kwargs)

Add “ip netns exec ” prefix to all commands to forward them into network namespace.

get_file(src, dst)

Put file to remote host.

Parameters:

• src (str) – File’s source.
• dst (str) – File’s destination.

login(*args, **kwargs)

Shell is always opened for PopenPexpect.

native_cmd(command, shell=False, wait=True, verbose=True)

Execute open command.

Parameters:

• command (str) – Command to be executed.
• shell (bool) – Flag specifies whether to use the shell as the program to execute.
• wait (bool) – Flag specifies whether to wait for command output.
• verbose (bool) – Flag specifies whether to print command output.

open_shell()

Shell is always opened for PopenPexpect.

put_file(src, dst)

Transfer file from/to remote host.

Parameters:

• src (str) – File’s source.
• dst (str) – File’s destination.

send_command(**kwargs)

Add “ip netns exec ” prefix to all commands to forward them into network namespace.

shell_command(**kwargs)

Add “ip netns exec ” prefix to all commands to forward them into network namespace.

shell_read(timeout=0, interval=0.1)

Read data from output buffer.

Parameters:

• timeout (int) – Increases time to read data from output buffer.
• interval (int) – Time delay between attempts to read data from output buffer.

taf.testlib.clinns.ip_net_namespace_mode(function)

Decorator: get clissh_instance for class methods.

taf.testlib.clissh

taf.testlib.clitelnet

clitelnet.py

Basic telnet class with command oriented functionality

class taf.testlib.clitelnet.TelnetCMD(host=None, port=23, username=None, password=None, page_break='--More--', prompt=None, pass_prompt='Password: ', sudo_prompt=None, timeout=10, login_prompt='login: ', page_break_lines=3, exit_cmd=None, quiet=False)

Bases: taf.testlib.cli_template.CLIGenericMixin

HighLevel telnet command oriented class. Unused parameters added to support the same interface for other CLI classes.

Examples:

client = TelnetCMD("1.1.1.1", 22)
client.login("username", "paSSword")

__init__(host=None, port=23, username=None, password=None, page_break='--More--', prompt=None, pass_prompt='Password: ', sudo_prompt=None, timeout=10, login_prompt='login: ', page_break_lines=3, exit_cmd=None, quiet=False)

Initialize TelnetCMD class.

Parameters:

• host (str) – Target host IP address.
• port (int) – SSH port.
• username (str) – SSH login user.
• password (str) – SSH user password.
• page_break (str) – Page brake marker.
• prompt (str, list[str]) – Shell prompt or list of shell prompts.
• pass_prompt (str) – Login password prompt.
• sudo_prompt (str) – Sudo password prompt.
• timeout (int) – Default timeout for commands.
• login_prompt (str) – Login prompt.
• page_break_lines (int) – Number of page brake lines.
• exit_cmd (str) – Command to perform telnet exit (str).
• quiet (bool) – Flag for return code verification.

_check_telnet_obj()

Check if telnet object exists (connection is established).

Raises:Exception – telnet connection is not established

_check_telnet_obj_connection()

Verify if telnet connection exists.

Returns:True if telnet connection exists Return type:bool

_normalize_output(output=None, cmd=None, prompt=None)

Remove everything from the response except the actual command output.

Parameters:

• output (str) – Output data to be normalized.
• cmd (str) – Command which was used for program execution.
• prompt (str) – Prompt which will be removed form output data.

check_shell()

Check if CLI connection is alive.

Raises:CLITelnetException – telnet connection is not established; user is not logged in

class_logger = <logging.LoggerAdapter object>

close()

Close CLI object connection.

close_shell()

Close interactive CLI shell on existing connection.

connect(with_login=True, wait_login=5, alternatives=None, wait_prompt=True, socket_closed=False)

Create telnet connection and do login if necessary.

Parameters:

• with_login (bool) – Perform login procedure or not. If param isn’t set try automatically determine login necessity. (True|False|None)
• wait_login (int) – time to wait login before sending <Enter>. <Enter> is necessary if login is already appeared.
• alternatives (tuple) –

  Tuples of (“expected line”, “action if line is found”, <Exit execution? (bool)>, <Use ones? (bool)>). action can be:

  • str - in case this is just command;
  • function - callable object to execute without parameters;

• wait_prompt (bool) – Wait for prompt message or not.
• socket_closed (bool) – Determines if socket has been closed or not.

Raises:

Exception – self.host is None; self.prompt is None

Returns:

telnet stdout

Return type:

str

disconnect(with_exit=True)

Do disconnect.

Parameters:with_exit (bool) – Flag specifies whether perform exit procedure.

enter_mode(cmd=None, new_prompt=None)

Enter config/priv or other mode with specific prompt.

Parameters:

• cmd (str) – Command to change mode.
• new_prompt (str) – Prompt which will replace current prompt after successful mode changing.

Raises:

Exception – undefined prompt, unexpected new prompt

Note

After success execution current prompt will be replaced with new propmt and saved in prompt_stack.

exec_command(command, timeout=None, sudo=False, ret_code=False)

Execute command without shell (tty).

Parameters:

• command (str) – Command to be executed.
• timeout (int) – Timeout for command execution.
• sudo (bool) – Flag if sudo should be added to the list of alternatives.
• ret_code (bool) – Flag if return code should be added to the list of alternatives (bool).

Returns:

output, “”, return code

Return type:

tuple(str, str, int)

exit(wait_close=True)

Do telnet exit/logout procedure. Wait until connection closed.

Parameters:wait_close (bool) – Flag specifies whether to verify successful exit and to return output data.

exit_mode(exit_cmd=None)

Exit config/priv or other mode with specific prompt.

Parameters:exit_cmd (str) – Command to exit from current mode. Raises:Exception – undefined prompt, unexpected new prompt

get_file()

This method isn’t supported by telnetlib.

Raises:CLITelnetException – unsupported

login(username=None, password=None, timeout=None, wait_login=0, alternatives=None, connect=True)

Do CLI object login procedure.

Parameters:

• username (str) – Host login (string).
• password (str) – Host password(string).
• timeout (int) – Time to execute login procedure (integer).
• wait_login (int) – time to wait login prompt before sending <Enter>. <Enter> is necessary if login prompt has been already appeared before connection is established.
• alternatives (list of tuples) – list of alternative prompts and actions.
• connect (bool) – Flag if connection should be established before login procedure (bool).

Returns:

None

Raises:

• Exception – username is not defined
• CLITelnetException – login timeout exceeded, unexpected login prompt

open_shell(raw_output=False)

Call login method. Added to support other CLI objects interface.

Parameters:raw_output (bool) – Flag whether to read output buffer. Raises:CLITelnetException – telnet connection is not established Returns:telnet output Return type:str

put_file()

This method isn’t supported by telnetlib.

Raises:CLITelnetException – unsupported

send_command(command)

Run command without waiting response.

Parameters:command (str) – Command to be executed.

shell_command(command, alternatives=None, timeout=None, sudo=False, ret_code=True, new_prompt=None, expected_rc='0', quiet=None, raw_output=False, interval=0.1)

Run interactive command on previously created shell (tty).

Parameters:

• command (str) – Command to be executed.
• alternatives (tuple) –

  Tuples of (“expected line”, “action if line is found”, <Exit execution? (bool)>, <Use ones? (bool)>). action can be:

  • str - in case this is just command;
  • function - callable object to execute without parameters;

• timeout (int) – Expecting timeout.
• sudo (bool) – Flag if sudo should be added to the list of alternatives.
• ret_code (bool) – Flag if return code should be added to the list of alternatives.
• expected_rc (int) – Sets return code and verifies if return code of executed command the same as expected return code (int or str).
• quiet (bool) – Flag to verify if expected return equals expected.
• raw_output (bool) – Flag whether to return ‘pure’ output.
• interval (int | float) – Interval between read data cycles.
• new_prompt (str) – Prompt which will replace current prompt after successful mode changing.

Raises:

CLITelnetException – unexpected return code

shell_read(timeout=0, interval=0.1)

Read data from output buffer.

Parameters:

• timeout (int) – Increases time to read data from output buffer.
• interval (int) – Time delay between attempts to read data from output buffer.

taf.testlib.common3

common3.py

Testlib common functionality version 3.x

class taf.testlib.common3.Cross(setup, env)

Bases: dict

New interface to cross object without device id.

__init__(setup, env)

Initialize Cross class.

cross_connect(conn_list)

Wrapper for cross_connect method defined in xconnect.py module.

Parameters:conn_list (list[list]) – List of connections Raises:Exception – conn_list is empty

cross_disconnect(disconn_list)

Wrapper for cross_disconnect method defined in xconnect.py module.

Parameters:disconn_list (list[list]) – List of connections Raises:Exception – disconn_list is empty

device_port_connect(dev_id, port_no)

Connect/Disconnect device port.

Parameters:

• dev_id (str) – Device ID/autoname/linkname (‘tg1’)
• port_no (int) – Device port number.

device_port_disconnect(dev_id, port_no)

Connect/Disconnect device port.

Parameters:

• dev_id (str) – Device ID/autoname/linkname (‘tg1’)
• port_no (int) – Device port number.

get_connection(dev_id, port_no)

Get connection for device port.

Parameters:

• dev_id (str) – Device ID/autoname/linkname (‘tg1’)
• port_no (int) – Device port number.

Raises:

Exception – no connection for current port

Returns:

Connection info

Return type:

list

get_device_id(connection)

Search device in setup object by given connection.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2] Raises:Exception – no device in connection Returns:device id which own connection Return type:int

xconnect(connection)

Wrapper for xconnect method defined in xconnect.py module.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

xdisconnect(connection)

Wrapper for xdisconnect method defined in xconnect.py module.

Parameters:connection (list) – in format [sw1, port1, sw2, port2]

class taf.testlib.common3.Environment(opts=None, **kwargs)

Bases: dict

Main class of all test environment.

Notes

This class has to be used as base fixture in all test cases. It provides number of common methods to initialize, shutdown, cleanup environment functions which basically call appropriate methods of particular device classes.

__init__(opts=None, **kwargs)

Read configuration files and create device objects.

Parameters:opts (OptionParser) – py.test config.option object which contains all py.test cli options. Raises:TAFCoreException – unexpected entry_type

_append_related_confs(conf_ids)

Create dictionary with related device configurations.

Parameters:conf_ids (list[str]) – List of related config IDs. Raises:Exception – configuration is not found for specific device ID Returns:Dictionary with related device configurations Return type:dict

_get_conf(file_name=None)

Load environment config from file.

Parameters:

file_name (str) – Name of a json file with a test environment configuration.

Raises:

• TAFCoreException – configuration file is not found
• IOError – error on reading configuration file

Returns:

dict of the selected configuration.

Return type:

dict

Notes

This method shouldn’t be used outside this class. Use “config” attribute to access environment configuration.

_get_device_conf(device_id)

Return config entry by given Id if one, else return None.

Parameters:device_id (str) – Entry ID. Returns:Entry config. Return type:dict

_get_prio_dict(prio)

Return dict of entries by prio.

Parameters:prio (str) – Priority name to order dict by . Returns:dict of lists where key = priority, value = list of device objects. Return type:dict

_get_setup(file_name)

Reads setup file based on provided name.

Parameters:

file_name (str) – Name of a json file with setup.

Raises:

• TAFCoreException – setup file is not found
• IOError – error on reading setup file

Returns:

setup json content.

Return type:

list[dict]

check(entry_types=None)

Checking environment.

Parameters:entry_types (list[str]) – List of entry types

class_logger = <logging.LoggerAdapter object>

cleanup(entry_types=None)

Cleaning environment.

Parameters:entry_types (list[str]) – List of entry types

create_conf_entry(setup_entry)

get_device_id(dut)

Search device in config object by device name.

Parameters:dut (str) – Could be one of: device LINK_NAME, ‘autoname’ or ‘id’ from config. Raises:TAFCoreException – unknown device type Returns:Device id which configured. Return type:str, int

Examples (Config object like):

{
    "env": [
            {"id": 5, "port_list": [["port1", 10000], ["port2", 40000]},
            {"id": 9, "autoname": "DEV2", "port_list": [["port1", 10000], ["port2", 40000]}
           ]
    "cross": {"ID": [[5, 1, 9, 2], [5, 2, 9, 1]]}
}

Result is:

# by LINK_NAME
env.get_device_id("sw1") == 5
# by "autoname"
env.get_device_id("DEV2") == 9
# by ID
env.get_device_id(9) == 9

get_port_speed(dut, port_id)

Search speed port in config object namely in ‘port_list’ by device name and port Id.

Parameters:

• dut (str) – Could be one of: device LINK_NAME, ‘autoname’ or ‘id’ from config.
• port_id (int) – Port Id from config object (ids starts from 1)

Raises:

TAFCoreException – port is not present in configuration’s ‘port_list’

Returns:

Port speed or exception if there is no port with given Id in config.

Return type:

int

Examples (Config object like):

{
    "env": [
            {"id": 5, "autoname": "DEV1", "port_list": [["port1", 10000], ["port2", 40000]},
            {"id": 9, "ports": ["port10", 11]}
           ]
    "cross": {"ID": [[5, 1, 9, 2], [5, 2, 9, 1]]}
}

Result is:

env.get_port_speed("sw1", 2) == 40000
env.get_port_speed("DEV1", 1) == 10000

get_ports(links=None)

Returns dictionary of ports based on links between devices.

Parameters:links (list[list]) – List of devices in format [[‘dev1’, ‘dev2’, number_of_links, port_speed], ] (list of lists). Where:  number_of_links - optional parameter(int or enum - “ALL”);  port_speed - optional parameter. Raises:TAFCoreException – wrong link format Returns:ports Return type:dict

Examples (Config object like):

{
    "env": [
            {"id": 99, "autoname": "DEV1", "port_list": [["port1", 10000], ["port2", 40000], ["port3", 10000]},
            {"id": 100, "port_list": [["port10", 40000], [11, 10000], ["port12", 40000]}
           ]
    "cross": {"ID": [[99, 1, 100, 2], [99, 2, 100, 1]]}
}

Result is:

ports = env.get_ports([['sw1', 'sw2', 1], ])
assert ports == {('sw2', 'sw1'): {1: "port10"}, ('sw1', 'sw2'): {1: "port1"}}

ports = env.get_ports([['DEV1', 'sw2', 2], ])
assert ports == {('sw2', 'sw1'): {1: "port10", 2: 11}, ('sw1', 'sw2'): {1: "port1", 2: "port2"}}

# with optional parameter "port_speed"
ports = env.get_ports([['sw1', 'sw2', 1, 10000], ])
assert ports == {('sw1', 'sw2'): {1: "port1"}, ('sw2', 'sw1'): {1: "11"}}

# Method returns all links between devices if no any optional parameters
ports = env.get_ports([['sw1', 'sw2', ], ])
assert ports == {('sw1', 'sw2'): {1: "port1", 2: "port2"}, ('sw2', 'sw1'): {1: "port10", 2: 11}}

# The same with enum "ALL"
ports = env.get_ports([['sw1', 'sw2', "ALL"], ])
assert ports == {('sw1', 'sw2'): {1: "port1", 2: "port2"}, ('sw2', 'sw1'): {1: "port10", 2: 11}}

# With optional parameters "port_speed" and "ALL"
ports = env.get_ports([['sw1', 'sw2', "ALL", 40000], ])
assert ports == {('sw1', 'sw2'): {1: "port2"}, ('sw2', 'sw1'): {1: "port10"}}

# Method returns all links between devices if no parameter
ports = env.get_ports()
assert ports == {('sw1', 'sw2'): {1: "port1", 2: "port2"}, ('sw2', 'sw1'): {1: "port10", 2: 11}}

get_real_port_name(dut, port_id)

Search real port number/name by device name and port Id in config object.

Parameters:

• dut (str) – Could be one of: device LINK_NAME, ‘autoname’ or ‘id’ from config.
• port_id (int) – Port Id from config object (ids starts from 1).

Raises:

TAFCoreException – port_id is not found in configuration; device doesn’t have ports or port_list attributes

Returns:

Real port number/name or exception if there is no port with given Id in config.

Return type:

int, str

Examples (Config object like):

{
    "env": [
            {"id": 99, "autoname": "DEV1", "port_list": [["port1", 10000], ["port2", 10000]},
            {"id": 100, "ports": ["port10", 11]}
           ]
    "cross": {"ID": [[99, 1, 100, 2], [99, 2, 100, 1]]}
}

Result is:

# by LINK_MAME
env.get_real_port_name("sw2", 2) == 11
# by "autoname"
env.get_real_port_name("DEV1", 1) == "port1"

id2instance(device_id)

Returns entry instance by device id.

Parameters:device_id (str) – Could be one of: device LINK_NAME, ‘autoname’ or ‘id’ from config. Returns:Entry instance Return type:GenericEntry

Examples:

# by LINK_NAME
env.id2instance("sw1")
# by "autoname"
env.get_device_id("DEV2")
# by ID
env.get_device_id("9")

initialize(entry_types=None)

Initialize test environment.

Parameters:entry_types (list[str]) – List of entry types

ordered_action(action, prio, entry_types)

Perform action on entries with type in entry_types and ordered by prio.

Parameters:

• action (str) – method name to execute.
• prio (str) – priority name to sort objects by.
• entry_types (list[str]) – entry types to apply action (apply action to all entry types if None).

Returns:

None

parallelize(objects, method, safe=False)

Run objects method in multiple threads.

Parameters:

• objects (list[GenericEntry]) – list of device objects.
• method (str) – method name that has to be executed.
• safe (bool) – Hide exception raisings, but print log message.

Returns:

None

Examples:

objects = [env.lhost[1], env.lhost[2]]
env.parallelize(objects, "cleanup", False)

safe_executor(obj, method, *args, **kwargs)

Invokes obj.method(*args, **kwargs) in try block and return error message with traceback.

Parameters:

• obj (GenericEntry) – Entry instance
• method (str) – method name that has to be executed

Returns:

Error message with traceback

Return type:

str

Warning

• Don’t use in case obj.method has to return something.
• Don’t use in case an exception has to be handled by py.test.

sanitize(entry_types=None)

Sanitizing environment.

Parameters:entry_types (list[str]) – List of entry types

shutdown(entry_types=None)

Stopping/Disconnecting environment.

Parameters:entry_types (list[str]) – List of entry types

Note

This method cares to release all environment even an exception is raised during destroy process.

taf.testlib.common3.softexit(message, env=None)

Sanitizing environment and exit py.test execution.

Parameters:

• message (str) – Exit message
• env (Environment) – Environment instance

taf.testlib.connpool

connpool.py

Connection Pool class

class taf.testlib.connpool.ConnectionPool(connection_class=None, max_connections=None, time_to_live=30, **connection_kwargs)

Bases: object

Generic connection pool`

__init__(connection_class=None, max_connections=None, time_to_live=30, **connection_kwargs)

Initialize ConnectionPool class.

Parameters:

• connection_class (TelnetCMD) – Telnet connection class
• max_connections (int) – Maximum available connections
• time_to_live (int) – Time to live for each connection
• connection_kwargs (dict) – Connection arguments

_created_connections()

Return number of created connections

Returns:Number of created connections Return type:int

class_logger = <logging.LoggerAdapter object>

del_connection_in_use(connection)

Delete connection in use (in case any error).

Parameters:connection (TelnetCMD) – Telnet connection

disconnect_all()

Disconnects all connections in the pool.

get_connection()

Get a connection from the pool.

Raises:ConnPoolException – connection is dead, error on connection creation Returns:Telnet connection Return type:TelnetCMD

make_connection()

Create a new connection.

Raises:ConnPoolException – too many connections Returns:Telnet connection Return type:TelnetCMD

release(connection)

Releases the connection back to the pool.

Parameters:connection (TelnetCMD) – Telnet connection

taf.testlib.custom_exceptions

custom_exceptions.py

Class to handle custom exceptions with basic introspection

exception taf.testlib.custom_exceptions.AFSException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle AFS exceptions with basic introspection.

exception taf.testlib.custom_exceptions.AccessError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for any access errors.

exception taf.testlib.custom_exceptions.ArgumentError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for any argument errors.

exception taf.testlib.custom_exceptions.ArgumentsCollision(**kwargs)

Bases: taf.testlib.custom_exceptions.CmdArgsException

exception taf.testlib.custom_exceptions.BoundaryError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for any boundary errors.

exception taf.testlib.custom_exceptions.CLICMDException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle clicmd exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CLIException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle CLI exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CLINNSException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CLIException

Base class to handle clinns exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CLISSHException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CLIException

Base class to handle clissh exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CLITelnetException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CLIException

Base class to handle clitelnet exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CmdArgsException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle command arguments exceptions with basic introspection.

exception taf.testlib.custom_exceptions.ConnPoolException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle ConnPoll exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CrossException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Xconnect exceptions with basic introspection.

exception taf.testlib.custom_exceptions.CustomException(value, trace=False)

Bases: Exception

Base class to handle custom exceptions with basic introspection.

__init__(value, trace=False)

Initialize CustomException class.

__str__()

String representation.

class_logger = <logging.LoggerAdapter object>

exception taf.testlib.custom_exceptions.ExistsError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for any duplication errors.

exception taf.testlib.custom_exceptions.HubException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Hub exceptions with basic introspection.

exception taf.testlib.custom_exceptions.InvalidCommandError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for any naming errors.

exception taf.testlib.custom_exceptions.InvalidNameError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for any naming errors.

exception taf.testlib.custom_exceptions.IxiaException(value, trace=False)

Bases: taf.testlib.custom_exceptions.TGException

Base class to handle IXIA exceptions with basic introspection.

exception taf.testlib.custom_exceptions.NotExistsError(value, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Abstracted error class for accessing non-existent parameters.

exception taf.testlib.custom_exceptions.OpenStackNoSuchImage

Bases: Exception

exception taf.testlib.custom_exceptions.OvsControllerException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Ovs Controller exceptions with basic introspection.

exception taf.testlib.custom_exceptions.PypackerException(value, trace=False)

Bases: taf.testlib.custom_exceptions.TGException

Base class to handle Pypacker exceptions with basic introspection.

exception taf.testlib.custom_exceptions.SshExecCommandFailed(command, ret)

Bases: Exception

exception taf.testlib.custom_exceptions.SwitchException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Switch operation exceptions with basic introspection.

exception taf.testlib.custom_exceptions.SysLogException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle SysLog exceptions with basic introspection.

exception taf.testlib.custom_exceptions.TAFCoreException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Switch operation exceptions with basic introspection.

exception taf.testlib.custom_exceptions.TAFLegacyException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Pypacker exceptions with basic introspection.

exception taf.testlib.custom_exceptions.TGException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

General TG exception.

exception taf.testlib.custom_exceptions.ToolException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle Linux Tool operation exceptions with basic introspection.

exception taf.testlib.custom_exceptions.TrexException(value, trace=False)

Bases: taf.testlib.custom_exceptions.TGException

Base class to handle TRex exceptions with basic introspection.

exception taf.testlib.custom_exceptions.UICmdException(value, command, stdout, stderr, rc, trace=False)

Bases: taf.testlib.custom_exceptions.UIException

Base class to handle test_ui command exceptions with basic introspection.

exception taf.testlib.custom_exceptions.UIException(value, trace=False)

Bases: taf.testlib.custom_exceptions.CustomException

Base class to handle test_ui exceptions with basic introspection.

exception taf.testlib.custom_exceptions.UnknownArguments(**kwargs)

Bases: taf.testlib.custom_exceptions.CmdArgsException

taf.testlib.dev_basecross

dev_basecross.py

Cross connect specific functionality

class taf.testlib.dev_basecross.GenericXConnectMixin(config, opts)

Bases: taf.testlib.entry_template.GenericEntry

General Cross object functionality.

Configuration examples:

{
 "name": "Zero Cross",
 "entry_type": "cross",
 "instance_type": "zero",
 "id": 31
}

Where:

•  entry_type and  instance_type are mandatory values and cannot be changed for current device type.

 id - int or str uniq device ID (mandatory)

 name - User defined device name (optional)

check()

Mandatory method for environment specific switch classes.

cleanup()

Mandatory method for environment specific switch classes.

create()

Create Cross connections.

destroy()

Destroy Cross connections.

sanitize()

Mandatory method for environment specific switch classes.

class taf.testlib.dev_basecross.ZeroCross(config, opts)

Bases: taf.testlib.dev_basecross.GenericXConnectMixin

Stub for cross object in environment. It should be used for static connected environment.

__init__(config, opts)

Initialize ZeroCross class

_get_device_from_environment(device_id)

Mandatory method for environment specific switch classes.

Parameters:device_id (str) – Device ID/autoname/linkname (‘tg1’)

cross_clear()

Mandatory method for environment specific switch classes.

cross_connect(conn_list=None)

Mandatory method for environment specific switch classes.

Parameters:conn_list (list[list]) – List of connections

cross_disconnect(disconn_list=None)

Mandatory method for environment specific switch classes.

Parameters:disconn_list (list[list]) – List of connections

xconnect(connection=None)

Mandatory method for environment specific switch classes.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

xdisconnect(connection=None)

Mandatory method for environment specific switch classes.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

taf.testlib.dev_chef

taf.testlib.dev_iperftg

taf.testlib.dev_ixia

taf.testlib.dev_linux_host

taf.testlib.dev_linux_host_vm

taf.testlib.dev_mrvcross

dev_mrvcross.py

MRV cross specific functionality

class taf.testlib.dev_mrvcross.StaticCrossMRV(config, opts)

Bases: taf.testlib.dev_staticcross_ons.StaticCrossONS

Cross connection device based on MRV layer 1 switch.

taf.testlib.dev_onsswcross

dev_onsswcross.py

`Cross Connection device based on ONS switches

class taf.testlib.dev_onsswcross.ONSSwitchCross(config, opts)

Bases: taf.testlib.dev_basecross.GenericXConnectMixin

Cross connection device based on ONS switch.

Configuration dictionary example:

{"id": "ONS_switch_cross_ID", "entry_type": "cross", "instance_type": "onssw",
 "ipaddr": "10.0.5.101", "port": "8081",
 "portmap":[[10, 1, 3], [10, 2, 4], [11, 1, 5], [11, 2, 6]]}

Where portmap: [[<device ID>, <device port ID>, <self ONS switch port number>], ]

__init__(config, opts)

Initialize ONSSwitchCross class.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

_add_vlan_tube(port1, port2)

Create connection between 2 ports with VLANs.

Parameters:

• port1 (int) – Port ID
• port2 (int) – Port ID

_clear_port_vlan_cfg(port)

Remove port from VLAN if such configuration exists.

Parameters:port (int) – Port ID

_del_vlan_tube(port1, port2)

Remove VLAN connection betwen 2 ports.

Parameters:

• port1 (int) – Port ID
• port2 (int) – Port ID

_get_free_vlan()

Return free vlan id.

Raises:Exception – no free vlans Returns:Free vlan id Return type:int

_get_self_port(dev_id, dev_port)

Return self port id by connected device id and port.

Parameters:

• dev_id (int) – Device ID
• dev_port (int) – Device’s port ID

Raises:

Exception – no device/port in port map

Returns:

self port id

Return type:

int

class_logger = <logging.LoggerAdapter object>

cross_connect(conn_list=None)

Peform all connections from conn_list.

Parameters:conn_list (list[list]) – List of connections

cross_disconnect(disconn_list=None)

Destroy all connections from conn_list.

Parameters:disconn_list (list[list]) – List of connections

xconnect(conn)

Perform single connection.

Parameters:conn (list) – Connection info in format [sw1, port1, sw2, port2]

xdisconnect(conn)

Destroy single connection.

Parameters:conn (list) – Connection info in format [sw1, port1, sw2, port2]

taf.testlib.dev_ovscontroller

taf.testlib.dev_settings

dev_settings.py

Settings entry class. ‘settings’ entry can contain any setup related info that is necessary for testing on that setup

class taf.testlib.dev_settings.GenericSettings(config, opts)

Bases: taf.testlib.entry_template.GenericEntry

Settings entry is used just to pass necessary info to tests and doesn’t require any actions.

Notes

Class returns settings dictionary without modifications on call.

__get__(instance, owner)

Get configuration.

check()

Mandatory method for environment specific switch classes.

cleanup()

Mandatory method for environment specific switch classes.

create()

Mandatory method for environment specific switch classes.

destroy()

Mandatory method for environment specific switch classes.

sanitize()

Mandatory method for environment specific switch classes.

taf.testlib.dev_staticcross_ons

dev_staticcross_ons.py

Staticcross_ons-specific functionality

class taf.testlib.dev_staticcross_ons.StaticCrossONS(config, opts)

Bases: taf.testlib.dev_basecross.GenericXConnectMixin

Stub for cross object in environment.

__init__(config, opts)

Initialize StaticCrossONS class.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

_get_device(device_id)

Return device object by device id.

Parameters:device_id (str) – Device ID Raises:Exception – device is not present in related configurations Returns:Device object Return type:GenericEntry

check()

Mandatory method for environment specific switch classes.

class_logger = <logging.LoggerAdapter object>

cross_clear()

Mandatory method for environment specific switch classes.

cross_connect(conn_list=None)

Configure Cross connect.

Parameters:conn_list (list[list]) – List of connections Raises:CrossException – devices from conn_list are not in related configurations

cross_disconnect(disconn_list=None)

Configure Cross disconnect.

Parameters:disconn_list (list[list]) – List of connections

start()

Mandatory method for environment specific switch classes.

stop()

Mandatory method for environment specific switch classes.

xconnect(conn=None)

We have to connect only dest port as far as it is only emulation.

Parameters:conn (list) – Connection info in format [sw1, port1, sw2, port2]

xdisconnect(conn=None)

We have to disconnect only dest port as far as it is only emulation.

Parameters:conn (list) – Connection info in format [sw1, port1, sw2, port2]

taf.testlib.dev_switch_lxc

taf.testlib.dev_switch_sdv_100g_rr

taf.testlib.dev_switch_seacliff

taf.testlib.dev_trextg

dev_trextg.py

TRex traffic generators specific functionality

Notes

To install TRex client api package:

1. Download package from http://trex-tgn.cisco.com/trex/release/
2. Unpack main package ‘v2.00.tar.gz’ and then client package ‘trex_client_v2.00.tar.gz’
3. Add path to the trex client stateless lib for PYTHONPATH environment variable: env PYTHONPATH=<your path>/trex_client/stl

class taf.testlib.dev_trextg.Trex(config, opts)

Bases: taf.testlib.TRex.Trex.TrexMixin, taf.testlib.TRex.TrexHLT.TrexHLTMixin, taf.testlib.packet_processor.PacketProcessor, taf.testlib.tg_template.GenericTG

TRex interaction base class.

Configuration examples:

TRex server Example:

{
 "name": "TRex"
 "entry_type": "tg",
 "instance_type": "trex",
 "id": "TG1",
 "ports": [0, 1],
 "ipaddr": "1.1.1.1",
 "ssh_user": "user",
 "ssh_pass": "PassworD",
}

Where:

 entry_type and  instance_type are mandatory values and cannot be changed

 id - int or str uniq device ID (mandatory)

 name - User defined device name (optional)

 ports or  port_list - short or long ports configuration (Only one of them has to be used)

 ipaddr - remote host IP address (mandatory)

 ssh_user - remote host login user (mandatory)

 ssh_pass - remote host login password (mandatory)

Notes:

You can safely add additional custom attributes.

__init__(config, opts)

Initializes connection to TRex.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

_get_speed_ports()

Get ports with speed from config.

Returns:Tuple with list of ports used in real config and list of port/speed values Return type:tuple(list[tuple], list[tuple, int])

Notes

This function check if port has speed in config file.

check()

Checking connection to TRex.

Returns:None

class_logger = <logging.LoggerAdapter object>

cleanup(mode='complete')

This method should do TRex ports cleanup (remove streams etc.)

Parameters:mode (str) – “fast” or “complete”. If mode == “fast”, method does not clear streams on the port (string)

create()

Obligatory class for entry_type = tg.

destroy()

Obligatory class for entry_type = tg.

get_os_mtu(iface=None)

Get MTU value in host OS.

Parameters:iface (str) – Interface for getting MTU in host OS Returns:Original MTU value Return type:int

Examples:

env.tg[1].get_os_mtu(iface=ports[('tg1', 'sw1')][1])

sanitize()

Clear ownership before exit.

set_os_mtu(iface=None, mtu=None)

Set MTU value in host OS.

Parameters:

• iface (str) – Interface for changing MTU in host OS
• mtu (int) – New MTU value

Returns:

Original MTU value

Return type:

int

Examples:

env.tg[1].set_os_mtu(iface=ports[('tg1', 'sw1')][1], mtu=1650)

start(wait_on=True)

Start Trex TG.

Parameters:wait_on (bool) – Wait for device is loaded

stop()

Shutdown TRex TG device.

taf.testlib.dev_vethcross

dev_vethcross.py

Cross connection based on creating Virtual Ethernet devices

class taf.testlib.dev_vethcross.VethCross(config, opts)

Bases: taf.testlib.dev_basecross.GenericXConnectMixin

Xconnect based on creating virtual ethernet interfaces.

Notes

It is used for simulated environment.

__init__(config, opts)

Initialize VethCross class.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

_exec_x_command(connection, action)

Create single connection.

Parameters:

• connection (list) – Connection info in format [sw1, port1, sw2, port2]
• action (str) – Action for connection.

Raises:

CrossException – incorrect action type, error on connection

class_logger = <logging.LoggerAdapter object>

cross_connect(conn_list=None)

Create connections.

Parameters:conn_list (list[list]) – List of connections

cross_disconnect(disconn_list=None)

Destroy connections.

Parameters:disconn_list (list[list]) – List of connections

generic_generic(connection, action)

Returns set of commands to create connection between 2 default NNS.

This is used to create just veth interfaces without netns.

Parameters:

• connection (list) – Connection info in format [sw1, port1, sw2, port2]
• action (str) – Action for connection.

generic_netns(connection, action)

Returns set of commands to create/destroy connection between default NNS and custom NNS.

Parameters:

• connection (list) – Connection info in format [sw1, port1, sw2, port2]
• action (str) – Action for connection.

get_name_port(dev_id, port_id)

Get port name.

Parameters:

• dev_id (int) – Device ID
• port_id (int) – Port ID

Returns:

Device name, Port name

Return type:

tuple

netns_netns(connection, action)

Returns set of commands to create/destroy connection between 2 Linux Network Namespaces.

Parameters:

• connection (list) – Connection info in format [sw1, port1, sw2, port2]
• action (str) – Action for connection.

xconnect(connection)

Create single connection.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

xdisconnect(connection)

Destroy single connection.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

taf.testlib.dev_vlabcross

dev_vlabcross.py`

ONS Vlab virtual cross specific functionality

class taf.testlib.dev_vlabcross.VlabCross(config, opts)

Bases: taf.testlib.dev_vlabcross.VlabEnv

Vlab from xconnect viewpoint.

_get_ports_from_config(connection=None)

Get ports from configuration.

Parameters:

connection (list) – Connection info in format [sw1, port1, sw2, port2]

Raises:

• CrossException – unsupported connection type
• ValueError – error in configuration file

Returns:

Ports from configuration

Return type:

list

class_logger = <logging.LoggerAdapter object>

cross_clear()

Clear all connections between switches

Raises:CrossException – error on connections clearing Returns:True if success or False if all connections were not cleared. Return type:bool

Examples:

cross_clear(env)

cross_connect(conn_list)

Make connections between switches.

Parameters:conn_list (list[list]) – Set of connections in format: [[sw1, port1, sw2, port2], … ] Raises:CrossException – devices from conn_list are not in related configurations, error on connection creation Returns:True if success or raise an error if connections were not created. Return type:bool

Examples:

cross_connect([[0, 1, 1, 1], [0, 2, 1, 2]])

cross_disconnect(disconn_list)

Destroy connections between switches.

Parameters:disconn_list (list[list]) – Set of connections in format: [[sw1, port1, sw2, port2], … ] Raises:CrossException – error on connection destroying Returns:True if success or False if connections were not destroyed. Return type:bool

Examples:

cross_disconnect([[0, 1, 1, 1], [0, 2, 1, 2]])

xconnect(connection)

Create single connection.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

xdisconnect(connection)

Destroy single connection.

Parameters:connection (list) – Connection info in format [sw1, port1, sw2, port2]

class taf.testlib.dev_vlabcross.VlabEnv(config, opts)

Bases: taf.testlib.dev_basecross.GenericXConnectMixin

Vlab from device viewpoint.

DEFAULT_TIMEOUT = 1

__init__(config, opts)

Initialize VlabEnv class.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

Raises:

CrossException – error in vlab path

check()

Checking Vlab instance status.

class_logger = <logging.LoggerAdapter object>

probe()

Check if Vlab instance is run.

Returns:Vlab status Return type:dict

probe_port()

Establishing a connection to a remote host.

Returns:True if connection is established Return type:bool

restart()

Restarting Vlab instance.

start()

Starts vlab based on provided host and port info with specified number of interfaces.

Raises:

• CrossException – not local environment, vlab is stopped
• Exception – error on vlab start

stop()

Stops vlab based on provided host and port info.

Raises:CrossException – error on vlab stop

waitoff(timeout=30)

Waiting until Vlab port is down.

Parameters:timeout (int) – Waiting timeout Raises:CrossException – error on vlab stop Returns:Vlab status Return type:dict

waiton(timeout=30)

Waiting until Vlab port is up.

Parameters:timeout (int) – Waiting timeout Raises:CrossException – error on vlab start Returns:Vlab status Return type:dict

taf.testlib.entry_template

entry_template.py

Interface class for any TAF environment configuration entry

class taf.testlib.entry_template.GenericEntry(config, opts)

Bases: object

Interface class for any TAF environment configuration entry.

__init__(config, opts)

Entry __init__ method has to take 2 parameters.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

check()

Called in environment check method. Used before/after TC to verify that environment is ready to use.

cleanup()

Called in environment cleanup method. It’s used to delete all runtime configuration and return entry to default state.

create()

Called on environment initialize step. It should contain any steps necessary to start using device instance.

destroy()

Called on environment shutdown step. It should contain any necessary steps to release used device.

get_env_prop(param)

related_conf = None

related_obj = None

sanitize()

Called in environment sanitize method. It’s used by pytest.softexit and should contain any steps to release possible locks.

start()

This method could be called by create method and has to prepare entry object for processing commands.

E.g. It can power on device, or launch some service on it. You can use create method itself for this. But usually create method checks some CLI options (–get_only) and decides to invoke start or not.

stop()

This method could be called by destroy method and has to release entry object configuration.

E.g. It can power off device, or stop some service on it. You can use destroy method itself for this. But usually destroy method checks some CLI options (–get_only, –leave_on) and decides to invoke stop or not.

taf.testlib.environment

environment.py

Environment-specific functionality

taf.testlib.environment.get_absolute_build_path(build_path=None)

Return absolute path to switchpp binaries.

Parameters:build_path (str) – path to switchpp binaries . Returns:absolute path to switchpp binaries if one exists or None if not Return type:str

Examples:

get_absolute_build_path(build_path="/some/path/switchpp")

taf.testlib.environment.get_conf_file(conf_name=None, conf_type='env')

Return full path to conf file.

Parameters:

• conf_name (str) – path to config file in json format.
• conf_type (str) – type of config: “env” - environment, “setup” - used setup. This value will be added to path as a last directory name.

Raises:

CustomException – conf_name is None

Raturns:

str: absolute path to configuration file if one exists or None if else.

Notes

Discovery order:

1. search if current folder;
2. search in /usr/local/bin/taf/<conf_type>/;
3. search in /etc/taf/<conf_type>/.

Examples:

get_conf_file(conf_name="simplified_setup.json", conf_type='cross')

taf.testlib.fixtures

taf.testlib.helpers

helpers.py

Helpers functions

taf.testlib.helpers.apply_action_and_add_finalizer(request, targets, action, reaction)

taf.testlib.helpers.change_stp_state(switche_instance, state)

Disables/Enables STP state on per-port basis.

Notes

Very useful to prevent packet storm caused by global STP disabling.

Parameters:

• switche_instance (object) – Switche xmlrpc.ServerProxy object (object)
• state (str) – Desired STP state change action (‘Enabled’ or ‘Disabled’)

Returns:

None

taf.testlib.helpers.clear_table(switch_instance=None, table_name=None)

Clear table for correct test cases work.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• table_name (str) – Specific table name

Returns:

None

Examples:

clear_table(switch_instance=env.switch[1], table_name="ACLRules")

taf.testlib.helpers.designated_port(priority, port_id)

Function for return hex string value that is sum of bridgePriority and port_id in hex format.

Parameters:

• priority (int) – Port bridgePriority
• port_id (int) – Port ID

Returns:

Hex Value without 0x prefix

Examples:

helpers.designated_port(32768, 28)

taf.testlib.helpers.disable_lldp_for_all_ports(switch_instance)

Disable Lldp on all device ports.

Parameters:switch_instance (object) – Switch instance to work with Returns:None

Examples:

helpers.disable_lldp_for_all_ports(env.switch[1])

taf.testlib.helpers.generate_random_ip(ip_exclusions=None, prefix=None, quantity=1)

Generate list that contains randomly generated IP addresses that starts from prefix.

Parameters:

• ip_exclusions (list) – list if IP addresses that should be excluded from generated list, for example Route Interface address used on device
• prefix (str) – Network part of IP address
• quantity (int) – quantity of ip addresses that should be generated

Returns:

ip_pool list of generated MAC addresses

Return type:

list

Examples:

source_ips = helpers.generate_random_ip(ip_exclusions = router_ips, prefix = "10.0.1", quantity = streams_quantity)

taf.testlib.helpers.generate_random_mac(mac_exclusions=None, quantity=1)

Generate list that contains randomly generated MAC addresses that starts from 00.

Parameters:

• mac_exclusions (list) – list if MAC addresses that should be excluded from generated list
• quantity (int) – quantity of mac addresses that should be generated (int)

Returns:

mac_pool list of generated MAC addresses

Return type:

list

Examples:

source_macs = helpers.generate_random_mac(mac_exclusions = ["01:00:02:00:00:01", ], quantity = 10)

taf.testlib.helpers.get_attribute_from_argvalue(argvalue, item)

Gets an attribute from argvalues.

Parameters:

item (int | str) – This can be an int (it will attempt to retrieve an index or a str (it will attempt to retrieve a NamedTuple’s field name)

Raises:

• AttributeError – when attribute not found
• IndexError – when index not found

Returns:

str

taf.testlib.helpers.get_json(path, filename)

Get json file.

taf.testlib.helpers.get_packet_from_the_port(sniff_port=None, params=None, sniff_data=None, tg=None)

Return packet from the sniffer data according to search criteria.

Parameters:

• sniff_port (list) – port where packet was sniffed
• params (list) – params list for packet search
• sniff_data – sniffed data
• tg (object) – traffic generator

Returns:

packet or None

Examples:

helpers.get_packet_from_the_port('vlab0', ({'layer': "Ether", 'field': "src", 'value': "00:00:00:01:02:03".lower()},), data, env.tg[1])

taf.testlib.helpers.get_stepped_value(value, step, step_type='Down')

Returns the rounded value, given an initial input value.

Parameters:

• value (int) – The input value
• step (int | OrderedDict) – The step value
• step_type (str) – Whether the value is incremented to the next step

Raises:

ValueError –

Returns:

int

taf.testlib.helpers.group_get(match, *args, **kwargs)

Parameters:match (_sre.SRE_Match) –

taf.testlib.helpers.grouper(iterable, n)

Collect data into fixed-length chunks or blocks

grouper(‘ABCDEFG’, 3) –> ABC DEF G

Modified from http://stackoverflow.com/users/1052325/reclosedev ‘s modification of http://stackoverflow.com/users/279627/sven-marnach ‘s answer

http://stackoverflow.com/a/8998040

Parameters:

• iterable (iter()) – iterable to group into chunks
• n (int) – chunk size

Returns:

itererable of chunks of size n

Return type:

iter(list)

taf.testlib.helpers.grouper_it(iterable, n)

Collect data into iterables of fixed-length chunks or blocks

grouper_it(‘ABCDEFG’, 3) –> iter(ABC) iter(DEF) iter(G)

Copyied from http://stackoverflow.com/users/1052325/reclosedev ‘s modification of http://stackoverflow.com/users/279627/sven-marnach ‘s answer

http://stackoverflow.com/a/8998040

Parameters:

• iterable (iter()) – iterable to group into chunks
• n (int) – chunk size

Returns:

return an iterable of iterables of chunk size n

Return type:

iter(iter())

taf.testlib.helpers.is_double_tag_packet_received(iface=None, destination_mac=None, eth_type=37120, prio_1=6, prio_2=None, vlan_1=None, vlan_2=None, result=True, type_1=None, type_2=None, tg_instance=None)

Check if proper packet is received.

Parameters:

• iface (str) – Interface
• destination_mac (str) – Destination MAC address
• eth_type (hex) – Ether.type
• prio_1 (int) – Dot1Q.prio of outer layer
• prio_2 (int) – Dot1Q.prio of inner layer
• vlan_1 (int) – Dot1Q.vlan of outer layer
• vlan_2 (int) – Dot1Q.vlan of inner layer
• type_1 (int) – Dot1Q.type of outer layer
• type_2 (int) – Dot1Q.type of inner layer
• result (bool) – Expected result: true or false
• tg_instance (str) – TG instance

Returns:

True or raise exception

Examples:

self._check_if_double_tag_packet_received(iface=data[sniff_ports[3]],destination_mac=self.destination_mac, vlan_1=self.vlan_id_20,
                                          vlan_2=self.vlan_id_10, type_1=0x8100, type_2=0x800, tg_instance=env.tg[1])

taf.testlib.helpers.is_entry_added_to_acl_actions_table(action=None, param=None, switch_instance=None)

Check entry in ACL Actions Table.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• action (str) – The actual action
• param (str) – Parameters required for the action

Returns:

True or False

Return type:

bool

Examples:

is_entry_added_to_acl_actions_table(action=action_acl, param=param_acl, switch_instance=env.switch[1].xmlproxy)

taf.testlib.helpers.is_entry_added_to_acl_expressions_table(field=None, mask=None, data=None, switch_instance=None)

Check entry in ACL Expressions Table.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with (xmlrpclib.ServerProxy object)
• field (str) – The field to operate on
• mask (str) – The bitwise mask to AND with the field
• data (str) – The expected result

Returns:

True or False

Return type:

bool

Examples:

is_entry_added_to_acl_expressions_table(field=field, mask=mask_acl, data=data_acl, switch_instance=env.switch[1].xmlproxy)

taf.testlib.helpers.is_entry_added_to_fdb(mac_address=None, port_id=1, vlan_id=1, switch_instance=None)

Check dynamic entry in FDB.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• mac_address (str) – Source mac-address to check entry in FDB
• port_id (int) – Port number from where packet was sent
• vlan_id (int) – Vlan number from where packet was sent

Returns:

True or False

Return type:

bool

Examples:

is_entry_added_to_fdb(mac_address=source_mac, port_id=1, vlan_id=vlab_id, switch_instance=env.switch[1].xmlproxy)

taf.testlib.helpers.is_entry_added_to_rules_table(rule_id=1, expression_id=1, action_id=1, stage=None, enabled='Disabled', priority=1, switch_instance=None)

Check entry in ACL Rules Table.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• rule_id (int) – Given ruleId from the ACLRules table
• expression_id (int) – Given expressionId from the ACLExpressions table
• action_id (int) – Given actionId from the ACLActions table
• stage (str) – “Ingress”, “Egress”, “Lookup”
• enabled (str) – Enable or disable rule
• priority (int) – Rules priority

Returns:

True or False

Return type:

bool

Examples:

is_entry_added_to_rules_table(rule_id=rule_id, expression_id=expression_id, action_id=action_id,
                              stage=rules_stage_ingress, enabled=rules_disabled,
                              priority=priority_zero, switch_instance=env.switch[1].xmlproxy)

taf.testlib.helpers.is_entry_added_to_table(switch_inst, table_name, srch_params, timeout=1, expect_count=1, exist=True)

Check if entry with specified fields values is added to table

Parameters:

• switch_inst (object) – Switch instance to work with
• table_name (str) – The name of table to perform search in
• srch_params (dict) – Parameters for entry search
• timeout (int) – Time to wait
• expect_count (int) – Number of entries in the table that correspond to search parameters (integer)
• exist (bool) – Decided if entry exists in table or not (boolean)

Returns:

True/False

Return type:

bool

Examples:

srch_params = {'macAddress': '00:00:00:BB:00:AA', 'portId': 24, 'valid': 'Enabled'}
assert helpers.is_entry_added_to_table(env.switch[1], "DcbxRemotes", srch_params, timeout=5)
assert helpers.is_entry_added_to_table(env.switch[1], "DcbxRemotes", srch_params, timeout=5, exist=False)

taf.testlib.helpers.is_error_in_expected(exception)

Check if exception is in list of expected errors.

taf.testlib.helpers.is_packet_received(data=None, iface_1=None, iface_2=None, layer_1='Ether', field_1='dst', value_1=None, layer_2='IP', field_2='dst', value_2=None, tg_instance=None, result=True, lag_available=False, f_result=False)

Checking if packet is received.

Parameters:

• data (str) – Captured data
• iface_1 (str) – Interface
• iface_2 (str) – Interface
• layer_1 (str) – Layer to analyze
• field_1 (str) – Field to look for
• value_1 (can be different) – Comparing value
• layer_2 (str) – Layer to analyze
• field_2 (str) – Field to look for
• value_2 (can be different) – Comparing value
• tg_instance (str) – TG instance
• result (bool) – Expected result: true or false
• lag_available (bool) – is ports are in LAG: true or false
• f_result (bool) – Flag to

Returns:

True or raise exception

Examples:

helpers.is_packet_received(data=data, iface_1=sniff_ports[1], iface_2=sniff_ports[0], value_1="ff:ff:ff:ff:ff:ff",
                           layer_2="ARP", field_2="pdst", value_2="10.0.31.2",
                           tg_instance=env.tg[1], lag_available=True)

taf.testlib.helpers.is_row_added_to_arp_table(timeout=60, switch_instance=None, net_address=None, if_id=None, result=True)

Wait until proper row will be added to ARP table.

taf.testlib.helpers.is_row_added_to_l2multicast_table(mac_address=None, port_id=None, vlan_id=1, switch_instance=None, result=True)

Check if row with specified parameters added to L2Multicast table.

taf.testlib.helpers.is_static_entry_added(mac_address=None, vlan_id=1, port_id=1, switch_instance=None)

Check static entry in StaticMAC table

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• mac_address (str) – Source mac-address to check entry in FDB
• port_id (int) – Port number from where packet was sent
• vlan_id (int) – Vlan number from where packet was sent

Returns:

True or False

Return type:

bool

Examples:

is_static_entry_added(mac_address=source_mac, port_id=1, vlan_id=vlab_id, switch_instance=env.switch[1].xmlproxy)

taf.testlib.helpers.merge_dicts(*dict_args)

Merge dictionaries from dict_args.

When same keys present the last dictionary in args list has the highest priority.

Parameters:dict_args (tuple(dict)) – Returns:merged dictionary Return type:dict

taf.testlib.helpers.pass_xmlrpc_fault(func)

Decorator to create raises functions with predefined xmlrpclib. Fault exception.

taf.testlib.helpers.print_sniffed_data_brief(sniffer_data)

Print sniffed packets with sniffed count.

Parameters:sniffer_data – sniffed data

Examples:

helpers.print_sniffed_data_brief(data)

taf.testlib.helpers.process_cli_results(return_list)

Returns list of errors in CLI set results.

taf.testlib.helpers.process_multicall(return_list)

Returns list of methods with errors.

taf.testlib.helpers.raises(expected_exception, fail_message, *args, **kwargs)

Assert that a code block/function call raises ExpectedException and raise a failure exception otherwise.

Notes

You can specify a callable by passing a to-be-called lambda or an arbitrary callable with arguments.

Parameters:

• expected_exception (Exception) – Expected exception type
• fail_message (str) – Fail message in case exception doesn’t raise (string)
• kwargs (args,) – function to test as first argument and it’s parameters

Returns:

Exception info or pytest.fail

Examples:

helpers.raises(ZeroDivisionError, "Expected ZeroDevision didn't raise.", lambda: 1/0)

def f(x): return 1/x
...
helpers.raises(ZeroDivisionError, "Expected exception didn't raise.", f, 0)
helpers.raises(ZeroDivisionError, "Did not raise", f, x=0)

taf.testlib.helpers.realswitch_only(x='Real switch only test case.')

taf.testlib.helpers.ri_find_wrapper(switch_instance, vlan=None, ip_address=None, bandwith=100, mtu=1500, vrf=0)

Wrapper of “find” function for RouteInterface table.

Notes

This is temporary function.

Parameters:

• switch_instance (object) – Switch instance
• vlan (str) – Vlan on which route interface is implemented.
• ip_address (str) – IP address of route interface.
• mtu (int) – MTU of route interface.
• bandwith (int) – Bandwith of route interface
• vrf (int) – Virtual route Id

Returns:

row

Examples:

_ri_find_wrapper(env.switch[1], vlan=10, ip_address="10.0.10.1/24", mtu=100, bandwith=1500, vrf=0)

taf.testlib.helpers.run_on_ixnetwork(y='Run on IxNetwork setup only')

taf.testlib.helpers.run_on_platforms(x, y='This test is skipped on current platform.')

taf.testlib.helpers.run_on_tg(x, y='This test is skipped on current platform.')

taf.testlib.helpers.run_on_ui(x, y='Skip with current UI')

taf.testlib.helpers.send_receive_packets(packet_definition=None, count=5, src_port=None, dst_port=None, sniff_time=5, sniff_filter='', tg=None, expect_rcv=True)

Send and verify receiving of packets.

taf.testlib.helpers.set_admin_mode_for_slave_ports(switch_instance=None, admin_mode='Down')

Set adminMode for logical ports.

Parameters:

• switch_instance (object) – Switch class instance to work with
• admin_mode (str) – Ports adminMode

Returns:

True or raise exception

Examples:

assert helpers.set_admin_mode_for_slave_ports(switch_instance=env.switch[1])

taf.testlib.helpers.set_all_ports_admin_disabled(switches, wait_status=True)

Sets all ports of all switches provided as an argument to admin state Down.

Parameters:

• switches (dict) – Switches configuration
• wait_status (bool) – Wait for ports desire necessary state

Returns:

none

Examples:

set_all_ports_admin_disabled(env.switch)

taf.testlib.helpers.set_equal_port_speed_and_cost(ports, switch, table, switch_id=None, cost=2000, speed=10000)

Sets equal ports cost and speed for switch ports, and disables slave 40G ports

Parameters:

• ports (dict) – ports dictionary
• switch (object) – switch instance
• switch_id (int) – id of switch in environment
• speed (int) – port speed value, integer
• table – table name or list of table names (f.e. “MSTPPorts” or [“RSTPPorts”, “MSTPPorts”])
• cost (int) – port cost value, integer

Returns:

None

Examples:

helpers.set_equal_port_speed_and_cost(ports, env.switch, ["MSTPPorts", "RSTPPorts", "RSTPPorts"])
helpers.set_equal_port_speed_and_cost(ports, env.switch, "RSTPPorts")
helpers.set_equal_port_speed_and_cost(ports, env.switch[1], "RSTPPorts", switch_id=1]

taf.testlib.helpers.set_equal_ports_costs(ports, switch, switch_id, table, cost=2000)

Sets equal port costs for switch ports.

Parameters:

• ports (dict of dict) – ports dictionary
• switch (SwitchX object) – switch instance
• switch_id (int) – id of switch in environment
• table (str) – table name (e.g. “MSTPPorts”)
• cost (int) – port cost value

Returns:

None

taf.testlib.helpers.set_equal_ports_speed(ports, switch, switch_id, speed=10000)

Sets equal port costs for switch ports.

Parameters:

• ports (dict) – ports dictionary
• switch (object) – switch instance
• switch_id (int) – id of switch in environment
• speed (int) – port speed value, integer

Returns:

None

taf.testlib.helpers.set_invalid_value(switch_instance=None, method=None, params=None)

Exception handler for negative testcases.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• method (str) – XML RPC method to invoke
• params (list of mixed types values) – Arguments to pass to XML RPC method

Returns:

None

Examples:

set_invalid_value(switch_instance=env.switch[1].xmlproxy, method="nb.Table.set.field", params=[param1, param2,param3])

taf.testlib.helpers.set_lldp_ports_admin_status(switch_instance, port_list, status='Disabled')

Set LLDP port admin status on ports.

Args

switch_instance(object): the switch instance. port_list(list): the list of ports. status(str): the LLDP port admin status.

Returns:True or False Return type:bool

taf.testlib.helpers.set_mstp_mode(switch_instance=None)

Obsoleted function. Use set_stp_mode with proper option instead.

taf.testlib.helpers.set_ports_admin_enabled(switches, ports, wait_status=True, fail_func=None)

Sets provided in arguments ports to admin state Up.

Parameters:

• switches (dict) – Switches configuration
• ports (dict(dict)) – List of ports to enable. Each element is a dictionary like {(‘sw1’, ‘sw2’): {link_id: port_id})}
• wait_status (bool) – Wait for ports desire necessary state
• fail_func (tuple) – function with params that will be executed in case of failure

Returns:

none

Examples:

ports = {('sw1', 'sw2'): {1: 2, 2: 4}}
set_ports_admin_enabled(env.switch, ports)
set_ports_admin_enabled(env.switch, ports, fail_func=(pytest.softexit, [mesasge, env]))

taf.testlib.helpers.set_rstp_mode(switch_instance=None)

Obsoleted function. Use set_stp_mode with proper option instead.

taf.testlib.helpers.set_stp_mode(switch_instance=None, mode='STP')

Set SpanningTree mode to STP|RSTP|MSTP.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch class instance to work with
• mode (str) – SpanningTree mode

Returns:

None

Examples:

set_stp_mode(env.switch[1], "MSTP")

taf.testlib.helpers.simswitch_only(x='SimSwitch only test case.')

taf.testlib.helpers.skip_on_platforms(x, y='This test is skipped on current platform.')

taf.testlib.helpers.skip_on_tg(x, y='This test is skipped on current platform.')

taf.testlib.helpers.skip_on_ui(x, y='Skip with current UI')

taf.testlib.helpers.skiptest(x)

taf.testlib.helpers.stat_counters_clear(switch_instances=None, ports=None, counter_name='EtherStatsPkts')

Clear all statistic counters on defined port.

Notes

DO NOT USE ON CONTINUOUSLY RUNNING TRAFFIC - FUNCTION IS WAITING THAT COUNTER BECOME 0

Parameters:

• switch_instances (list) – list of switch instances from env
• ports (list) – list of ports where statistic should be cleared
• counter_name (str) – counter that checked to become 0 after clear action. “EtherStatsPkts” used by default.

Returns:

None

Examples:

helpers.stat_counters_clear(switch_instances=[env.switch[2], ], ports=ports["sw2", "sw1"].values(), counter_name="EtherStatsPkts256to511Octets")

taf.testlib.helpers.stat_counters_read(switch_instances=None, ports=None, counter_name='EtherStatsPkts', previous_results=None, expected_results=None, timeout=30)

Read statistic counter on defined port on defined switch.

In addition if previous_results defined then sum of read results compared with sum of previous results. If expected results defined then sum of current results compared with sum of expected results and if match occurred then return results In case expected results is not defined then sum of currently read results == sum of previous results counters read again until timeout

Parameters:

• switch_instances (list) – list of switch instances from env
• ports (list) – list of ports where statistic should be read
• counter_name (str) – counter that should be read
• previous_results (list) – list with previous statistics results.
• expected_results (list) – list with expected counters values from ports
• timeout (int) – timeout for re-reading counters in case sum of previous results matchedwith current results, default = 30

Returns:

list with read counters from ports

Return type:

list

Examples:

sw2_statistics = helpers.stat_counters_read(switch_instances=[env.switch[2], ], ports=ports["sw2","sw1"].values(),
                                            counter_name="EtherStatsPkts256to511Octets", previous_results=previous_sw2_stats, timeout=30)

taf.testlib.helpers.update_table_params(switch_inst, table_name, params, find_params=None, row_id=None, validate_updates=True)

Configure port parameters in table.

Parameters:

• switch_inst (object) – Switch instance to work with
• table_name (str) – The name of table to work with
• params (dict) – Parameters and values that should be configured for port
• find_params (list) – List of parameters to find a row in table
• row_id (int) – Row ID in table
• validate_updates (bool) – Verify if updates were set

Returns:

None

Examples:

helpers.update_table_params(env.switch[1], "DcbxPfcPortsAdmin", {"willing": "Enabled", "enabled": "0,0,0,1,0,0,0,0"}, [port_id, ])
helpers.update_table_params(env.switch[1], "DcbxPorts", {"adminStatus": 'Disabled'}, row_id=24)

taf.testlib.helpers.validate_frame_against_multiple_layers(packet, src_mac=None, unexpected_layers=None, expected_layers=None)

Verify that frame doesn’t contain specific set of Layers and contains correct values for other expected Layers.

Parameters:

• packet (pypacker packet) – Packet that should be validated
• src_mac (None or MacAddress) – Expected src mac address of packet
• unexpected_layers (list of strings) – List of layers that should not be present is the frame
• expected_layers (list or dict) – List or Dictionary of layers with expected values that should be present in frame. If the frame contains multiple number of specific layer than value in expected_layer dictionary should be a list (list or dictionary)

Returns:

None (Raises error if packet is incorrect)

Examples:

expected_layers = {"DCBXApplicationPriority": {"type": 127L, "length": 5L, "oui": 0x80c2, "subtype": 12L, "reserved": 0},
                   "DCBXApplicationPriorityTable": [{'priority': 7, 'protocolid': 125, 'sel': 2, 'reserved': 0L},
                                                    {'priority': 6, 'protocolid': 555, 'sel': 4, 'reserved': 0L}],
                   "LLDPDUEnd": {"type": 0, "length": 0}}
unexpected_layers = ["DCBXCongestionNotification", "DCBXConfiguration"]
helpers.validate_frame_against_multiple_layers(packet, dut_mac, unexpected_layers=unexpected_layers, expected_layers=expected_layers)

taf.testlib.helpers.verify_port_params_in_table(switch_inst, table_name, params, find_params=None, row_id=None, timeout=0, interval=0.25)

Verify that row has correct parameters’ values in table.

Parameters:

• switch_inst (object) – Switch instance to work with
• table_name (str) – The name of table to validate the row parameters
• params (dict) – Parameters and values the row in table should match
• find_params (list) – List of parameters to find a row in table
• row_id (int) – Row ID in table
• timeout (int) – Time to wait
• interval (int) – interval in seconds between re-reading the table

Returns:

None (Raises error if values are incorrect)

Examples:

helpers.verify_port_params_in_table(env.switch[1], "DcbxPfcPortsLocal", {"willing": "Enabled", "enabled": "0,0,0,0,0,0,0,1"}, [port_id, ], timeout=1)
helpers.verify_port_params_in_table(env.switch[1], "DcbxPorts", {"multiplePeers": 'Enabled'}, row_id=24, timeout=1)

taf.testlib.helpers.wait_for_route_iface_deleted(switch, iface_id, ip_address, bandwidth, mtu, vrf, timeout)

Wait for RouteInterface deleted if it is not after ‘timeout’ seconds elapsed

Parameters:

• switch (SwitchX object) – switch for checking RouteInterface being deleted
• iface_id (str) – interface id for checking RouteInterface being deleted
• ip_address (str) – ip address for checking RouteInterface being deleted
• bandwidth (int) – bandwidth for checking RouteInterface being deleted
• mtu (int) – mtu for checking RouteInterface being deleted
• vrf (int) – vrf for checking RouteInterface being deleted
• timeout (int) – seconds for checking RouteInterface being deleted

Returns:

None

Examples:

_wait_for_route_iface_deleted(env.switch[1], 3210, '2001:db8:85a3::8a3e:370:7377/96', 1000, 1280, 0, 120)

taf.testlib.helpers.wait_for_route_iface_status(switch, iface_id, timeout, status)

Wait for RouteInterface changed its oper status to expected value or raise exception if it is not after ‘interval’ seconds elapsed

Parameters:

• switch (SwitchX object) – switch for checking RouteInterface oper state value
• iface_id (str) – interface id for checking oper state value
• timeout (int) – seconds for checking RouteInterface oper state value
• status (str) – RouteInterface oper state value (lower)

Returns:

None

Examples:

_wait_for_route_iface_status(env.switch[1], 2, 120, 'down')

taf.testlib.helpers.wait_until_entry_is_expired(expected_timeout=1, switch_instance=None, table_name='L2Multicast')

Wait until entry is expired from table

Parameters:

• switch_instance (object) – Switch class instance to work with
• expected_timeout (int) – Time to wait
• table_name (str) – XML-RPC table name

Returns:

True or raise exception

Examples:

assert self.wait_until_entry_is_expired(timeout=10, switch_instance=env.switch[2])

taf.testlib.helpers.wait_until_fdb_entry_is_added(mac_address=None, port_id=1, timeout=5, vlan_id=1, switch_instance=None)

Wait some time for adding dynamic entry to FDB.

Parameters:

• switch_instance (xmlrpclib.ServerProxy object) – Switch instance to work with
• mac_address (str) – MAC address for check
• port_id (int) – Port ID corresponding to MAC address
• vlan_id (int) – VLAN ID corresponding to MAC address
• timeout (int) – seconds for checking Fdb entry being added

Returns:

None

Examples:

wait_time_for_adding_entry_to_fdb(mac_address=source_mac, port_id=port_id_2, timeout=10, vlan_id=vlan_id, switch_instance=env.switch[1])

taf.testlib.helpers.wait_until_ops_state(switch_instance=None, port=1, state='Up', timeout=30)

Obsoleted function. Use wait_until_value_is_changed with proper option instead.

taf.testlib.helpers.wait_until_row_is_added_to_routes_table(timeout=60, switch_instance=None, network_ip='10.0.2.0/24', nexthop=None)

Wait until proper row will be added to Route table.

Parameters:

• switch_instance (str) – Switch class instance to work with
• timeout (int) – Time to wait
• network_ip (str) – Network IP address
• nexthop (str) – Network IP

Returns:

Raise exception

Examples:

helpers._wait_until_row_is_added_to_routes_table(switch_instance=env.switch[1], nexthop="10.0.31.2")

taf.testlib.helpers.wait_until_stp_state(switch_instance=None, table='STPPorts', port=1, state='Disabled', timeout=30)

Obsoleted function. Use wait_until_value_is_changed with proper option instead.

taf.testlib.helpers.wait_until_value_is_changed(switch_instance=None, table_name=None, parameter_name=None, value=None, row_id=1, timeout=30, findlist=None)

Wait until value is changed in table

Parameters:

• switch_instance (object) – Switch class instance to work with
• table_name (str) – Specific getTable method
• parameter_name (str) – Parameter in table
• value (int) – Checking value
• row_id (int) – Row id in table
• timeout (int) – time to wait until value is changed
• findlist (list) – list of parameters to find a row in the given table

Returns:

True or False

Return type:

bool

Examples:

wait_until_value_is_changed(env.switch[1], "Ports2LagRemote", "partnerOperSystemPriority", 0, 1)
wait_until_value_is_changed(switch_instance=env.switch[1], table_name="Ports2LagRemote", parameter_name="partnerOperSystemPriority",
                            value=0, row_id=1, timeout=30, [1, 1])

taf.testlib.helpers.waiting_table_is_loaded(switch_instance, table_name, expected_table_length, watch_interval=1, timeout=120, deviation=None, direction=None, verbose=False)

Wait until any table become necessary size, for example it can ba Dynamic ARP table or Route table.

Use deviation parameter if expected_table_length is not strict and can be larger (“+”) then expected on deviation value or smaller (“-“) then expected on deviation value or near (None=+/-) expected on deviation value

Parameters:

• switch_instance (object) – switch instance from env
• table_name (str) – table name under test, example: “ARP”
• expected_table_length (int) – value table length should increase to
• watch_interval (int) – interval in seconds between re-reading table from switch instance
• timeout (int) – 120 - timeout in seconds for function execution
• deviation (int) – if expected_table_length is not strict value and some deviation in entries is acceptable
• direction (str) – “+”, “-” or None. In case None then assuming that deviation is bidirectional, both “+” and “-“
• verbose (bool) – can be True or False. In case True then monitoring table content will be displayed in debug log

Returns:

True/False

Return type:

bool

Examples:

helpers.waiting_table_is_loaded(env.switch[1], "Route", 21, watch_interval = 1, timeout = 120, deviation = 2, direction = "+")

taf.testlib.helpers.xmlrpc_raises(*args, **kwargs)

Decorator to create raises functions with predefined xmlrpclib.

taf.testlib.hub

hub.py

Hub-specific functionality

class taf.testlib.hub.VlabHub(config, opts)

Bases: object

Class for simulated hub by Vlab.

__init__(config, opts)

Initialize VlabHub class.

Parameters:

• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

Raises:

HubException – vlab is not present in configuration

check()

Checking simulated hub.

class_logger = <logging.LoggerAdapter object>

create()

Creating simulated hub.

destroy()

Stopping simulated hub.

sanitize()

Sanitizing simulated hub.

start()

Starting simulated hub.

stop()

Stopping simulated hub.

taf.testlib.lab

lab.py

General lab functionality

class taf.testlib.lab.ConsoleServer(host=None, port=23, username=None, password=None, page_break='--More--', prompt=None, pass_prompt='Password: ', sudo_prompt=None, timeout=10, login_prompt='login: ', page_break_lines=3, exit_cmd=None, quiet=False)

Bases: taf.testlib.lab.GeneralPortServer

Class responsible for console connection.

static __new__(cls, config)

Get PortServer class related to console type.

Raises:SwitchException – unsupported console type

telnet_connect(timeout=45, with_login=None, wait_login=0)

Perform telnet connection to the device.

Parameters:

• timeout (int) – time out to wait connection
• with_login (bool) – Perform login procedure or not. If param isn’t set try automatically determine login necessity. (True|False|None)
• wait_login (int) – time to wait login before sending <Enter>. <Enter> is necessary if login is already appeared.

Raises:

NotImplementedError – not implemented

class taf.testlib.lab.GeneralPortServer(host=None, port=23, username=None, password=None, page_break='--More--', prompt=None, pass_prompt='Password: ', sudo_prompt=None, timeout=10, login_prompt='login: ', page_break_lines=3, exit_cmd=None, quiet=False)

Bases: taf.testlib.clitelnet.TelnetCMD

General functionality for console connection.

class_logger = <logging.LoggerAdapter object>

close_serial()

Close telnet connection to switch.

get_serial(timeout=90, with_login=None, wait_login=0)

Connect to switch via serial.

Parameters:

• timeout (int) – time out to wait connection
• with_login (bool) – Perform login procedure or not. If param isn’t set try automatically determine login necessity. (True|False|None)
• wait_login (int) – time to wait login before sending <Enter>. <Enter> is necessary if login is already appiered.

Notes

Create(or check) class attribute telnet with active telnet connection to switch and do login.

initial_config(config)

Performs initial configuration.

Parameters:config (dict) – Configuration information.

telnet_connect(timeout=45, with_login=None, wait_login=0)

Perform telnet connection to the device.

Parameters:

• timeout (int) – time out to wait connection
• with_login (bool) – Perform login procedure or not. If param isn’t set try automatically determine login necessity. (True|False|None)
• wait_login (int) – time to wait login before sending <Enter>. <Enter> is necessary if login is already appeared.

Raises:

NotImplementedError – not implemented

class taf.testlib.lab.KVMServer(config)

Bases: taf.testlib.lab.ConsoleServer

KVM functionality.

kvm_connect(output)

Perform connection via KVM.

Parameters:output (list[str]) – KVM output Raises:SwitchException – error on connection

telnet_connect(timeout=45, with_login=None, wait_login=0)

Perform telnet connection to the device.

Parameters:

• timeout (int) – time out to wait connection
• with_login (bool) – Perform login procedure or not. If param isn’t set try automatically determine login necessity. (True|False|None)
• wait_login (int) – time to wait login before sending <Enter>. <Enter> is necessary if login is already appeared.

Raises:

NotImplementedError – not implemented

class taf.testlib.lab.PortServer(config)

Bases: taf.testlib.lab.ConsoleServer

PortServer functionality.

release_serial()

Release serial port on port server.

Raises:SwitchException – error on terminating

Notes

This method kill tty on power server.

telnet_connect(timeout=45, with_login=None, wait_login=0)

Perform telnet connection to the device.

Parameters:

• timeout (int) – time out to wait connection
• with_login (bool) – Perform login procedure or not. If param isn’t set try automatically determine login necessity. (True|False|None)
• wait_login (int) – time to wait login before sending <Enter>. <Enter> is necessary if login is already appeared.

Raises:

NotImplementedError – not implemented

taf.testlib.linux_app_host

taf.testlib.linux_host_bash

linux_host_bash.py

Linux Host UI wrappers Implementation

class taf.testlib.linux_host_bash.InvalidPortContext(ui, ports)

Bases: object

Class to create a invalid port.

__enter__()

Returns:list of ports Return type:list

__exit__(exc_type, exc_val, exc_tb)

Deletes invalid port created.

__init__(ui, ports)

“Initialize Invalidport class.

Parameters:

• ui (UiOnpssShell) – instance of switch
• ports (iter()) – port id of invalid port

class taf.testlib.linux_host_bash.LinuxHostBash(host)

Bases: testlib.linux_host_interface.LinuxHostInterface

Class with UiOnpssShell wrappers.

BRIDGE_VLAN_COMMAND_STRING = 'bridge vlan {command} vid {vlan} dev {port} self {tagged}'

INDEX_NAME_RE = re.compile('(?P<index>\\d*):\\s(?P<name>\\w*)[@:]')

LLDP_ADMINSTATUS = {'Disabled': 'disabled', 'RxOnly': 'rx', 'TxAndRx': 'rxtx', 'TxOnly': 'tx'}

MAX_EXIT_STATUS = 256

MULTICALL_THRESHOLD = 100

RX_TX_RE = re.compile('\\s+[RT]X[^:]*:')

SSH_NO_EXIT_STATUS = -1

__init__(host)

Initialize UiOnpssShell class.

Parameters:host (SwitchGeneral) – Switch instance

_configure_fc_mode(ports=None, tx_mode='normal', traffic_class=255)

Determines 802.3x pause frame format used.

Maps priorities to traffic classes.

Parameters:

• ports (list[int]) – list of port ids
• tx_mode (str) – transmit mode (normal or class based)
• traffic_class (int) – traffic class bitmask

_disable_rx_fc(ports=None)

Disables receive of 802.3x pause frames.

Parameters:ports (list[int]) – list of port ids

_disable_tx_fc(ports=None)

Disables transmit of 802.3x pause frames per port.

Parameters:ports (list[int]) – list of port ids

_enable_rx_fc(ports=None, tc=1)

Enables receive of 802.3x pause frames.

Parameters:

• ports (list[int]) – list of port ids
• tc (int) – traffic class

_enable_tx_fc(ports=None)

Enables transmit of 802.3x pause frames per port.

Parameters:ports (list[int]) – list of port ids

classmethod _generate_bridge_vlan_commands(command, ports, vlans, tagged='')

Generate Bridge VLAN commands.

Parameters:

• command (str) – Bridge VLAN command
• ports (list[str]) – list of port IDs
• vlans (list[int]) – list of VLAN IDs
• tagged (str) – port tagging attribute

Returns:

list of Bridge VLAN commands

Return type:

list[str]

_parse_networkctl(res)

Parsing networkctl status output.

Parameters:res (str) – command output Returns:Returns networkctl status in dictionary format Return type:dict

_return_user_mode(results)

Maintained for abstraction compatibility.

Method that returns to user mode of a switch.

Parameters:results (list) – list of command execution results

check_device_state()

Attempts to connect to the shell retries number of times.

Raises:SwitchException – device is not ready

clear_config()

Clear device configuration.

clear_lldp_config()

clear_networkd_settings()

Clear networkd settings.

Raises:UiCmdException – when restart fails

clear_statistics()

Clear Statistics.

Returns:None

Examples

env.switch[1].ui.clear_statistics()

cli_get_all(commands, timeout=None, split_lines=True, expected_rcs=frozenset({0}), multicall_treshold=100)

Sends a list of commands, will return [‘’] if exception.

Parameters:

• commands (list[list[str]) – list of commands to be executed
• timeout (int) – command execution timeout
• split_lines (bool) – split command execution results by lines or not
• expected_rcs (int | set | list | frozenset) – expected return code
• multicall_treshold (int) – minimum number of commands to be executed using multicall

Returns:

list of outputs for each command

Return type:

list[list[str]]

cli_multicall(commands, timeout=None, expected_rcs=frozenset({0}))

Sends a list of commands.

Parameters:

• commands (list[str]) – list of commands to be executed
• timeout (int) – command execution timeout
• expected_rcs (int | set | list | frozenset) – expected return code

Returns:

list[tuple(str, tuple(str, str, int))]

Raises:

UICmdException – when rc not in expected_rcs

cli_send_command(command, timeout=None, expected_rcs=frozenset({0}))

Sends a single bash command.

Parameters:

• command (str) – command to be executed
• timeout (int) – command execution timeout
• expected_rcs (int | set | list | frozenset) – expected return code

Raises:

UIException – unexpected return code

Returns:

Returns CmdStatus namedtuple of stdout, stderr, return code

Return type:

tuple(str, str, int) | CmdStatus

cli_set(commands, timeout=None, split_lines=True, expected_rcs=frozenset({0}), multicall_treshold=100)

Sends a list of commands.

Will halt on exception from cli_send_command.

Parameters:

• commands (list[list[str]]) – list of commands to be executed
• timeout (int) – command execution timeout
• split_lines (bool) – split command execution results by lines or not
• expected_rcs (int | set | list | frozenset) – expected return code
• multicall_treshold (int) – minimum number of commands to be executed using multicall

Returns:

list of execution statuses for each command

Return type:

list[list[tuple(str | list, str, int)]]

Raises:

UICmdException – when rc not in expected_rcs

configure_application_priority_rules(ports, app_prio_rules)

Configure Application Priority rules.

Parameters:

• ports (list[int]) – list of ports
• app_prio_rules (list[dict]) – list of rules dictionaries

Returns:

None

Examples:

env.switch[1].ui.configure_application_priority_rules([1, 2], [{"selector": 1, "protocol": 2, "priority":1}, ])

Raises:SwitchException – not implemented

configure_arp(garp=None, refresh_period=None, delay=None, secure_mode=None, age_time=None, attemptes=None)

Configure ARPConfig table.

Parameters:

• garp (str) – AcceptGARP value. ‘True’|’False’
• refresh_period (int) – RefreshPeriod value
• delay (int) – RequestDelay value
• secure_mode (str) – SecureMode value. ‘True’|’False’
• age_time (int) – AgeTime value
• attemptes (int) – NumAttempts value

Returns:

None

Examples:

env.switch[1].ui.configure_arp(garp='Enabled')

Raises:SwitchException – not implemented

configure_dcbx_app(ports, **kwargs)

Configure DCBx APP parameter for the ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “willing”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_app([1, 2])

Raises:SwitchException – not implemented

configure_dcbx_cn(ports, **kwargs)

Configure DCBx CN parameter for the ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “cnpvSupported”; “cnpvReady”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_cn([1, 2], cnpvSupported='Enabled')

Raises:SwitchException – not implemented

configure_dcbx_ets(ports, **kwargs)

Configure DCBx ETS Conf/Reco parameter for ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “willing”; “cbs”; “maxTCs”; “confBandwidth”; “confPriorityAssignment”; “confAlgorithm”; “recoBandwidth”; “recoPriorityAssignment”; “recoAlgorithm”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_ets([1, 2], confBandwidth=100)

Raises:SwitchException – not implemented

configure_dcbx_pfc(ports, **kwargs)

Configure DCBx PFC parameter for the ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “mbc”; “enabled”; “willing”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_pfc([1, 2])

Raises:SwitchException – not implemented

configure_global_lldp_parameters(**kwargs)

Configure global LLDP parameters.

Parameters:**kwargs (dict) – parameters to be modified: ‘messageFastTx’; ‘messageTxHoldMultiplier’; ‘messageTxInterval’; ‘reinitDelay’; ‘txCreditMax’; ‘txFastInit’; ‘locChassisIdSubtype’. Returns:None

Examples:

env.switch[1].ui.configure_global_lldp_parameters(messageTxInterval=5)

Raises:SwitchException – not implemented

configure_lldp_ports(ports, **kwargs)

Configure LldpPorts records.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: ‘adminStatus’; ‘tlvManAddrTxEnable’; ‘tlvPortDescTxEnable’; ‘tlvSysCapTxEnable’; ‘tlvSysDescTxEnable’; ‘tlvSysNameTxEnable’.

Returns:

None

Examples:

env.switch[1].ui.configure_lldp_ports([1, 2], adminStatus='Disabled')

Raises:SwitchException – not implemented

configure_ovs_resources(**kwargs)

Configure OvsResources table.

Parameters:**kwargs (dict) – parameters to be configured: “controllerRateLimit”; “vlansLimit”; “untaggedVlan”; “rulesLimit”. Returns:None

Examples:

env.switch[1].ui.configure_ovs_resources(rulesLimit=2000)

Raises:SwitchException – not implemented

configure_tunneling_global(**kwargs)

Configure TunnelingGlobalAdmin table.

Parameters:**kwargs (dict) – parameters to be modified: “vnTag”; “vxlanInnerVlanProcessing”; “mode”, “vxlanDestUDPPort”. Returns:None

Examples:

env.switch[1].ui.configure_tunneling_global()

Raises:SwitchException – not implemented

connect()

Attempts to create a ssh session to the switch.

cpu_port = 0

create_arp(ip, mac, port)

Create StaticARP record.

Parameters:

• ip (str) – ARP ip address
• mac (str) – ARP mac address
• port (int) – port id

Returns:

None

create_dhcp_client_lease(file_name='/var/lib/dhcpd/dhcpd.leases')

Creates a lease file via SFTP.

Parameters:file_name (str) – File name Returns:None

Steps:

-# Create a lease file when DHCP Server is started

create_dhcp_config_file(lines, file_name='/etc/dhcp/dhcpd.conf')

Writes configuration required to file.

Parameters:

• lines (str) – Configuration
• file_name (str) – File name

Returns:

None

Steps:

-# Using sftp client open the config file in write mode
-# Write the minimum configuration required by DHCP Server to the file

create_dhcp_relay(iface_name='global', server_ip=None, fwd_iface_name=None)

Configure DhcpRelayAdmin or DhcpRelayV6Admin table.

Parameters:

• iface_name (str) – VLAN inteface name
• server_ip (str) – DHCP Server IP address
• fwd_iface_name (str) – VLAN forward interface name (for IPv6 config only)

Returns:

None

Examples:

env.switch[1].ui.create_dhcp_relay(iface_name='global', server_ip='10.10.0.2')

Raises:SwitchException – not implemented

create_folder(name, options=None, command=None, **kwargs)

Create folder.

Parameters:name (str) – folder name

create_invalid_ports(ports=None, num=1)

Creates port name if port id is passed say [Swop100, if 100 is passed as port id]

Else creates port name with a value incremented to 10 to existing length of ports Ex[sw0p34 , currently sw0p24 is last port]

Parameters:

• ports (iter()) – list of port_ids to generate port_names for
• num (int) – generate num new invalid ports

create_lag(lag=None, key=0, lag_type='Static', hash_mode='None')

Creates a lag group.

Parameters:

• lag (str|int) – lag ID|name
• key (int) – lag key
• lag_type (str) – lag type. Static | Dynamic
• hash_mode (str) – lag hash mode

Raises:

ExistsError –

Returns:

None

create_lag_ports(ports, lag, lag_mode='Passive', timeout='Long')

Set port to a LAG.

Parameters:

• ports (list[int]) – list of ports to be added into LAG
• lag (int) – LAG Id
• lag_mode (str) – mode of lag
• timeout (str) – LAG timeout

Returns:

lag

Return type:

int | str

Raises:

NotExistsError | AccessError –

create_match_api_rule(prio_id, handle_id, table_id, match_field_value_mask_list, action, action_value=None)

Set a rule into the table using the method defined in maa.py

Parameters:

• prio_id (int) – Higher id has a higher priority.
• handle_id (int) – handle for match.
• table_id (int) – the source table id where match to be set.
• match_field_value_mask_list (list[tuple(str, str, str)]) – field with match field, value and mask.
• action (str) – given action for source table
• action_value (int) – action value for a specified action

create_match_api_tcam_subtable(source_id, table_id, table_name, max_table_entries, match_field_type_pairs, actions)

Create a sub-table of tcam using the method defined in maa.py.

Parameters:

• source_id (int) – the source id in the tcam table.
• table_id (int) – a given table id. If switchd running, table id starts from 5 If matchd is running, table id starts from 4
• table_name (str) – a given table name.
• max_table_entries (int) – maximum number of rules can be set.
• match_field_type_pairs (list[tuple(str, str)]) – list of given match field with match type
• actions (list[str]) – list of actions for configurable matches

create_namespace(name)

Create network namespace.

Parameters:name (str) – netns name

create_ovs_bridge(bridge_name)

Create OvsBridges record.

Parameters:bridge_name (str) – OVS bridge name Returns:None

Examples:

env.switch[1].ui.create_ovs_bridge('spp0')

Raises:SwitchException – not implemented

create_ovs_bridge_controller(bridge_name, controller)

Create OvsControllers record.

Parameters:

• bridge_name (str) – OVS bridge name
• controller (str) – controller address

Returns:

None

Examples:

env.switch[1].ui.create_ovs_bridge_controller("spp0", "tcp:127.0.0.1:6633")

Raises:SwitchException – not implemented

create_ovs_flow_actions(bridge_id, table_id, flow_id, action, param, priority=2000)

Add row to OvsFlowActions table.

Parameters:

• bridge_id (int) – OVS bridge ID
• table_id (int) – Table ID
• flow_id (int) – Flow ID
• priority (int) – Rule priority
• action (str) – Action name
• param (str) – Action parameter

Returns:

None

Examples:

env.switch[1].ui.create_ovs_flow_actions(0, 0, 1, 'Output', '25')

Raises:SwitchException – not implemented

create_ovs_flow_qualifiers(bridge_id, table_id, flow_id, field, data, priority=2000)

Add row to OvsFlowQualifiers table.

Parameters:

• bridge_id (int) – OVS bridge ID
• table_id (int) – Table ID
• flow_id (int) – Flow ID
• priority (int) – Rule priority
• field (str) – Expression name
• data (str) – Expression data

Returns:

None

Examples:

env.switch[1].ui.create_ovs_flow_qualifiers(0, 0, i, 'EthSrc', '00:00:00:00:00:01')

Raises:SwitchException – not implemented

create_ovs_flow_rules(bridge_id, table_id, flow_id, priority, enabled)

Create OvsFlowRules table.

Parameters:

• bridge_id (int) – OVS bridge ID
• table_id (int) – Table ID
• flow_id (int) – Flow ID
• priority (int) – Rule priority
• enabled (str) – Rule status

Returns:

None

Examples:

env.switch[1].ui.create_ovs_flow_rules(0, 0, 1, 2000, "Enabled")

Raises:SwitchException – not implemented

create_ovs_port(port, bridge_name)

Create OvsPorts record.

Parameters:

• port (int) – port Id
• bridge_name (str) – OVS bridge name

Returns:

None

Examples:

env.switch[1].ui.create_ovs_port(1, 'spp0')

Raises:SwitchException – not implemented

create_static_route(ip, nexthop, network, distance=-1, mode='ip')

Create StaticRoute record.

Parameters:

• ip (str) – Route IP network
• nexthop (str) – Nexthop IP address
• network (str) – RouteInterface network
• distance (int) – Route distance
• mode (str) – ‘ip’ or ‘ipv6’

Returns:

None

Examples:

env.switch[1].ui.create_static_route('20.20.20.0/24', '10.0.5.102', '10.0.5.101/24')

Raises:SwitchException – not implemented

create_syslog(syslog_proto, syslog_ip, syslog_port, syslog_localport, syslog_transport, syslog_facility, syslog_severity)

Configure Syslog settings.

Parameters:

• syslog_proto (str) – syslog host protocol Udp | Tcp
• syslog_ip (str) – syslog host IP address
• syslog_port (int) – syslog host port
• syslog_localport (int) – syslog host local port
• syslog_transport (str) – syslog host transport
• syslog_facility (int) – syslog host facility
• syslog_severity (str) – syslog host severity

Raises:

SwitchException – not implemented

create_tunnels(tunnel_id=None, destination_ip=None, vrf=0, encap_type=None)

Configure TunnelsAdmin table.

Parameters:

• tunnel_id (int) – Tunnel ID
• destination_ip (str) – Destination IP address
• vrf (int) – Tunnel VRF
• encap_type (str) – Tunnel encapsulation type

Returns:

None

Examples:

env.switch[1].ui.create_tunnels(tunnel_id=records_count, destination_ip=ip_list, encap_type='VXLAN')

Raises:SwitchException – not implemented

create_vlan_ports(ports=None, vlans=None, tagged='Tagged')

Creates VLANs on ports using tagged, untagged, or pvid.

Parameters:

• ports (list[int] | set(int)) – list of port IDs
• vlans (list[int] | set(int)) – list of VLAN IDs
• tagged (str) – port tagging attribute

Raises:

ValueError – invalid tagged type

create_vlans(vlans=None)

Add vlans to the ‘fake’ Vlans table.

delete_arp(ip, network, mode='arp')

Delete ARP record.

Parameters:

• ip (str) – ARP ip address
• network (str) – RouteInterface network
• mode (str) – ‘arp’ or ‘ipv6 neigbor’

Returns:

None

Examples:

env.switch[1].ui.delete_arp('10.0.5.102', '10.0.5.101/24')

Raises:SwitchException – not implemented

delete_folder(name)

Delete folder.

Parameters:name (str) – folder name

delete_lag_ports(ports)

Deletes ports from a lag.

Parameters:ports (list[int]) – list of ports to be added into LAG Raises:UIException – Returns:None

delete_lags(lags=None)

Delete lag groups.

Parameters:lags (iter() | str | list[str|int]) – list of lag IDs Raises:NotExistsError – Returns:None

delete_match_api_rule(handle_id, table_id)

Delete a match from the table using the method defined in maa.py.

Parameters:

• handle_id (int) – handle for match.[MANDATORY]
• table_id (int) – the source table id where match to be set.[MANDATORY]

delete_match_api_tcam_subtable(source_id, table_id=0, table_name=None)

Destroy a sub-table of tcam using the method defined in maa.py.

Parameters:

• source_id (int) – the source id in the tcam table.[MANDATORY]
• table_id (int) – a given table id.[MANDATORY if table_name not specified]
• table_name (str) – a given table name.[MANDATORY if table_id not specified]

delete_namespace(name)

Delete network namespace.

Parameters:name (str) – netns name

delete_ovs_bridge()

Delete OVS Bridge.

Returns:None

Examples:

env.switch[1].ui.delete_ovs_bridge()

Raises:SwitchException – not implemented

delete_ovs_flow_actions(bridge_id, table_id, flow_id, action, priority=2000)

Delete row from OvsFlowActions table.

Parameters:

• bridge_id (int) – OVS bridge ID
• table_id (int) – Table ID
• flow_id (int) – Flow ID
• priority (int) – Rule priority
• action (str) – Action name

Returns:

None

Examples:

env.switch[1].ui.delete_ovs_flow_actions(0, 0, 1, 'Output')

Raises:SwitchException – not implemented

delete_ovs_flow_qualifiers(bridge_id, table_id, flow_id, field, priority=2000)

Delete row from OvsFlowQualifiers table.

Parameters:

• bridge_id (int) – OVS bridge ID
• table_id (int) – Table ID
• flow_id (int) – Flow ID
• priority (int) – Rule priority
• field (str) – Expression name

Returns:

None

Examples:

env.switch[1].ui.delete_ovs_flow_qualifiers(bridgeId, tableId, flowId, field)

Raises:SwitchException – not implemented

delete_ovs_flow_rules(bridge_id, table_id, flow_id, priority)

Delete row from OvsFlowRules table.

Parameters:

• bridge_id (int) – OVS bridge ID
• table_id (int) – Table ID
• flow_id (int) – Flow ID
• priority (int) – Rule priority

Returns:

None

Examples:

env.switch[1].ui.delete_ovs_flow_rules(bridgeId, tableId, flowId, priority)

Raises:SwitchException – not implemented

delete_static_route(network)

Delete StaticRoute record.

Parameters:network (str) – RouteInterface network Returns:None

Examples:

env.switch[1].ui.delete_static_route('10.0.5.101/24')

Raises:SwitchException – not implemented

delete_vlan_ports(ports=None, vlans=None)

Removes vlans from ports.

Parameters:

• ports (list[int]) – list of port IDs
• vlans (list[int]) – list of VLAN IDs

delete_vlans(vlans=None)

Remove vlans from the ‘fake’ Vlans table.

disable_lldp_on_device_ports(ports=None)

Disable Lldp on device ports (if port=None Lldp should be disabled on all ports).

Parameters:ports (list[int]) – list of ports Returns:None

Examples:

env.switch[1].ui.disable_lldp_on_device_ports()

disconnect()

Disconnects the ssh session from the switch.

enable_dcbx_tlv_transmission(ports, dcbx_tlvs='all', mode='Enabled')

Enable/Disable the transmission of all Type-Length-Value messages.

Parameters:

• ports (list[int]) – list of ports
• dcbx_tlvs (str) – TLV message types
• mode (str) – “Enabled” or ‘Disabled’

Returns:

None

Examples:

env.switch[1].ui.enable_dcbx_tlv_transmission([1, 2], dcbx_tlvs="all", mode="Enabled")

Raises:SwitchException – not implemented

enter_namespace(name)

Add netns prefix to the command.

Parameters:name (str) – netns name

exit_namespace()

Remove netns prefix from the command.

generate_port_name(port)

Attempts to translate port in the port_map.

Parameters:port (int | str) – port in the port_map Raises:UIException – Returns:int | str

generate_port_name_mapping()

Returns the device name (e.g. sw0p1), given a port number and vice versa.

get_cpu()

Returns cpu utilization from switch.

Returns:cpu utilization from switch Return type:float

get_cpu_cores()

Return number of CPU cores.

Returns:Number of cpu cores Return type:int

get_current_date(date_format='+%Y-%m-%d %T')

Returns current date on device.

Parameters:date_format (str) – Date format to be returned Returns:Current date on device Return type:str

get_dhcp_client_lease_time(ports=None)

Returns the lease time configured for the interfaces, will raise

exception on any error

Parameters:ports (list[int]) – List of port ids which denotes the switch interfaces Raises:SwitchException – no switch ports found Returns:Dictionary with port id as key and lease time as value Return type:dict

get_dhcp_status(ps_name='dhcpd')

Gets DHCP status.

Parameters:ps_name (str) – Service name Returns:Values ‘active’|’unknown’ Return type:str

Steps:

-# Execute the command to get the status of "dhcpd" service
-# Return the result

get_icmp_ping_result(ip_addr, ip_version, options='', timeout=None, expected_rcs=frozenset({0}))

Return parsed result of ping command.

Parameters:

• ip_addr (str) – the destination ip address to be pinged
• ip_version (int) – user specified ip address version
• options (str) – options for the ping command
• timeout (int) – command execution timeout
• expected_rcs (int | set | list | frozenset) – expected results

Returns:

a dictionary containing various statistics related to a ping command

Return type:

dict

get_memory(mem_type='usedMemory')

Returns free cached/buffered memory from switch.

Parameters:mem_type (str) – memory type Returns:memory size Return type:float

get_port_configuration(port, **kwargs)

Returns attribute value (int) for given port.

Parameters:

• port (int | str) – port ID
• kwargs (dict) – Possible parameters: “getPortAttr”, “getPortStats”

Returns:

port attribute value

Return type:

int | str

Raises:

AccessError –

get_port_configuration_snapshot(port, stats='attributes', skip_list=frozenset({0}))

Get a list of port attributes and their values.

Parameters:

• port (int) – port id
• stats (str) – stats to retrieve (attributes only currently)
• skip_list (list | set) – names to skip

Returns:

dict

get_proc_mem_info()

Get proc mem info.

Returns:Returns /proc/meminfo dictionary Return type:dict

get_rules_match_api(table_id, handle_id=None)

Lists the match api rules of the table using the method defined in maa.py.

Parameters:

• table_id (int) – table ID (mandatory parameter)
• handle_id (int) – optional parameter

Returns:

list[dict]

get_table_arp(mode='arp')

Get ARP table.

Parameters:mode (str) – ‘arp’ or ‘ipv6 neigbor’ Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_arp()

Raises:SwitchException – not implemented

get_table_basic_statistics(port, stat_name=None)

Returns a list of basic statistics found in /sys/class/net/.

Parameters:

• port (int) – Port ID
• stat_name (str) – Statistics name

Returns:

Statistics table or specific statistics value

Return type:

dict | int

get_table_bridge_info(param=None, port=None)

Retrieves switch properties organized under ONS 1.x.

Parameters:

• param (str) – parameter name
• port (int) – port ID

Raises:

KeyError | UIException – port/management port info required

Returns:

switch properties

Return type:

str | int

get_table_dcbx_app_maps(table_type='Admin', port=None)

Get DcbxAppMaps* table

Parameters:

• table_type (str) – “Admin”, “Local” or “Remote”
• port (int) – port Id (optional)

Returns:

table (list of dictionaries)

Return type:

list[dict]

Examples:

env.switch[1].ui.get_table_dcbx_app_maps("Admin", 1)

Raises:SwitchException – not implemented

get_table_dcbx_pfc(table_type='Local', port=None)

Get DcbxRemotes* table.

Parameters:

• port (int) – port Id (optional)
• table_type (str) – Table types “Admin”| “Local”| “Remote”

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_dcbx_pfc()

Raises:SwitchException – not implemented

get_table_dcbx_ports(port=None, param=None)

Get DcbxPorts table.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]

Examples:

env.switch[1].ui.get_table_dcbx_ports()

Raises:SwitchException – not implemented

get_table_dcbx_remotes(port=None, param=None)

Get DcbxRemotes* table.

Args:

port(int): port Id (optional) param(str): parameter name (optional)

Returns:

list[dict]|int|str: table (list of dictionaries) or value

Examples:

env.switch[1].ui.get_table_dcbx_remotes(1)

Raises:SwitchException – not implemented

get_table_dhcp_client_lease(file_name='/var/lib/dhcpd/dhcpd.leases')

Gets data from a lease file

Parameters:file_name (str) – File name Returns:dictionary or a particular key’s value based on the options provided Return type:dict

Steps:

-# Using sftp client open the config file in read only mode
-# Store the configurations in dictionary format
-# Return the entire dictionary or a particular key's value based on
   the options provided

get_table_dhcp_config(file_name='/etc/dhcp/dhcpd.conf', options=None)

Gets configuration from file.

Parameters:

• file_name (str) – File name
• options (str) – Name of options to get

Returns:

DHCP configuration

Return type:

dict

Steps:

-# Using sftp client open the config file in read only mode
-# Store the configurations in dictionary format
-# Return the entire dictionary or a particular key's value based on the options provided

get_table_dhcp_relay(dhcp_relay_ipv6=False)

Return DhcpRelayAdmin or DhcpRelayV6Admin table

Parameters:dhcp_relay_ipv6 (bool) – is IPv6 config defined Returns:None

Examples:

env.switch[1].ui.get_table_dhcp_relay(dhcp_relay_ipv6=False)

Raises:SwitchException – not implemented

get_table_dhcp_server_ip_list(pool=None)

Gets configured range of ip address.

Parameters:pool (str) – “range” Returns:Range of ip address Return type:list

Steps:

-# From server configuration file get the range
-# Return the list

get_table_lags()

Get lag group information.

Notes

We can also use networkctl lag command

Returns:list[dict]

get_table_link_aggregation()

Get LinkAggregation table.

Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_link_aggregation()

get_table_lldp(param=None)

Get Lldp table.

Parameters:param (str) – parameter name (optional) Returns:table (list of dictionaries) Return type:list[dict]|int|str

Examples:

env.switch[1].ui.get_table_lldp()

get_table_lldp_ports(port=None, param=None)

Get LldpPorts table.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_lldp_ports(1)

Raises:SwitchException – not implemented

get_table_lldp_ports_stats(port=None, param=None)

Get LldpPorts table statistics.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_lldp_ports_stats(1)

Raises:SwitchException – not implemented

get_table_lldp_remotes(port=None)

Get LldpRemotes table.

Parameters:port (int) – port Id (optional) Returns:table (list of dictionaries) or value Return type:list[dict]

Examples:

env.switch[1].ui.get_table_lldp_remotes(1)

Raises:SwitchException – not implemented

get_table_match_api(table_id=None)

Lists the match api tables using the method defined in maa.py

Parameters:table_id (int) – table ID Returns:list[dict]

get_table_ovs_bridges()

Get OvsBridges table.

Returns:table (list of dictionaries)) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ovs_bridges()

Raises:SwitchException – not implemented

get_table_ovs_controllers()

Get OvsControllers table.

Returns:table (list of dictionaries)) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ovs_controllers()

Raises:SwitchException – not implemented

get_table_ovs_flow_actions()

Get OvsFlowActions table.

Returns:table (list of dictionaries)) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ovs_flow_actions()

Raises:SwitchException – not implemented

get_table_ovs_flow_qualifiers()

Get OvsFlowQualifiers table.

Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ovs_flow_qualifiers()

Raises:SwitchException – not implemented

get_table_ovs_ports()

Get OvsPorts table.

Returns:table (list of dictionaries)) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ovs_ports()

Raises:SwitchException – not implemented

get_table_ovs_rules()

Get OvsFlowRules table.

Returns:table (list of dictionaries)) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ovs_rules()

Raises:SwitchException – not implemented

get_table_platform()

Get ‘Platform’ table.

get_table_ports(ports=None, all_params=False, ip_addr=False)

Returns the table ports dictionary.

Parameters:

• ports (list[int] | None) – list of port IDs
• all_params (bool) – get additional port properties
• ip_addr (bool) – Get IP address

Raises:

SwitchException – No switch ports found

Returns:

ports table

Return type:

list[dict]

get_table_ports2vlans()

Gets the ports to vlan table.

Returns:list[dict]

get_table_remotes_mgmt_addresses(port=None)

Get LldpRemotesMgmtAddresses table.

Parameters:port (int) – port Id (optional) Returns:table (list of dictionaries) or value Return type:list[dict]

Examples:

env.switch[1].ui.get_table_remotes_mgmt_addresses(1)

get_table_static_route(mode='ip')

Get StaticRoute table.

Args:

mode(str): ‘ip’ or ‘ipv6’

Returns:

list[dict]: table (list of dictionaries)

Examples:

env.switch[1].ui.get_table_static_route()

Raises:

SwitchException: not implemented

;

get_table_statistics(port=None, stat_name=None)

Returns port statistics via ethtool -S command.

Parameters:

• port (int) – Port ID
• stat_name (str) – Statistics name

Raises:

KeyError – invalid port ID

Returns:

Statistics table or specific statistics value

Return type:

dict | str

get_table_tunnels_admin()

Return TunnelsAdmin table.

Returns:table Return type:list[dict]

Examples:

env.switch[1].ui.get_table_tunnels_admin()

Raises:SwitchException – not implemented

get_table_ufd_config()

Get UFDConfig table.

Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ufd_config()

Raises:SwitchException – not implemented

get_table_vlans()

Returns the ‘fake’ Vlans table.

get_temperature()

Get temperature from Sensors table.

Returns:CPU temperature information (Sensors table) Return type:dict

get_ufd_networkctl_status(ports)

Checking networkctl status.

Parameters:ports (list[int]) – ports to check networkctl status Returns:

Returns Port Status as Dictionary format for list of Ports with attribute

as key and attribute value as value If Uplink port -> Returns Values for Keys {‘Carrier Bound By’, ‘Link File’, ‘Driver’, ‘MTU’, ‘Network File’, ‘State’, ‘Address’, ‘Type’} If Downlink port-> Returns Values for Keys {‘Carrier Bound To’, ‘Link File’, ‘Driver’, ‘MTU’, ‘Network File’, ‘State’, ‘Address’, ‘Type’}

Return type:dict

icmp_ping_request(ip_addr, ip_version, options='', timeout=None, expected_rcs=frozenset({0}))

Execute ping command.

Parameters:

• ip_addr (str) – the destination ip address to be pinged
• ip_version (int) – user specified ip address version
• options (str) – options for the ping command
• timeout (int) – command execution timeout
• expected_rcs (int | set | list | frozenset) – expected results

Returns:

str

iputils_version(options='')

Verify the versions of ping and ping6 in the iputils package.

Parameters:options (str) – options for the ping command Returns:str

is_process_active(process_name)

Check if process active or not.

Parameters:process_name (str) – name of process Returns:True or False Return type:bool

logs_add_message(level, message)

Add message into device logs.

Parameters:

• level (str) – log severity
• message (str) – log message

Raises:

SwitchException – not implemented

map_stat_name(generic_name)

Returns actual counter name.

Parameters:generic_name (str) – counter Name Returns:str

modify_bridge_info(**kwargs)

Used for mac aging time, maintained for ONS 1.x compatibility.

Raises:KeyError – cpu port is not defined

modify_dhcp_status(ps_name='dhcpd', operation='start')

Changes DHCP status.

Parameters:

• ps_name (str) – Service name
• operation (str) – Operations “start” | “stop”

Returns:

Result of execution

Return type:

str

Steps:

-# Execute the command to start/stop the "dhcpd" service
-# Return the result

modify_lags(lag, key=None, lag_type=None, hash_mode=None)

Modify LagsAdmin table.

Parameters:

• lag (int) – LAG id
• key (int) – LAG key
• lag_type (str) – LAG type (Static or Dynamic)
• hash_mode() – LAG hash mode

Returns:

None

Examples

env.switch[1].ui.modify_lags(lag=3800, lag_type=”Static”)

Raises:SwitchException – not implemented

modify_ports(ports, expected_rcs=frozenset({0}), **kwargs)

Modifies settings on a list of ports.

Parameters:

• ports (list[int | str]) – list of port IDs
• expected_rcs (int | list | set | frozenset) – expected return code
• kwargs (dict) – Possible parameters

Raises:

BoundaryError | AccessError –

Returns:

None

modify_vlan_ports(ports=None, vlans=None, tagged='tagged')

Changes vlan classification. Since no modify method exists in NOS, we need to delete the origin entry and re-add.

Parameters:

• ports (list[int]) – list of port IDs
• vlans (list[int]) – list of VLAN IDs
• tagged (str) – port tagging attribute

ntp_update()

Update date and time stamp using NTP server.

Parameters:| None (bool) – status of operation (bool) or None if ntp_server not set

classmethod parse_icmp_ping_result(ping_output)

Parses the output of ping command.

Parameters:ping_output (str) – output of ping command Returns:dict

classmethod parse_ip_show_stats(input_lines)

Returns list of IP statistics.

Parameters:input_lines (list[str]) – command output Returns:list of IP statistics Return type:dict

classmethod parse_row_lag(row)

Yield lag group information.

Will convert lagId to int for ONS 1.x compatibility.

Parameters:row (dict) – dictionary Returns:dict

classmethod parse_table_ports(ports_table)

Returns list of dictionary of port properties.

Parameters:ports_table (list[str]) – port information Returns:port properties Return type:dict

static parse_table_statistics(stats_table)

parse_table_vlan(vlan_table)

Parses the vlan table. This needs to be a loop because previous the table is built based on previous entries.

Parameters:vlan_table (list[str] | iter()) – List of vlan raw output Returns:A dictionary containing the portId, vlanId, and tagged state for each vlan Return type:iter()

static process_table_data(data, table_keys_mapping)

Returns dictionary of items, given a table of elements.

Parameters:

• data (list[str]) – Command execution return data
• table_keys_mapping (dict) – User column name to output column name mapping

Returns:

dict

reinit()

Re-initialize class attributes.

restart()

Restarts the switch via command line ‘reboot’ command.

restart_networkd_service()

Restarting systemd-networkd process.

Returns:True if result is none otherwise false Return type:bool Raises:UiCmdException – when restart fails

restore_config()

Restore device configuration.

Raises:SwitchException – not implemented

static row_with_header(header, data)

Returns dictionary per row of values for ‘ip show stats’.

Parameters:

• header (str) – output header
• data (str) – output data

Returns:

dictionary per row of values for ‘ip show stats’

Return type:

dict

save_config()

Save device configuration.

Raises:SwitchException – not implemented

set_all_ports_admin_disabled()

Disables all ports in port_map on switch.

set_dcb_admin_mode(ports, mode='Enabled')

Enable/Disable DCB on ports.

Parameters:

• ports (list[int]) – list of ports
• mode (str) – “Enabled” or ‘Disabled’

Returns:

None

Examples:

env.switch[1].ui.set_dcb_admin_mode([1, 2], "Enabled")

Raises:SwitchException – not implemented

set_flow_control_type(ports=None, control_type=None, tx_mode='normal', tc=None)

Sets the flow control type.

Parameters:

• ports (list[int]) – list of port ids
• control_type (str) – flow control type
• tx_mode (str) – transmit mode (normal or class based)
• tc (int) – traffic class

start_switchd()

Restarts the switchd instance of the switch.

wait_all_ports_admin_disabled()

Checks if all the ports are set to down.

taf.testlib.linux_host_interface

linux_host_interface.py

LinuxHostInteface class

class taf.testlib.linux_host_interface.LinuxHostInterface

Bases: object

Abstract class to store UI wrapper interface methods.

check_device_state()

Attempts to connect to the shell retries number of times.

clear_config()

Clear device configuration.

clear_statistics()

Clear Statistics.

Returns:None

Examples

env.switch[1].ui.clear_statistics()

clear_table_fdb()

Clear Fdb table.

Returns:None

Examples:

env.switch[1].ui.clear_table_fdb()

configure_application(application, loglevel)

Set application loglevel.

Parameters:

• application (str) – Application Name.
• loglevel (str) – Application loglevel.

Returns:

None

Examples:

env.switch[1].ui.configure_application('L1PortControlApp', 'Debug')

configure_application_priority_rules(ports, app_prio_rules)

Configure Application Priority rules.

Parameters:

• ports (list[int]) – list of ports
• app_prio_rules (list[dict]) – list of rules dictionaries

Returns:

None

Examples:

env.switch[1].ui.configure_application_priority_rules([1, 2], [{"selector": 1, "protocol": 2, "priority":1}, ])

configure_arp(garp=None, refresh_period=None, delay=None, secure_mode=None, age_time=None, attemptes=None)

Configure ARPConfig table.

Parameters:

• garp (str) – AcceptGARP value. ‘True’|’False’
• refresh_period (int) – RefreshPeriod value
• delay (int) – RequestDelay value
• secure_mode (str) – SecureMode value. ‘True’|’False’
• age_time (int) – AgeTime value
• attemptes (int) – NumAttempts value

Returns:

None

Examples:

env.switch[1].ui.configure_arp(garp='Enabled')

configure_dcbx_app(ports, **kwargs)

Configure DCBx APP parameter for the ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “willing”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_app([1, 2])

configure_dcbx_cn(ports, **kwargs)

Configure DCBx CN parameter for the ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “cnpvSupported”; “cnpvReady”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_cn([1, 2], cnpvSupported='Enabled')

configure_dcbx_ets(ports, **kwargs)

Configure DCBx ETS Conf/Reco parameter for ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “willing”; “cbs”; “maxTCs”; “confBandwidth”; “confPriorityAssignment”; “confAlgorithm”; “recoBandwidth”; “recoPriorityAssignment”; “recoAlgorithm”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_ets([1, 2], confBandwidth=100)

configure_dcbx_pfc(ports, **kwargs)

Configure DCBx PFC parameter for the ports list.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: “mbc”; “enabled”; “willing”.

Returns:

None

Examples:

env.switch[1].ui.configure_dcbx_pfc([1, 2])

configure_global_lldp_parameters(**kwargs)

Configure global LLDP parameters.

Parameters:**kwargs (dict) – parameters to be modified: ‘messageFastTx’; ‘messageTxHoldMultiplier’; ‘messageTxInterval’; ‘reinitDelay’; ‘txCreditMax’; ‘txFastInit’; ‘locChassisIdSubtype’. Returns:None

Examples:

env.switch[1].ui.configure_global_lldp_parameters(messageTxInterval=5)

configure_lldp_ports(ports, **kwargs)

Configure LldpPorts records.

Parameters:

• ports (list[int]) – list of ports
• **kwargs (dict) – parameters to be modified: ‘adminStatus’; ‘tlvManAddrTxEnable’; ‘tlvPortDescTxEnable’; ‘tlvSysCapTxEnable’; ‘tlvSysDescTxEnable’; ‘tlvSysNameTxEnable’.

Returns:

None

Examples:

env.switch[1].ui.configure_lldp_ports([1, 2], adminStatus='Disabled')

connect()

Mandatory method for UI wrapper connection.

create_arp(ip, mac, network, mode='arp')

Create StaticARP record.

Parameters:

• ip (str) – ARP ip address
• mac (str) – ARP mac address
• network (str) – RouteInterface network
• mode (str) – ‘arp’ or ‘ipv6 neigbor’

Returns:

None

Examples:

env.switch[1].ui.create_arp('10.0.5.102', '00:00:22:22:22', '10.0.5.101/24')

create_folder(name)

Create folder.

Parameters:name (str) – folder name

create_match_api_rule(prio_id, handle_id, table_id, match_field_value_mask_list, action, action_value=None)

Set a rule into the table.

Parameters:

• prio_id (int) – Higher id has a higher priority.
• handle_id (int) – handle for match.
• table_id (int) – the source table id where match to be set.
• match_field_value_mask_list (list[tuple(str, str, str)]) – field with match field, value and mask.
• action (str) – given action for source table
• action_value (int) – action value

Raises:

• UIException – for TypeError - Not enough arguments for format string
• UIException – In case of Command execution Error reported in MatchAPI

create_match_api_tcam_subtable(source_id, table_id, table_name, max_table_entries, match_field_type_pairs, actions)

Create a sub-table of tcam.

Parameters:

• source_id (int) – the source id in the tcam table.
• table_id (int) – a given table id. If switchd running, table id starts from 5 If matchd is running, table id starts from 4
• table_name (str) – a given table name.
• max_table_entries (int) – maximum number of flows can be set.
• match_field_type_pairs (list[tuple(str, str)]) – list of given match field with match type
• actions (list[str]) – list of actions for configurable matches

create_namespace(name)

Create network namespace

Parameters:name (str) – netns name

create_static_route(ip, nexthop, network, distance=-1, mode='ip')

Create StaticRoute record.

Parameters:

• ip (str) – Route IP network
• nexthop (str) – Nexthop IP address
• network (str) – RouteInterface network
• distance (int) – Route distance
• mode (str) – ‘ip’ or ‘ipv6’

Returns:

None

Examples:

env.switch[1].ui.create_static_route('20.20.20.0/24', '10.0.5.102', '10.0.5.101/24')

create_syslog(syslog_proto, syslog_ip, syslog_port, syslog_localport, syslog_transport, syslog_facility, syslog_severity)

Configure Syslog settings.

Parameters:

• syslog_proto (str) – syslog host protocol Udp | Tcp
• syslog_ip (str) – syslog host IP address
• syslog_port (int) – syslog host port
• syslog_localport (int) – syslog host local port
• syslog_transport (str) – syslog host transport
• syslog_facility (int) – syslog host facility
• syslog_severity (str) – syslog host severity

create_vlan_ports(ports=None, vlans=None, tagged='Tagged')

Create new Ports2Vlans records.

Parameters:

• ports (list[int]) – list of ports to be added to Vlans.
• vlans (list[int] | set(int)) – list of vlans.
• tagged (str) – information about ports tagging state.

Returns:

None

Examples:

Port 1 will be added into the vlans 3 and 4 as Untagged and port 2 will be added into the vlans 3 and 4 as Untagged
env.switch[1].ui.create_vlan_ports([1, 2], [3, 4], 'Untagged')

create_vlans(vlans=None)

Create new Vlans

Parameters:vlans (list[int]) – list of vlans to be created. Returns:None

Examples:

env.switch[1].ui.create_vlans([2, 3])

delete_arp(ip, network, mode='arp')

Delete ARP record.

Parameters:

• ip (str) – ARP ip address
• network (str) – RouteInterface network
• mode (str) – ‘arp’ or ‘ipv6 neigbor’

Returns:

None

Examples:

env.switch[1].ui.delete_arp('10.0.5.102', '10.0.5.101/24')

delete_folder(name)

Delete folder.

Parameters:name (str) – folder name

delete_match_api_rule(handle_id, table_id)

Delete a rule from the table.

Parameters:

• handle_id (int) – handle for match.[MANDATORY]
• table_id (int) – the source table id where match to be set.[MANDATORY]

delete_match_api_tcam_subtable(source_id, table_id=0, table_name=None)

Destroy a sub-table of tcam.

Parameters:

• source_id (int) – the source id in the tcam table.[MANDATORY]
• table_id (int) – a given table id.[MANDATORY if table_name not specified]
• table_name (str) – a given table name.[MANDATORY if table_id not specified]

delete_namespace(name)

Delete network namespace.

Parameters:name (str) – netns name

delete_static_route(network)

Delete StaticRoute record.

Parameters:network (str) – RouteInterface network Returns:None

Examples:

env.switch[1].ui.delete_static_route('10.0.5.101/24')

delete_vlan_ports(ports=None, vlans=None)

delete_vlans(vlans=None)

Delete existing Vlans.

Parameters:vlans (list[int]) – list of vlans to be deleted. Returns:None

Examples:

env.switch[1].ui.delete_vlans([2, 3])

disable_lldp_on_device_ports(ports=None)

Disable Lldp on device ports (if port=None Lldp should be disabled on all ports).

Parameters:ports (list[int]) – list of ports Returns:None

Examples:

env.switch[1].ui.disable_lldp_on_device_ports()

disconnect()

Mandatory method for UI wrapper disconnection.

enable_dcbx_tlv_transmission(ports, dcbx_tlvs='all', mode='Enabled')

Enable/Disable the transmission of all Type-Length-Value messages.

Parameters:

• ports (list[int]) – list of ports
• dcbx_tlvs (str) – TLV message types
• mode (str) – “Enabled” or ‘Disabled’

Returns:

None

Examples:

env.switch[1].ui.enable_dcbx_tlv_transmission([1, 2], dcbx_tlvs="all", mode="Enabled")

enter_namespace(name)

Add netns prefix to the command.

Parameters:name (str) – netns name

exit_namespace()

Remove netns prefix from the command.

get_cpu()

Returns cpu utilization from switch.

Returns:cpu utilization from switch Return type:float

get_memory(mem_type='usedMemory')

Returns free cached/buffered memory from switch.

Parameters:mem_type (str) – memory type Returns:memory size Return type:float

get_rules_match_api(table_id=None, handle_id=None)

Lists the match api rules of the table.

Parameters:

• table_id (int) – table ID (mandatory parameter)
• handle_id (int) – optional parameter

Returns:

list[dict]

get_table_applications()

Get ‘Applications’ table.

Returns:‘Applications’ table Return type:list[dict]

get_table_arp(mode='arp')

Get ARP table.

Parameters:mode (str) – ‘arp’ or ‘ipv6 neigbor’ Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_arp()

get_table_arp_config()

Get ARPConfig table.

Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_arp_config()

get_table_dcbx_app_maps(table_type='Admin', port=None)

Get DcbxAppMaps* table

Parameters:

• table_type (str) – “Admin”, “Local” or “Remote”
• port (int) – port Id (optional)

Returns:

table (list of dictionaries)

Return type:

list[dict]

Examples:

env.switch[1].ui.get_table_dcbx_app_maps("Admin", 1)

get_table_dcbx_pfc(table_type='Local', port=None)

Get DcbxRemotes* table.

Parameters:

• port (int) – port Id (optional)
• table_type (str) – Table types “Admin”| “Local”| “Remote”

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_dcbx_pfc()

get_table_dcbx_ports(port=None, param=None)

Get DcbxPorts table.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]

Examples:

env.switch[1].ui.get_table_dcbx_ports()

get_table_dcbx_remotes(port=None, param=None)

Get DcbxRemotes* table.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_dcbx_remotes(1)

get_table_fdb(table='Fdb')

Get Fbd table.

Parameters:table (str) – Fbd record type to be returned (‘Fbd’ or ‘Static’) Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_fdb()
env.switch[1].ui.get_table_fdb('Static')

get_table_lldp(param=None)

Get Lldp table.

Parameters:param (str) – parameter name (optional) Returns:table (list of dictionaries) Return type:list[dict]|int|str

Examples:

env.switch[1].ui.get_table_lldp()

get_table_lldp_ports(port=None, param=None)

Get LldpPorts table.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_lldp_ports(1)

get_table_lldp_ports_stats(port=None, param=None)

Get LldpPorts table statistics.

Parameters:

• port (int) – port Id (optional)
• param (str) – parameter name (optional)

Returns:

table (list of dictionaries) or value

Return type:

list[dict]|int|str

Examples:

env.switch[1].ui.get_table_lldp_ports_stats(1)

get_table_lldp_remotes(port=None)

Get LldpRemotes table.

Parameters:port (int) – port Id (optional) Returns:table (list of dictionaries) or value Return type:list[dict]

Examples:

env.switch[1].ui.get_table_lldp_remotes(1)

get_table_match_api(table_id=None)

Lists the match api tables.

Parameters:table_id (int) – table ID Returns:list[dict]

get_table_platform()

Get ‘Platform’ table.

get_table_ports(ports=None, all_params=False)

Get ‘Ports’ table.

Parameters:

• ports (list) – list of port IDs.
• all_params (bool) – get additional port properties

Returns:

table (list of dictionaries)

Return type:

list(dict)

Notes

Return all table or information about particular ports.

Examples:

env.switch[1].ui.get_table_ports()
env.switch[1].ui.get_table_ports([1, 2])

get_table_ports2vlans()

Get ‘Ports2Vlans’ table.

Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_ports2vlans()

get_table_remotes_mgmt_addresses(port=None)

Get LldpRemotesMgmtAddresses table.

Parameters:port (int) – port Id (optional) Returns:table (list of dictionaries) or value Return type:list[dict]

Examples:

env.switch[1].ui.get_table_remotes_mgmt_addresses(1)

get_table_static_route(mode='ip')

Get StaticRoute table.

Parameters:mode (str) – ‘ip’ or ‘ipv6’ Returns:table (list of dictionaries) Return type:list[dict]

Examples:

env.switch[1].ui.get_table_static_route()

get_table_statistics(port=None, stat_name=None)

Get Statistics table.

Parameters:

• port (str|int|None) – port Id to get info about (‘cpu’ or port id) (optional)
• stat_name (str) – name of statistics parameter (optional)

Returns:

table (list of dictionaries)

Return type:

list[dict]|int

Examples:

env.switch[1].ui.get_table_statistics()
env.switch[1].ui.get_table_statistics(port=1)
env.switch[1].ui.get_table_statistics(port='cpu')

get_table_vlans()

Get ‘Vlans’ table.

Returns:table (list of dictionaries) Return type:list(dict)

Examples:

env.switch[1].ui.get_table_vlans()

get_temperature()

Get temperature from Sensors table.

Returns:CPU temperature information (Sensors table) Return type:dict

logs_add_message(level, message)

Add message into device logs.

Parameters:

• level (str) – log severity
• message (str) – log message

map_stat_name(generic_name)

Get the UI specific stat name for given generic name.

Parameters:generic_name (str) – generic statistic name Returns:UI specific stat name Return type:str

modify_ports(ports, **kwargs)

Modify records in ‘Ports’ table.

Parameters:

• ports (list) – list of port IDs.
• kwargs (dict) – Parameters to be modified. Parameters names should be the same as in XMLRPC nb.Ports.set.* calls: - “pvid” - set pvid value; - “pvpt” - set pvpt value; - “adminMode” - set adminMode value; - “ingressFiltering” - set ingressFiltering value; - “maxFrameSize” - set maxFrameSize value; - “discardMode” - set discardMode value; - “cutThrough” - set cutThrough value; - “flowControl” - set flowControl value; - “speed” - set speed value; - “learnMode” - set learnMode value.

Returns:

None

Examples:

env.switch[1].ui.modify_ports([1, 2], adminMode='Down')

modify_vlan_ports(ports=None, vlans=None, tagged='Tagged')

Modify Ports2Vlans records.

Parameters:

• ports (list) – list of ports to be added to Vlans.
• vlans (list[int] | set(int)) – list of vlans.
• tagged (str) – information about ports tagging state.

Returns:

None

Examples:

Port 1 will be modified in the vlans 3 and 4 as Tagged
env.switch[1].ui.create_vlan_ports([1, ], [3, 4], 'Tagged')

restart()

Perform device reboot via User Interface.

restore_config()

Restore device configuration.

save_config()

Save device configuration.

set_all_ports_admin_disabled()

Set all ports into admin Down state.

Returns:None

Notes

This method is used in helpers.set_all_ports_admin_disabled() for all functional test case.

set_dcb_admin_mode(ports, mode='Enabled')

Enable/Disable DCB on ports.

Parameters:

• ports (list[int]) – list of ports
• mode (str) – “Enabled” or ‘Disabled’

Returns:

None

Examples:

env.switch[1].ui.set_dcb_admin_mode([1, 2], "Enabled")

set_flow_control_type(ports=None, control_type=None)

Enable/disable sending/accepting pause frames.

Parameters:

• ports (list) – list of port IDs
• control_type (str) – ‘Rx’, ‘Tx’, ‘RxTx’ and ‘None’

Returns:

None

Examples:

env.switch[1].ui.set_flow_control([1, 2], 'RxTx')

wait_all_ports_admin_disabled()

Wait for all ports into admin Down state.

Returns:None

Notes

This method is used in helpers.set_all_ports_admin_disabled() for all functional test case.

taf.testlib.lldp

lldp.py

class taf.testlib.lldp.Tlv

Bases: object

static get_chassis_tlv_row(tlv)

Parameters:tlv (list) – list or dict of the chassis TLV Returns:remMan style dict Return type:dict

static get_local_cap_tlv_row(tlv)

Parameters:tlv (list) – list or dict of the port TLV Returns:row style dict Return type:dict

static get_local_chassis_tlv_row(tlv)

Parameters:tlv (list) – list or dict of the chassis TLV Returns:row style dict Return type:dict

static get_local_port_tlv_row(tlv)

Parameters:tlv (list) – list or dict of the port TLV Returns:remMan style dict Return type:dict

static get_mgmt_row(tlv)

Parameters:tlv (list) – list or dict of the mgmt TLV sub-tlvs Returns:remMan style dict Return type:dict

static get_port_tlv_row(tlv)

Parameters:tlv (list) – list or dict of the port TLV Returns:row style dict Return type:dict

static get_simple_tlv_row(row_name, value)

static get_sys_cap_tlv_row(tlv)

Parameters:tlv (list) – list or dict of the port TLV Returns:row style dict Return type:dict

static get_tlv_from_list(tlvs, predicate)

Parameters:

• tlvs (list) – list tlvs
• predicate (function) – predicate function

Returns:

list

taf.testlib.loggers

loggers.py

logging functionality for TAF

class taf.testlib.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 (object) – class instance Returns:logger adaptor Return type:logging.LoggerAdapter

Notes

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 loging.

Parameters:

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

Returns:

logger adaptor

Return type:

logging.LoggerAdapter

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

Bases: threading.Thread

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

Notes

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.

Notes

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

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

Bases: argparse.ArgumentParser

ArgumentParser class that handle only predefined for an instance options.

Notes

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.testlib.loggers.mkdir_p(path)

Create a directory.

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

Return LoggerAdapter for module level logging.

taf.testlib.loggers.parse_options()

Parse additional cli logging options.

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

Return LoggerWrapper for pipe logging.

taf.testlib.lxc

lxc.py

LXC-specific functionality

class taf.testlib.lxc.Lxc(lxc_ip=None, lxc_iface=None, vlab_ip=None, switch_bin_path=None, switch_id=None, switch_port=8081)

Bases: object

Base class to work with Linux containers.

__init__(lxc_ip=None, lxc_iface=None, vlab_ip=None, switch_bin_path=None, switch_id=None, switch_port=8081)

Init method for container.

Parameters:

• lxc_ip (str) – IP of the container
• lxc_iface (str) – Network interface for container
• vlab_ip (str) – real ip of the local vlab
• switch_bin_path (str) – path to switchpp binary
• switch_id (str) – simswitch id
• switch_port (int) – switchpp xmlrpc port

Returns:

None

class_logger = <logging.LoggerAdapter object>

start(stdout=None, stderr=None)

Starts container.

Parameters:

• stdout (str) – Output file
• stderr (str) – Error file

Returns:

None

stop()

Stops container.

Returns:None

taf.testlib.lxc.mkdir_p(path)

Create dir and pass exception if one already exists.

taf.testlib.magnum

magnum.py

Support for magnum feature of Openstack

Uses magnum tempest client that was taken form the upstream magnum repository. When using magnum clients, the environment.json file should contain some additional entries:

class taf.testlib.magnum.Magnum(venv)

Bases: object

CLASS_LOGGER = <logging.LoggerAdapter object>

create_cluster(wait=True, **kwargs)

create_cluster_template(template, **kwargs)

delete_cluster(uuid, wait=True)

delete_cluster_template(uuid)

taf.testlib.multicall

multcall.py

Implement simple python command line based RPC

taf.testlib.multicall._ssh_cli_test()

Private function to test RPC on the command line with ssh.

taf.testlib.multicall._test_env_var_size()

Private function to verify the maximum command line payload size.

This should be smatter and bisect, but it doesn’t

taf.testlib.multicall.bisect_if_too_large(cmd_list, payload_generator, max_size=98304)

Check the compressed payload lenght, if it is larger than the max CLI length (usually 127K) we bisect until we get under the limit.

If an individual cmd is greater than the max_size then we will return the original list since we can’t split a single command.

Parameters:

• cmd_list (list) – list of commands to split based on encoded payload size
• payload_generator (function) – function that generates the payload
• max_size (int) – maximun network payload size

Returns:

iterator over all generator payloads for all the split cmd_list chunks

Return type:

iter(str)

taf.testlib.multicall.decode(payload_str)

Decode base64 encoded bz2 compress network payload

Parameters:payload_str (str) – base64 encoded bz2 compressed byte stream Returns:original byte-stream Return type:str

taf.testlib.multicall.encode(payload)

Encode a string for tranmission over the network using base64 encoding of the bz2 compressed string.

We bz2 compress because we can and also to counteract the inefficiency of the base64 encoding.

Parameters:payload (str) – string we want to transmit over the wire Returns:base64 encoded, bz2 compressed string Return type:str

taf.testlib.multicall.gen_payload(runner, parallel_limit, cmd_list)

taf.testlib.multicall.generate_calls(cmd_list, parallel_limit=None, remote_module_template=<module 'taf.testlib.remote_multicall_template' from '/home/docs/checkouts/readthedocs.org/user_builds/taf-docs/checkouts/master/taf/testlib/remote_multicall_template.py'>)

Generate a sequence of simple command line python RPC

Take the supplied python module, get the source and then insert the command list as a base64 encoded bz2 compressed json

We use the double encoding because there doesn’t seem to be a safe way to inject JSON into a python module source. We know the base64, bz2 encoding is safe and compact, so just re-use that.

The template uses a line like this, the raw triple-quoted string will be replaced with the encoded template:

cmd_list = json.loads(bz2.decompress(base64.urlsafe_b64decode(r"""{}""")))

Parameters:

• cmd_list (iter(str)) – list of command
• parallel_limit (int) – maximum number of parallel commands, None for automatic setting based on number of remote cpus
• remote_module_template (module) – python module used as template for remote python module

taf.testlib.multiple_tg

pytest_onsenv.py

Multiple traffic generator specific functionality

class taf.testlib.multiple_tg.MultipleTG(traffic_generators, config, opts)

Bases: taf.testlib.packet_processor.PacketProcessor, taf.testlib.tg_template.GenericTG

Class for general TG instance combined with multiple different TGs.

__init__(traffic_generators, config, opts)

Initialize RemoteMultiHostTG class.

Parameters:

• traffic_generators (dict) – Dictionary with TG instances in format {id:tg_instance}
• config (dict) – Configuration information.
• opts (OptionParser) – py.test config.option object which contains all py.test cli options.

_get_speed_ports()

Get ports with speed from TG instances.

Returns:Tuple with list of ports used in real config and list of port/speed values Return type:tuple(list[tuple], list[tuple, int])

check()

Check TG instances.

class_logger = <logging.LoggerAdapter object>

cleanup(*args, **kwargs)

Cleanup TG instances.

clear_received_statistics(iface)

Clear statistics.

clear_statistics(sniff_port_list)

Clearing statistics on TG ports.

clear_streams()

Stop and remove all streams.

connect_port(iface)

Simulate port link connecting (set it to admin up etc).

Parameters:iface (str) – Interface to connect. Raises:NotImplementedError – not implemented Returns:None or raise and exception.

create()

Start TG instances or get running ones.

destroy()

Stop or release TG instances.

disconnect_port(iface)

Simulate port link disconnecting (set it to admin down etc).

Parameters:iface (str) – Interface to disconnect. Raises:NotImplementedError – not implemented Returns:None or raise and exception.

get_filtered_frames_count(iface)

Read statistics - filtered frames received.

get_os_mtu(iface=None)

Get MTU value in host OS.

Parameters:iface (str) – Interface for getting MTU in host OS Returns:Original MTU value Return type:int

Examples:

env.tg[1].get_os_mtu(iface=ports[('tg1', 'sw1')][1])

get_port_id(tg_id, port_id)

Return port’s sequence number in list of ports.

Parameters:

• tg_id (int) – TG instance ID
• port_id (int) – TG instance port’s sequence number

Raises:

ValueError – in case expected port is not in list of ports

Returns:

Port sequence number in list of ports starting from 1

Return type:

int

get_port_qos_rxrate(iface, qos)

Get ports Rx rate for specific qos.

get_port_rxrate(iface)

Get port Rx rate.

get_port_txrate(iface)

Get port Tx rate.

get_qos_frames_count(iface, prio)

Get QoS packets count.

get_received_frames_count(iface)

Read statistics - framesReceived.

get_sent_frames_count(iface)

Read statistics - framesSent.

get_tg_port_map(ifaces)

Return ports related to specific TG.

Parameters:ifaces (list(tuple)) – list of interfaces in format (tg_id, port_id) Returns:dictionary in format {‘host id’: [port ids]} Return type:dict

get_uds_3_frames_count(iface)

Read statistics - UDS3 - Capture Trigger (UDS3) - count of non-filtered received packets (valid and invalid).

sanitize()

Perform any necessary operations to leave environment in normal state.

send_stream(stream_id, **kwargs)

Sends the stream created by ‘set_stream’ method.

Parameters:stream_id (int) – ID of the stream to be send. Returns:timestamp. Return type:float

set_flow_control(iface, mode)

Set Flow Control.

set_os_mtu(iface=None, mtu=None)

Set MTU value in host OS.

Parameters:

• iface (str) – Interface for changing MTU in host OS
• mtu (int) – New MTU value

Returns:

Original MTU value

Return type:

int

Examples:

env.tg[1].set_os_mtu(iface=ports[('tg1', 'sw1')][1], mtu=1650)

set_qos_stat_type(iface, ptype)

Set QoS stats type.

set_stream(*args, **kwargs)

Set traffic stream with specified parameters on specified TG port.

Returns:stream id Return type:int

Notes

It’s not expected to configure a lot of incrementation options. Different traffic generator could have different limitations for these options.

Examples:

stream_id_1 = tg.set_stream(pack_ip, count=100, iface=iface)
stream_id_2 = tg.set_stream(pack_ip, continuous=True, inter=0.1, iface=iface)
stream_id_3 = tg.set_stream(pack_ip_udp, count=5, protocol_increment=(3, 5), iface=iface)
stream_id_4 = tg.set_stream(pack_ip_udp, count=18, sip_increment=(3, 3), dip_increment=(3, 3), iface=iface,
                            udf_dependancies={'sip_increment': 'dip_increment'})

start(wait_on=True)

Start TG instances.

start_sniff(ifaces, **kwargs)

Starts sniffing on specified interfaces.

Parameters:

• ifaces (list) – List of TG interfaces for capturing.
• kwargs (dict) – Possible parameters to configure.

Returns:

None

Notes

This method introduces additional 1.5 seconds timeout after capture enabling. It’s required by Ixia sniffer to wait until capturing is started.

Examples:

env.tg[1].start_sniff(['eth0', ], filter_layer='IP', src_filter='00:00:00:01:01:01', dst_filter='00:00:00:22:22:22')

start_streams(stream_list)

Enable and start streams from the list simultaneously.

Parameters:stream_list (list[int]) – List of stream IDs. Returns:None

stop()

Shutdown TG instances.

stop_sniff(*args, **kwargs)

Collect sniffed data from all TG instances.

stop_streams(stream_list=None)

Disable streams from the list.

Parameters:stream_list (list[int]) – Stream IDs to stop. In case stream_list is not set all running streams will be stopped. Returns:None

class taf.testlib.multiple_tg.Port(tg, port)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, tg, port)

Create new instance of Port(tg, port)

__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>, len=<built-in function len>)

Make a new Port object from a sequence or iterable

_replace(**kwds)

Return a new Port object replacing specified fields with new values

port

Alias for field number 1

tg

Alias for field number 0

taf.testlib.netconfcmd

netconfcmd.py

Module for working with devices via Netconf over ssh connection

class taf.testlib.netconfcmd.NETCONF(config)

Bases: object

Class for configure device using Netconf.

Parameters:config – Switch configuration dictionary.

Examples:

client=NETCONF(env.switch[1].config)

__init__(config)

Initialize NETCONF class.

Parameters:config (dict) – Switch configuration dictionary.

check_session()

Check if connection exist.

Returns:True if session connected Return type:bool

class_logger = <logging.LoggerAdapter object>

close()

Close netconf connection.

connect(host, port=830, username=None, password=None, timeout=None)

Establish netconf session.

Parameters:

• host (str) – Host name or IP address
• port (int) – Host port
• username (str) – Host user
• password (str) – Host password
• timeout (int) – Session timeout

Returns:

NETCONF session

Return type:

ncclient.transport.Session

exec_operation(param, filtering=None)

Return result of the executed Netconf operation.

Parameters:

• param (str) – Name of Netconf operation (Currently we use only “get” operation, but also can be used all other operation as well)
• filtering (tuple) – Consists of two elements Filter type (“xpath” or “subtree”) and criteria

Returns:

Netconf GetReply instance

Notes

This is wrapper for netconf calls.

Examples:

In [3]:  netconf.exec_operation("get", ("xpath", "//system/uname"))
Out[3]:
<?xml version="1.0" encoding="UTF-8"?>
<rpc-reply xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0"
  message-id="urn:uuid:1951ee4a-df07-11e2-9ddb-20cf3095bf10"
  last-modified="2013-06-27T08:33:02Z"
  xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>
    <system xmlns="http://netconfcentral.org/ns/yuma-system">
      <uname>
        <sysname>Linux</sysname>
        <release>3.2.0-48-generic</release>
        <version>#74-Ubuntu SMP Thu Jun 6 19:43:26 UTC 2013</version>
        <machine>x86_64</machine>
        <nodename><username></nodename>
      </uname>
    </system>
  </data>
</rpc-reply>

taf.testlib.ovshelpers

ovshelpers.py

Helpers functions for OVS test suites

taf.testlib.ovshelpers.add_flow_via_controller(qualifiers, action, controller, name=None, dpid=None)

Add flow to OVS bridge via Controller.

Parameters:

• qualifiers (str) – flow qualifiers
• action (str) – flow actions
• controller (OvsControllerGeneralMixin) – Controller
• name (str) – flow name
• dpid (bool) – Get process ID

Returns:

None

taf.testlib.ovshelpers.create_packet_definition(packet_to_send)

Create packet definition to send.

Parameters:packet_to_send (dict) – dictionary with specified packet type and fields with values Returns:packet_definition Return type:tuple(dict)

taf.testlib.ovshelpers.delete_flow_via_controller(qualifiers, controller, name=None)

Delete flow from OVS bridge via Controller.

Parameters:

• qualifiers (str) – flow qualifiers
• controller (OvsControllerGeneralMixin) – Controller
• name (str) – flow name

Returns:

None

taf.testlib.ovshelpers.set_ovs_complex_config_with_controller(ports, switch, sw_id, controller)

Set OVS configuration on Switch.

Parameters:

• ports (dict) – Links information
• switch (SwitchGeneral) – switch
• sw_id (int) – id of switch in complex setup on which OVS is configured
• controller (OvsControllerGeneralMixin) – Controller

Returns:

None

taf.testlib.ovshelpers.set_ovs_complex_test_preconditions(ports, test_preconditions, switch, controller, dpid=None)

Set test preconditions, e.g. flows.

Parameters:

• ports (dict) – ports
• test_preconditions (tuple) – test preconditions
• switch (SwitchGeneral) – switch
• controller (OvsControllerGeneralMixin) – Controller
• dpid (bool) – Get process ID

Returns:

None

taf.testlib.ovshelpers.set_ovs_config_with_controller(ports, switch, controller)

Set OVS configuration on Switch.

Parameters:

• ports (dict) – Links information
• switch (SwitchGeneral) – switch
• controller (OvsControllerGeneralMixin) – Controller

Returns:

None

taf.testlib.ovshelpers.set_ovs_test_preconditions(ports, test_preconditions, switch, controller)

Set test preconditions, e.g. flows.

Parameters:

• ports (dict) – ports
• test_preconditions (tuple) – test preconditions
• switch (SwitchGeneral) – switch
• controller (OvsControllerGeneralMixin) – Controller

Returns:

None

taf.testlib.packet_processor

packet_processor.py

Packet processor specific functionality

class taf.testlib.packet_processor.PacketProcessor

Bases: object

This class contents only one method to build packet from tuple of dictionaries.

_build_pypacker_packet(packet_definition, adjust_size=True, required_size=64)

Builds pypacker packet based on provided packet definition.

Parameters:

• packet_definition (tuple(dict)) – Packet representation (tuple of dictionaries of dictionaries)
• adjust_size (bool) – If set to True packet size will be increased to 64 bytes for Pypacker TG (CRC is included) and to 60 bytes for Ixia TG without CRC. Otherwise Ixia TG will add 4 bytes(CRC), Pypacker TG will not add CRC.
• required_size (int) – Size (in bytes) of the result packet

Returns:

pypacker.packet or list of pypacker packets (if fragsize defined)

Return type:

pypacker.Packet

Examples:

packet_definition = ({"Ethernet": {"dst": "00:80:C2:00:00:00", "src": "00:00:00:00:00:02"}},
                     {"Dot1Q": {"vid": 4}},
                     {"IP": {}})
packet=env.tg[1]._build_pypacker_packet(packet_definition)

_build_trex_packet(packet_definition, adjust_size=True, required_size=64)

Builds trex packet based on provided packet definition.

Parameters:

• packet_definition (tuple(dict)) – Packet representation (tuple of dictionaries of dictionaries)
• adjust_size (bool) – If set to True packet size will be increased to 60 bytes (CRC not included)
• required_size (int) – Size (in bytes) of the result packet

Returns:

trex.packet or list of pypacker packets (if fragsize defined)

Return type:

trex.Packet

Examples:

packet_definition = ({"Ether": {"dst": "00:80:C2:00:00:00", "src": "00:00:00:00:00:02"}},
                     {"Dot1Q": {"vlan": 4}},
                     {"IP": {}})
packet=env.tg[1]._build_trex_packet(packet_definition)

_get_pypacker_layer(layer)

Get Pypacker protocol object.

Parameters:layer (str) – Protocol name e.g IP, ICMP.Echo Returns:return Pypacker object Return type:pypacker.Packet Raises:PypackerException – Pypacker library does not support protocol

static _get_pypacker_layer_fields(packet)

Get packet fields.

Parameters:packet (pypacker.Packet) – pypacker packet Returns:set of packet fields Return type:set

assemble_fragmented_packets(packets)

Method for assembling packets from fragments in packet list.

Parameters:packets (list[pypacker.Packet]) – List of fragmened packets Returns:List of assembled packets Return type:list[pypacker.Packet]

check_packet_field(packet=None, layer=None, field=None, value=None)

Check if specified field is present (for specified layer) and checks if field value matches specified value.

Parameters:

• packet (pypacker.Packet) – Packet to analyze
• layer (str) – Layer to analyze
• field (str) – Field to look for
• value (str, int) – Filed value to compare (may be different types, depending on field)

Returns:

True or False

Return type:

bool

Examples:

assert check_packet_field(packet=pypacker_packet, layer="Dot1Q", field="prio", value=4)
assert check_packet_field(packet=pypacker_packet, layer="Dot1Q", field="type")

class_logger = <logging.LoggerAdapter object>

flt_patterns = {'ARP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 06', '00 00', '12']}, 'BOOTP': {'cfp': 'pattern1', 'mt1': 'matchUser', 'ptrn1': ['08 00', '00 00', '42']}, 'Dot1Q': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchVlan', 'ptrn1': None}, 'Dot1Q.ARP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 08 06', '00 00 FF FF 00 00', '12']}, 'Dot1Q.ICMP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 08 00 00 00 00 00 00 00 00 00 00 01', '00 00 FF FF 00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'Dot1Q.ICMPv6': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 86 dd 00 00 00 00 00 00 00 00 00 01', '00 00 FF FF 00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'Dot1Q.IP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 08 00', '00 00 FF FF 00 00', '12']}, 'Dot1Q.IPv6': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 86 dd', '00 00 FF FF 00 00', '12']}, 'Dot1Q.TCP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 08 00 00 00 00 00 00 00 00 00 00 06', '00 00 FF FF 00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'Dot1Q.UDP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['81 00 00 00 08 00 00 00 00 00 00 00 00 00 00 11', '00 00 FF FF 00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'ICMP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 00 00 00 00 00 00 00 00 00 00 01', '00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'ICMPv6': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['86 dd 00 00 00 00 00 00 00 00 00 01', '00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'IP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 00', '00 00', '12']}, 'IPv6': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['86 dd', '00 00', '12']}, 'LLDP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['01 80 c2 00 00 0e 00 00 00 00 00 00 88 cc', '00 00 00 00 00 00 FF FF FF FF FF FF 00 00', '0']}, 'PAUSE': {'cfp': 'pattern1', 'mt1': 'matchUser', 'ptrn1': ['88 08', '00 00', '12']}, 'STP': {'cfp': 'pattern1', 'mt1': 'matchUser', 'ptrn1': ['42 42 03 00 00', '00 00 00 00 00', '14']}, 'TCP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 00 00 00 00 00 00 00 00 00 00 06', '00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'UDP': {'cfp': 'pattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 00 00 00 00 00 00 00 00 00 00 11', '00 00 FF FF FF FF FF FF FF FF FF 00', '12']}, 'notARP': {'cfp': 'notPattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 06', '00 00', '12']}, 'notIP': {'cfp': 'notPattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 00', '00 00', '12']}, 'notSTP': {'cfp': 'notPattern1', 'mt1': 'matchUser', 'ptrn1': ['42 42 03 00 00', '00 00 00 00 00', '14']}, 'notUDP': {'cfp': 'notPattern1', 'lfilter': <function PacketProcessor.<lambda>>, 'mt1': 'matchUser', 'ptrn1': ['08 00 00 00 00 00 00 00 00 00 00 11', '00 00 FF FF FF FF FF FF FF FF FF 00', '12']}}

get_packet_field(packet=None, layer=None, field=None)

Returns field value (for specified layer) from specified packet.

Parameters:

• packet (pypacker.Packet) – Packet to analyze
• layer (str) – Layer to analyze
• field (str) – Field to look for

Raises:

PypackerException – unknown layer or field

Returns:

value (may be different types, depending on field)

Return type:

int, str

Examples:

value = get_packet_field(packet=pypacker_packet, layer="Dot1Q", field="vlan")

get_packet_layer(packet=None, layer=None, output_format='pypacker')

Returns packet layer in pypacker or hex format.

Parameters:

• packet (pypacker.Packet) – Packet to analyze
• layer (str) – Layer to analyze
• output_format (str) – Output format - “pypacker” or “hex” or “bytes_array”

Returns:

pypacker.Packet or str

Return type:

pypacker.Packet, str

Examples:

packet_definition = ({"Ethernet": {"dst": "00:80:C2:00:00:00", "src": "00:00:00:00:00:02"}}, {"Dot1Q": {"vlan": 4}}, {"IP": {}})
pypacker_packet = _build_pypacker_packet(packet_definition)
ip_layer_hex = get_packet_layer(packet=pypacker_packet, layer="IP", output_format="hex")
assert ip_layer_hex[-8:] == '7f000001'

inner_classes = {'Echo', 'Error', 'ParamProb', 'Redirect', 'TimeExceed', 'TooBig', 'Unreach'}

lldp_lacp_tlvs = {'DCBXApplicationPriority', 'DCBXApplicationPriorityTable', 'DCBXConfiguration', 'DCBXCongestionNotification', 'DCBXPriorityBasedFlowControlConfiguration', 'DCBXRecommendation', 'LACPActorInfoTlv', 'LACPCollectorInfoTlv', 'LACPPartnerInfoTlv', 'LACPReserved', 'LACPTerminatorTlv', 'LLDPChassisId', 'LLDPDUEnd', 'LLDPDot1PortVlanId', 'LLDPGeneric', 'LLDPManagementAddress', 'LLDPOrgSpecGeneric', 'LLDPPortDescription', 'LLDPPortId', 'LLDPSystemCapabilities', 'LLDPSystemDescription', 'LLDPSystemName', 'LLDPTTL'}

packet_dictionary(packet)

Get packet dictionary from pypacker.Packet.

Parameters:packet (pypacker.Packet) – pypacker packet Returns:dictionary created from pypacker.Packet Return type:dict

Examples:

p = pypacker.Ethernet(dst="ff:ff:ff:ff:ff:ff", src="90:e6:ba:c3:17:13", type=0x806)/
    pypacker.ARP(hwtype=0x1, ptype=0x800, hwlen=6, plen=4,
                 op=1, hwsrc="90:e6:ba:c3:17:13", psrc="172.20.20.175", hwdst="00:00:00:00:00:00", pdst="172.20.20.211")/
    pypacker.Padding(load='')
pd = PacketProcessor().packet_dictionary(p)
assert pd == ({'Ether': {'src': '90:e6:ba:c3:17:13', 'dst': 'ff:ff:ff:ff:ff:ff', 'type': 2054}},
              {'ARP': {'hwdst': '00:00:00:00:00:00', 'ptype': 2048, 'hwtype': 1, 'psrc': '172.20.20.175',
                       'plen': 4, 'hwlen': 6, 'pdst': '172.20.20.211', 'hwsrc': '90:e6:ba:c3:17:13', 'op': 1}},
              {'Padding': {'load': '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'}})

packet_fragment(packet, adjust_size=True, required_size=64, fragsize=None)

Method for packet fragmentation.

Parameters:

• packet (pypacker.Packet) – Packet to be fragmented
• adjust_size (bool) – If set to True packet size will be increased to 60 bytes
• required_size (int) – Size (in bytes) of the result packet
• fragsize (int) – length of each fragment

Returns:

list of pypacker.Packets - fragments of received packet

Return type:

list[pypacker.Packet]

taf.testlib.powerboard

powerboard.py

Functionality related to Power boards which support SNMP actions

class taf.testlib.powerboard.SnmpPowerControl(config)

Bases: object

__init__(config)

Initialize SnmpPowerControl class.

get_power_status()

Get Power status of device on power board

Returns:‘On’|’Off’ Return type:(str)

power_cycle()

Perform power cycle of device

power_off()

Perform power Off of device

power_on()

Perform power On of device

snmpget(snmp_get_oid)

Returns snmpget result connected to specified port on specified host via SNMP ()

Parameters:snmp_get_oid (tuple) – SNMP OID Returns:SNMP get result Return type:(str)

snmpset(snmp_set_oid, snmp_set_value, snmp_set_type='INTEGER')

Perform snmpset for specified OID to specified value

Parameters:

• snmp_set_oid (tuple) – SNMP OID
• snmp_set_value (str) – SNMP OID
• snmp_set_type (str) – SNMP SET Data Type

taf.testlib.powerboardxmlrpc

powerboardxmlrpc.py

Functionality for power control of devices via XML-RPC connected via RaspberryPi

class taf.testlib.powerboardxmlrpc.PowerBoardXmlRpc(config)

Bases: object

Power control of devices via XML-RPC connected via RaspberryPi.

__init__(config)

Initialize PowerBoardXmlRpc class.

Parameters:config (dict) – Configuration information.

do_action(pwboard, pwport, command)

Perform specific action.

get_status(*args)

Get Power Board status.

log(action)

Function for logging actions.

power_off()

Switch off Power Board.

power_on()

Switch on Power Board.

power_reset()

Reset Power Board.

power_status()

Get Power Board status.

taf.testlib.read_csv

read_csv.py

Reading Ixia CSV files

class taf.testlib.read_csv.ReadCsv(filename)

Bases: object

Class to read Ixia CSV files.

__init__(filename)

Initialize ReadCsv class.

Parameters:filename (str) – File name. Raises:Exception – error on openning/reading csv file

get_ave_max_min(start, stop, name)

Get average, maximum, minimum cell.

Parameters:

• start (int) – Start row ID value.
• stop (int) – Stop row ID value.
• name (str) – Column name.

Returns:

average, maximum, minimum cell values

Return type:

tuple

get_cell(row, col)

Get cell.

Parameters:

• row (int) – Row ID.
• col (int) – Column ID.

Returns:

Column value

Return type:

str

get_cell_by_name(row, name)

Get cell by name.

Parameters:

• row (int) – Row ID.
• name (str) – Column name.

Returns:

Column value

Return type:

str

logger = <logging.LoggerAdapter object>

taf.testlib.remote_multicall_template

remote_multicall_template.py

taf.testlib.remote_multicall_template.call(cmd)

taf.testlib.remote_multicall_template.grouper_it(iterable, n)

taf.testlib.remote_multicall_template.main()

taf.testlib.remote_multicall_template.multicall(cmd_list, chunk)

taf.testlib.restfloodlight

restfloodlight.py

Functionality related to RestFloodlight OVS controller

class taf.testlib.restfloodlight.RestFloodlightController(ip, port)

Bases: object

RestFloodlight OVS controller.

__init__(ip, port)

Initialize RestFloodlightController class.

Parameters:

• ip (str) – Controller IP address
• port (int) – Controller port

clear(switch_id, timeout=10)

Method for clearing all flows from switch.

Parameters:

• switch_id (str) – Switch ID
• timeout (int) – reply waiting timeout

Returns:

True if flows cleared

Return type:

bool

flow_add(switch_id, command_string, name, wildcards=None, priority=32768, reply=False, timeout=30)

Method for adding flows via Floodlight Controller.

Parameters:

• switch_id (str) – Switch ID
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space
• name (str) – Flow name
• wildcards (str) – Flow wildcards
• priority (int) – Flow priority
• reply (bool) – specifies wait for reply or not
• timeout (int) – reply waiting timeout

Returns:

True if flow added

Return type:

bool

flow_delete(command, name, timeout=30)

Method for deleting flows via Floodlight Controller.

Parameters:

• command (str) – command, e.g “flow_delete”
• name (str) – Flow name
• timeout (int) – reply waiting timeout

Returns:

True if flow deleted

Return type:

bool

get_dpid(timeout=10)

Method for getting Switch dpid from Floodlight Controller.

Parameters:timeout (int) – timeout Returns:Switch dpid Return type:str

get_features(switch_id, command, reply=True, timeout=30)

Method for getting OVS Switch features via Floodlight Controller.

Parameters:

• switch_id (str) – Switch ID
• command (str) – command: “features_request”
• reply (bool) – wait for reply
• timeout (int) – reply waiting timeout

Returns:

Switch features statistics

Return type:

list

get_multiple_dpids(timeout=10)

Method for getting Switches dpids in complex setup via Floodlight Controller.

Parameters:timeout (int) – timeout Returns:Switches dpids Return type:list

get_stats(switch_id, stat_type, timeout=30)

Method for getting OVS statistics from Switch via Floodlight Controller.

Parameters:

• switch_id (str) – Switch ID
• stat_type (str) – Statistics type
• timeout (int) – reply waiting timeout

Returns:

Switch statistics

Return type:

list

probe(timeout=10)

Method for probing Floodlight Controller.

Parameters:timeout (int) – timeout

rest_call(data, action, path, timeout=10)

Method for executing call via Floodlight REST API.

Parameters:

• data (dict) – Rest data
• action (str) – Action name
• path (str) – REST path
• timeout (int) – timeout

Returns:

response.status, response.reason, response.read()

Return type:

tuple

taf.testlib.rrdtool_graph

rrdtool_graph.py

Generate commands for rrdtool graph creation

class taf.testlib.rrdtool_graph.Calculation(vname, data_defs, calc)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, vname, data_defs, calc)

Create new instance of Calculation(vname, data_defs, calc)

__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>, len=<built-in function len>)

Make a new Calculation object from a sequence or iterable

_replace(**kwds)

Return a new Calculation object replacing specified fields with new values

calc

Alias for field number 2

data_defs

Alias for field number 1

vname

Alias for field number 0

class taf.testlib.rrdtool_graph.FileInfo(file, vname, dsname)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, file, vname, dsname)

Create new instance of FileInfo(file, vname, dsname)

__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>, len=<built-in function len>)

Make a new FileInfo object from a sequence or iterable

_replace(**kwds)

Return a new FileInfo object replacing specified fields with new values

dsname

Alias for field number 2

file

Alias for field number 0

vname

Alias for field number 1

class taf.testlib.rrdtool_graph.GraphHrule(value, color)

Bases: object

Graph Hrule representation HRULE:value#color.

class taf.testlib.rrdtool_graph.LineDef(data_def, cdef, line, area, label)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, data_def, cdef, line, area, label)

Create new instance of LineDef(data_def, cdef, line, area, label)

__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>, len=<built-in function len>)

Make a new LineDef object from a sequence or iterable

_replace(**kwds)

Return a new LineDef object replacing specified fields with new values

area

Alias for field number 3

cdef

Alias for field number 1

data_def

Alias for field number 0

label

Alias for field number 4

line

Alias for field number 2

class taf.testlib.rrdtool_graph.Print(data_defs, fstring)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, data_defs, fstring)

Create new instance of Print(data_defs, fstring)

__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>, len=<built-in function len>)

Make a new Print object from a sequence or iterable

_replace(**kwds)

Return a new Print object replacing specified fields with new values

data_defs

Alias for field number 0

fstring

Alias for field number 1

class taf.testlib.rrdtool_graph.Variable(name, label)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, name, label)

Create new instance of Variable(name, label)

__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>, len=<built-in function len>)

Make a new Variable object from a sequence or iterable

_replace(**kwds)

Return a new Variable object replacing specified fields with new values

label

Alias for field number 1

name

Alias for field number 0

class taf.testlib.rrdtool_graph.XGrid(min, max, value)

Bases: tuple

__getnewargs__()

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

static __new__(_cls, min, max, value)

Create new instance of XGrid(min, max, value)

__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>, len=<built-in function len>)

Make a new XGrid object from a sequence or iterable

_replace(**kwds)

Return a new XGrid object replacing specified fields with new values

max

Alias for field number 1

min

Alias for field number 0

value

Alias for field number 2

taf.testlib.rrdtool_graph._graph(graph_vars, start, end, units_length=8, destination='/tmp/rrd.png', **kwargs)

Generate rrdtool command for graph creation.

Parameters:

• graph_vars (list) – graph calculations and data definitions
• start (int) – graph’s start time in seconds
• end (int) – graph’s end time in seconds
• units_length (int) – length of y-axis labels
• destination (str) – destination PNG file
• kwargs (dict) – additional graph configuration

Returns:

rrdtool graph command

Return type:

str

taf.testlib.rrdtool_graph.get_fetch_commands(plugin_dir, start, end, gtype='CPU')

Generate rrdtool fetch commands for RRD files from folder.

Parameters:

• plugin_dir (str) – folder with rrd files
• start (int) – graph’s start time in seconds
• end (int) – graph’s end time in seconds
• gtype (str) – rrd info type

Returns:

rrdtool fetch command

Return type:

str

taf.testlib.rrdtool_graph.get_graph_command(plugin_dir, start, end, gtype='CPU', destination='/tmp/rrd.png')

Generate rrdtool command for graph creation.

Parameters:

• plugin_dir (str) – folder with rrd files
• start (int) – graph’s start time in seconds
• end (int) – graph’s end time in seconds
• gtype (str) – rrd info type
• destination (str) – destination PNG file

Returns:

rrdtool graph command

Return type:

str

taf.testlib.sendjsoncommand

sendjsoncommand.py

JSON communication with OVS Nox Conroller functionality

class taf.testlib.sendjsoncommand.SendJsonCommand(ip, json_port)

Bases: object

Class for sending and receiving Json commands to/from OVS Nox Controller.

Parameters:

• ip (str) – Controller IP address
• json_port (int) – Controller port to send to

__init__(ip, json_port)

Initialize SendJsonCommand class.

connect(controller_ip=None, controller_port=None)

Method for connecting to Ovs Controller socket.

Parameters:

• controller_ip (str) – Controller IP address
• controller_port (int) – Controller port to send to

disconnect()

Method for disconnecting from Ovs Controller socket.

flow_add(command, command_string, reply=False, timeout=30)

Method for sending flow command to Ovs Controller (connect, send json command, disconnect).

If reply is True - wait for reply from the Controller

Parameters:

• command (str) – command, e.g “flow add”
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space
• reply (bool) – specifies wait for reply or not
• timeout (int) – timeout

flow_delete(command, command_string, reply=False, timeout=30)

Method for sending flow delete command to Ovs Controller (connect, send json command, disconnect).

If reply is True - wait for reply from the Controller

Parameters:

• command (str) – command, e.g “flow add”
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space
• reply (bool) – specifies wait for reply or not
• timeout (int) – timeout

flow_with_flags_add(command, command_string, reply=False, timeout=30)

Method for sending flow command to Ovs Controller (connect, send json command, disconnect).

If reply is True - wait for reply from the Controller

Parameters:

• command (str) – command, e.g “flow add”
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space
• reply (bool) – specifies wait for reply or not
• timeout (int) – timeout

flow_with_prio_add(command, command_string, reply=False, timeout=30)

Method for sending flow command to Ovs Controller (connect, send json command, disconnect).

If reply is True - wait for reply from the Controller

Parameters:

• command (str) – command, e.g “flow add”
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space
• reply (bool) – specifies wait for reply or not
• timeout (int) – timeout

flow_with_wildcards_add(command, command_string, reply=False, timeout=30)

Method for sending flow command to Ovs Controller (connect, send json command, disconnect).

If reply is True - wait for reply from the Controller

Parameters:

• command (str) – command, e.g “flow add”
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space
• reply (bool) – specifies wait for reply or not
• timeout (int) – timeout

get_features(command, reply=True, timeout=30)

Method for getting OVS Switch features via Nox Controller.

Parameters:

• command (str) – command, e.g “flow add”
• reply (bool) – specifies wait for reply or not
• timeout (int) – reply waiting timeout

get_stats(command, reply=True, timeout=30)

Method for getting OVS statistics from Switch via Nox Controller.

Parameters:

• command (str) – command, e.g “flow add”
• reply (bool) – specifies wait for reply or not
• timeout (int) – reply waiting timeout

probe(timeout=10)

Method for probing Nox Controller.

Parameters:timeout (int) – timeout

send(command, command_string)

Method for sending json command to Ovs Controller.

Parameters:

• command (str) – command, e.g “flow add”
• command_string (str) – command string, e.g. flow qualifiers and actions, delimited with space

set_config(command, port_no, port_config, reply=False, timeout=30)

Method for setting OVS Switch port configuration via Nox Controller.

Parameters:

• command (str) – command, e.g “flow add”
• port_no (int) – port number
• port_config (str) – port configuration
• reply (bool) – specifies wait for reply or not
• timeout (int) – reply waiting timeout

taf.testlib.si_fm10k

si_fm10k.py

FM10k silicon properties

class taf.testlib.si_fm10k.SiliconFM10K

Bases: object

ONP Red Rock Canyon/FM10K device class.

taf.testlib.si_fm6k

si_fm6k.py

FM6k silicon properties

class taf.testlib.si_fm6k.SiliconFM6K

Bases: object

ONS devices class.

taf.testlib.snmpcmd

snmpcmd.py

Module for SNMP specific functionality

class taf.testlib.snmpcmd.SNMPCmd(config, env_switches, mib_dir)

Bases: object

`SNMP specific functionality class.

Parameters:

• config (list[dict]) – environment config
• env_switches (dict) – switches dictionary in format {switch_id: switch_object}
• mib_dir (str) – MIB module name

__init__(config, env_switches, mib_dir)

Initialize SNMPCmd class

Parameters:

• config (list[dict]) – environment config
• env_switches (dict) – switches dictionary in format {switch_id: switch_object}
• mib_dir (str) – MIB module name

_find_and_load_mib(mibs_dict, sym_name)

Find MIB name and load it to MibBuilder.

Parameters:

• mibs_dict (dict) – dictionary that contains MIBs.
• sym_name (str) – MIB symbol name

Returns:

Name of MIB in which symbol name is. ‘None’ if MIB’s name wasn’t found.

Return type:

str

Examples:

self._find_and_load_mib(helpers.MIBS_DICT, 'onsSwitchppControlRouteInterfaceMtu')

_get_oid(mod_name, sym_name)

Getting values from source by SNMP.

Parameters:

• mod_name (dict) – MIB module name.
• sym_name (str) – MIB symbol name

Returns:

List of MIBs oids.

Return type:

list

Examples:

self._get_oid('ons_stat', 'onsSnmpAgentStatisticsPortId', 'tests/ui/mibs/')

_get_previous(mod_name, sym_name)

Getting OID and NodeName of previous SNMP element of sequence.

Parameters:

• mod_name (dict) – MIB module name.
• sym_name (str) – MIB symbol name

Returns:

List with OID and symbol name of previous element.

Return type:

list

Examples:

self._get_previous('ONS-SWITCH-MIB', 'onsSwitchppControlBridgeInfoInbandIpNetMaskInetAddress')

_normalize_result(mod_name, sym_name, result, to_oid=False)

Normalize SNMP GET result according syntax from MIB.

Parameters:

• mod_name (dict) – MIB module name.
• sym_name (str) – MIB symbol name
• result (list) – List with one SNMP GET result for all types except InetAddress, for InetAddress - list with two elements [‘InetAddress’, ‘InetAddressType’].
• to_oid (bool) – indicator of formatting given result for OID.

Returns:

Normalized result according to syntax.

Return type:

str, int

Examples:

self._normalize_result('ONS-SWITCH-MIB', 'onsSwitchppControlBridgeInfoInbandIpNetMaskInetAddress',
                       [OctetString(hexValue='ffffff00'), Integer(1)])

_snmp_get_call(switch_id, arguments, community, version, to_oid=False, poll_timeout=20)

Getting data from source via SNMP.

Parameters:

• switch_id (int) – ID of switch to get SNMP call to.
• arguments (list) – SNMP call (SNMP symbol name, index). Index can have inserted calls.
• community (str) – SNMP community to read.
• version (str) – version of SNMP protocol to use.
• to_oid (bool) – indicator of returned result’s OID format
• poll_timeout (int) – timeout to appearing SNMP data.

Returns:

Normalized received SNMP data.

Return type:

str, int

Examples:

self._snmp_get_call(1, ['onsSwitchppControlBridgeInfoInbandIpNetMaskInetAddress', "1"],
                        "sppCommunity", "v2", False, 20)
self._snmp_get_call(1, ['onsSwitchppControlBridgeInfoInbandIpNetMaskInetAddress', ["1.{}.1",
                        ["PortId", "1.2.3"]]], "sppCommunity", "v2", False, 20)

snmp_get(elements_list, community, version, poll_timeout=20)

Walking through list of element to get and calling self._snmp_call() method.

Parameters:

• elements_list (list) – List of (SNMP symbol name, index) pairs. Index can have inserted calls.
• community (str) – SNMP community to read.
• version (str) – version of SNMP protocol to use.
• poll_timeout (int) – timeout to appearing SNMP data.

Returns:

List of SNMP-GET command results.

Return type:

list

Examples:

self._snmp_get([{"1":[["onsSnmpAgentStatisticsPortId", "1"]]}])
self._snmp_get([{"1":[["onsSnmpAgentStatisticsPortId", ["1.{}.3", ["onsSnmpAgentStatisticsPortKey", "2.4.5"]]]]}],
               "sppCommunity", "v2")

snmp_set(elements_list, community, mib_dir=None)

Setting values by SNMP.

Parameters:

• elements_list (list) – List of (SNMP symbol name, index) pairs. Index can have inserted calls.
• community (str) – SNMP community to read.
• mib_dir (str) – MIB module name.

Returns:

List of SNMP-SET command results.

Return type:

list

Examples:

self._snmp_get(conf[test]['snmp_set'])

static snmp_walk(community, host, port, oid)

Perform SNMP walk for submitted oid.

Parameters:

• community (str) – SNMP community to read.
• host (str) – SNMP host.
• port (int) – SNMP host port.
• oid (str) – SNMP OID.

Raises:

CustomException –

suite_logger = <logging.LoggerAdapter object>

taf.testlib.snmphelpers

taf.testlib.sshtun

taf.testlib.switch_general

taf.testlib.switch_ons

taf.testlib.switch_userver

taf.testlib.syslog

taf.testlib.testenv

testenv.py

Environment verifying functionality

class taf.testlib.testenv.TestLinks(env)

Bases: object

Links verification class.

__init__(env)

Initialize TestLinks class.

Parameters:env (Environment) – Environment instance

_check_sw_links(ports, sw1, sw2, check_method='direct')

This function verifies links between Devices.

Parameters:

• ports (dict) – Ports dictionary in format {(“sw1”, “tg1”):{1: 25, 2: 26}}
• sw1 (str) – Device acronym, e.g.”sw1”
• sw2 (str) – Device acronym, e.g.”sw2”
• check_method (str) – Validation type. direct|indirect

Raises:

ValueError – unknown check_method value

Notes

Verification based on operational state change as a response to admin disable/enable on the other end of the link. (applicable only for real devices)

_check_tg_links(ports, sw)

This function verifies links between Traffic Generator and Device.

Parameters:

• ports (dict) – Ports dictionary in format {(“sw1”, “tg1”):{1: 25, 2: 26}}
• sw (str) – Device acronym, e.g.”sw1”

Notes

Verification based on STP packet “portid” field contents.

_setup()

Prepare env for test_links.

_teardown()

Check procedure on teardown.

class_logger = <logging.LoggerAdapter object>

test_links_diamond(**kwargs)

test_links_golden(**kwargs)

test_links_mixed(**kwargs)

test_links_simplified2(**kwargs)

test_links_simplified3(**kwargs)

test_links_simplified4(**kwargs)

test_links_simplified5(**kwargs)

taf.testlib.testenv.get_env_prop(env)

Read properties from all devices.

Parameters:env (Environment) – Environment instance

taf.testlib.testenv.setup_teardown(function)

Setup/Teardown decorator.

taf.testlib.tg_generators

tg_generators.py

This module contains collection of TG related generators

class taf.testlib.tg_generators.BaseGenerator(start_value, end_value, increment, count)

Bases: object

Base generator used for creating field values generators.

__init__(start_value, end_value, increment, count)

Initialize BaseGenerator class.

Parameters:

• start_value (int, str) – Generator’s start value
• end_value (int, str) – Generator’s start value
• increment (int) – Generator’s step value
• count (int) – Generator’s count value

Raises:

ValueError – start_value is mandatory

__iter__()

Return iterator object.

__next__()

Return next item from container.

class taf.testlib.tg_generators.PypackerFlowLabelGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of Flow Label generation.

Parameters:

• start_value (str) – initial Flow value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated Flow Label value.

class taf.testlib.tg_generators.PypackerIPGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of IP addresses generation.

Parameters:

• start_value (str) – initial IP value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated IP address.

class taf.testlib.tg_generators.PypackerIPv6Generator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of IPv6 addresses generation.

Parameters:

• start_value (str) – initial IPv6 value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated IPv6 address.

class taf.testlib.tg_generators.PypackerIncrementPayloadGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for incremented payload generation.

Parameters:

• start_value (str) – initial payload value
• end_value (str) – maximum payload value
• increment (int) – incrementation step

__next__()

Get next generated payload value.

class taf.testlib.tg_generators.PypackerLspIdGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of LSP IDs one byte generation.

Parameters:

• start_value (str) – initial LSP ID value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated MAC address.

class taf.testlib.tg_generators.PypackerMacGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of MAC addresses generation.

Parameters:

• start_value (str) – initial MAC value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated MAC address.

class taf.testlib.tg_generators.PypackerProtocolGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of protocols generation.

Parameters:

• start_value (str) – initial protocol value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated protocol value.

class taf.testlib.tg_generators.PypackerRandomPayloadGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for random payload generation.

Parameters:

• start_value (str) – initial payload value
• end_value (str) – maximum payload value

__next__()

Get next generated payload value.

class taf.testlib.tg_generators.PypackerTCPOrUDPGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of UDP/TCP addresses generation.

Parameters:

• start_value (str) – initial UDP/TCP value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated UDP address.

class taf.testlib.tg_generators.PypackerTypeGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of types generation.

Parameters:

• start_value (str) – initial Ether.type value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next type generation value.

class taf.testlib.tg_generators.PypackerVlanGenerator(start_value, end_value, increment, count)

Bases: taf.testlib.tg_generators.BaseGenerator

Iteration class for list of VLANs generation.

Parameters:

• start_value (str) – initial Vlan value
• increment (int) – incrementation step
• count (int) – number of iteration steps

__next__()

Get next generated VLAN.

taf.testlib.tg_helpers

tg_helpers.py

TGHelperMixin class for mixin for tg object

class taf.testlib.tg_helpers.TGHelperMixin

Bases: object

Mixin class for switch.tg

arp_packets_sending(env, tg_port, sw_port, packet_num, ipgap, arp_packet=None, offset=0, retry=False)

check_traffic(env, packets, ports, tg_indexes, rate=10, bi_dir=True, time_run=30)

Send traffic: packet_1, packet_2 between port_1 and port_2, with tg indexes: ix_1, ix_2.

class_logger = <logging.LoggerAdapter object>

emulate_isis_neighbor_on_port(tg_port, mac_address=None, system_id=None, mtu=None, ipv4_address=None, ipv6_address=None, hostname=None, isis_emul_hosts=1, ipgap=1, wait_for_packets=10)

Emulation of IS-IS router on specified TG’s port.

Parameters:

• tg_port (str) – Name of TG port for emulation
• mac_address (str) – Source MAC address for IS-IS PDUs
• system_id (str) – System ID for IS-IS PDUs in format “xxxx.xxxx.xxxx”
• ipv4_address (str) – IPv4 address to announce as management IPv4 address
• ipv6_address (str) – IPv6 address for announcement
• hostname (str) – Hostname for announcement
• isis_emul_hosts (int) – number of ISIS Hosts to emulate
• ipgap (int) – rate of ISIS LSP ID’s packets to emulate
• wait_for_packets (int) – Time in seconds for sniffing packets

Returns:

IS-IS neighbor parameters that are propagated in IS-IS packets

e.g. MAC address or area ID.

Return type:

dict

Notes

The second IS-IS neighbor emulation on other port leads to link state changes and updates (LSPs) sent to already existing neighbors. LSPs require confirmations. The correct way seems to be running an interactive sniffer in separate thread and confirm every received LSP by an appropriate PSNP/CSNP. But current IXIA TAF implementation doesn’t support threading, so for now, in terms of compatibility, this approach isn’t implemented.

isis_packets_sending(env, tg_port, switch_id_port, mac_address=None, ipgap=1.0, isis_nodes_cnt=5, mtu=1500)

Emulation of IS-IS DCRP node on specified TG’s port.

Parameters:

• env (Environment) – environment data
• tg_port (str) – Name of TG port for emulation
• switch_id_port (tuple(str, str)) – Switch id and port id to set port UP
• mac_address (str) – Source MAC address for IS-IS PDUs
• ipgap (int) – rate of ISIS LSP ID’s packets to emulate
• isis_nodes_cnt (int) – Number of ISIS DCRP nodes to emulate
• mtu (int) – MTU of

Returns:

IS-IS neighbor parameters that are propagated in IS-IS packets

e.g. MAC address or area ID.

Return type:

dict

Notes

The second IS-IS neighbor emulation on other port leads to link state changes and updates (LSPs) sent to already existing neighbors. LSPs require confirmations. The correct way seems to be running an interactive sniffer in separate thread and confirm every received LSP by an appropriate PSNP/CSNP. But current IXIA TAF implementation doesn’t support threading, so for now, in terms of compatibility, this approach isn’t implemented.

table_9_test_preparation(env, tg_port, sw_port, packet_num, ipgap, offset=0, arp_packet=None)

Prepare ports, packets for table 9 related tests.

taf.testlib.tg_template

tg_template.py

Interface class for TG entries. This file contain description for public TG object API

class taf.testlib.tg_template.GenericTG(config, opts)

Bases: taf.testlib.entry_template.GenericEntry

Traffic Generator interface class.

DEFAULT_MAX_SNIFF_TIME = 3600

check()

Check if TG object is alive and ready for processing.

Returns:None or raise and exception.

cleanup(mode='complete')

This method should do Ixia ports cleanup (remove streams etc).

Parameters:mode (str) – “fast” or “complete”. If mode == “fast”, method does not clear streams on the port, but stops them (str). Returns:None or raise and exception.

clear_statistics(sniff_port_list)

Clear statistics - number of frames.

Parameters:sniff_port_list (list) – List of interface names. Returns:None

clear_streams()

Stop and clear all traffic streams.

connect_port(iface)

Simulate port link connecting (set it to admin up etc).

Parameters:iface (str) – Interface to connect. Raises:NotImplementedError – not implemented Returns:None or raise and exception.

create()

Perform all necessary procedures to initialize TG device and prepare it for interaction.

Returns:None or raise and exception.

Notes

Method has to check –get_only option.

Set of steps to configure TG device is related to particular TG type.

destroy()

Perform all necessary procedures to uninitialize TG device.

Returns:None or raise and exception.

Notes

Method has to check –get_only and –leave_on options. Set of steps to unconfigure TG device is related to particular TG type. Method has to clear all connections and stop all captures and data streams.

disconnect_port(iface)

Simulate port link disconnecting (set it to admin down etc).

Parameters:iface (str) – Interface to disconnect. Raises:NotImplementedError – not implemented Returns:None or raise and exception.

get_filtered_frames_count(iface)

Read statistics - number of received frames which fit filter criteria.

Parameters:iface (str) – Interface name. Returns:Number of filtered frames. Return type:int

get_os_mtu(iface=None)

Get MTU value in host OS.

Parameters:iface (str) – Interface for getting MTU in host OS Returns:Original MTU value Return type:int

Examples:

env.tg[1].get_os_mtu(iface=ports[('tg1', 'sw1')][1])

get_port_qos_rxrate(iface, qos)

Return port receiving rate for specific qos.

Parameters:

• iface (str) – Interface name.
• qos (int) – Qos value.

Returns:

Frames per second (int)

Return type:

int

get_port_rxrate(iface)

Return port receiving rate.

Parameters:iface (str) – Interface name. Returns:Frames per second. Return type:int

get_port_txrate(iface)

Return port transmission rate.

Parameters:iface (str) – Interface name. Returns:Frames per second. Return type:int

get_qos_frames_count(iface, prio)

Get captured QoS frames count.

Parameters:

• iface (str) – Interface name.
• prio (int) – Priority.

Returns:

captured QoS frames count.

Return type:

int

get_received_frames_count(iface)

Read statistics - number of received valid frames.

Parameters:iface (str) – Interface name. Returns:Number of received frames. Return type:int

get_sent_frames_count(iface)

Read statistics - number of sent frames.

Parameters:iface (str) – Interface name. Returns:Number of sent frames. Return type:int

get_uds_3_frames_count(iface)

Read statistics - number of non-filtered received frames (valid and invalid).

Parameters:iface (str) – Interface name. Returns:Number of received frames Return type:int

iface_config(port, *args, **kwargs)

High-level interface config utility.

Parameters:port (str) – Interface name Raises:NotImplementedError – not implemented

Note

This method has to support parameters supported by ::ixia::interface_config function for compatibility. You have to check already implemented parameters for other TG types.

Examples:

env.tg[1].iface_config(tgport1, autonegotiation=1, duplex="auto", speed="auto",
                       intf_ip_addr="10.1.0.101", gateway="10.1.0.1", netmask="255.255.255.0",
                       src_mac_addr="0000.0a01.0065")
env.tg[1].iface_config(tgport2, autonegotiation=1, duplex="auto", speed="auto",
                       intf_ip_addr="40.0.0.2", gateway="40.0.0.1", netmask="255.255.255.0",
                       src_mac_addr="0000.2801.0065",
                       connected_count=increment_count, gateway_step='0.0.0.0')

sanitize()

This method has to clear all stuff which can cause device inconsistent state after exit or unexpected exception.

Notes

E.g. clear connections, stop threads. This method is called from pytest.softexit

send_stream(stream_id)

Sends the stream created by ‘set_stream’ method.

Parameters:stream_id (int) – ID of the stream to be send. Returns:timestamp. Return type:float

set_flow_control(iface, mode)

Enable/Disable flow control on the port.

Parameters:

• iface (str) – Interface name.
• mode (bool) – True/False.

Returns:

None

set_os_mtu(iface=None, mtu=None)

Set MTU value in host OS.

Parameters:

• iface (str) – Interface for changing MTU in host OS
• mtu (int) – New MTU value

Returns:

Original MTU value

Return type:

int

Examples:

env.tg[1].set_os_mtu(iface=ports[('tg1', 'sw1')][1], mtu=1650)

set_qos_stat_type(iface, ptype)

Set the QoS counters to look for priority bits for given packets type.

Parameters:

• iface (str) – Interface name.
• ptype (str) – Priority type: VLAN/IP.

Returns:

None

set_stream(packet_def=None, count=None, inter=0, rate=99, continuous=False, iface=None, adjust_size=True, required_size=64, fragsize=None, build_packet=True, is_valid=False, sa_increment=None, da_increment=None, sip_increment=None, dip_increment=None, arp_sa_increment=None, arp_sip_increment=None, igmp_ip_increment=None, lldp_sa_increment=None, vlan_increment=None, sudp_increment=None, dudp_increment=None, eth_type_increment=None, dscp_increment=None, protocol_increment=None, sipv6_increment=None, dipv6_increment=None, fl_increment=None, dhcp_si_increment=None, in_vlan_increment=None, tc_increment=None, nh_increment=None, isis_lspid_increment=None, cont_burst=False, force_errors=None, udf_dependancies=None)

Set traffic stream with specified parameters on specified TG port.

Parameters:

• packet_def (tuple(dict{dict})) – Packet definition. Tuple of dictionaries of dictionaries in format: ({layerX: {field1: value, field2: value}, {layerY: {field1:value, fieldN: value}})
• count (int) – How many packets to send in a stream.
• inter (int) – Interval between sending each packet.
• rate (int) – Interface rate in percents.
• continuous (bool) – Should stream be sent continuously or not. Continuous streams have to be started using start_streams method.
• iface (str, tuple) – Interface to use for packet sending (type depends on particular tg ports type).
• adjust_size (bool) – See description for _build_pypacker_packet function.
• required_size (int, tuple) – Integer or tuple of parameters needed to be set when packet size should be incremented. Tuple examples: (‘Increment’, <step>, <min>, <max>), (‘Random’, <min>, <max>)
• fragsize (int) – Max size of packet’s single frame
• is_valid (bool) – Recalculate check sum and length for each packet layer (by default pypacker do this automatically in case length and check sum aren’t set). This parameter has to be set True with all incrementation parameters.
• build_packet (bool) – Build packet from definition or use already built pypacker packet.
• sa_increment (tuple) – Source MAC increment parameters. Tuple (<step>, <count>). Use count=0 for continuous increment.
• da_increment (tuple) – Destination MAC increment parameters. Tuple (<step>, <count>). Use count=0 for continuous increment.
• sip_increment (tuple) – Source IPv4 increment parameters. Tuple (<step>, <count>). Use count=0 for continuous increment.
• dip_increment (tuple) – Destination IPv4 increment parameters. Tuple (<step>, <count>). Use count=0 for continuous increment.
• arp_sa_increment (tuple) – Source MAC increment parameters for ARP packet. Tuple (<step>, <count>). Has to be used in pair with arp_sip_increment.
• arp_sip_increment (tuple) – Source IP increment parameters for ARP packet. Tuple (<step>, <count>). Has to be used in pair with arp_sa_increment.
• igmp_ip_increment (tuple) – Destination IP increment parameters for IGMP packet. Tuple (<step>, <count>).
• lldp_sa_increment (tuple) – Source MAC increment parameters for LLDP packet. Tuple (<step>, <count>).
• vlan_increment (tuple) – VLAN increment parameters for tagged packet. Tuple (<step>, <count>).
• sudp_increment (tuple) – UDP source port increment parameters.
• dudp_increment (tuple) – UDP destination port increment parameters.
• eth_type_increment (tuple) – Ethernet frame type increment parameters.
• dscp_increment (tuple) – DSCP increment parameters.
• protocol_increment (tuple) – IP protocol incrementation..
• sipv6_increment (tuple) – Source IPv6 increment parameters.
• dipv6_increment (tuple) – Destination IPv6 increment parameters.
• fl_increment (tuple) – Flow label increment parameters.
• dhcp_si_increment (tuple) – DHCP IP increment parameters.
• in_vlan_increment (tuple) – Inner vlan ID increment parameters for double tagged frames. Tuple (<step>, <count>).
• tc_increment (tuple) – IPv6 Traffic Class increment parameters.
• nh_increment (tuple) – IPv6 Next Header increment parameters.
• cont_burst (bool) – Should stream be sent as continuous burst or not. Continuous streams have to be started using start_streams method.
• force_errors (str) – Emulate Errors for configured stream. Enum (“bad” /*streamErrorBadCRC, “none” /*streamErrorNoCRC, “dribble” /*streamErrorDribble, “align” /*streamErrorAlignment).
• udf_dependancies (dict) – Set UDF dependencies in case one incerement is dependant from another. Dictionary {<dependant_increment> : <initial_increment>}

Returns:

stream id

Return type:

int

Notes

It’s not expected to configure a lot of incrementation options. Different traffic generator could have different limitations for these options.

Examples:

stream_id_1 = tg.set_stream(pack_ip, count=100, iface=iface)
stream_id_2 = tg.set_stream(pack_ip, continuous=True, inter=0.1, iface=iface)
stream_id_3 = tg.set_stream(pack_ip_udp, count=5, protocol_increment=(3, 5), iface=iface)
stream_id_4 = tg.set_stream(pack_ip_udp, count=18, sip_increment=(3, 3), dip_increment=(3, 3), iface=iface,
                            udf_dependancies={'sip_increment': 'dip_increment'})

start_sniff(ifaces, sniffing_time=None, packets_count=0, filter_layer=None, src_filter=None, dst_filter=None)

Starts sniffing on specified interfaces.

Parameters:

• ifaces (list) – List of TG interfaces for capturing.
• sniffing_time (int) – Time in seconds for sniffing.
• packets_count (int) – Count of packets to sniff (no count limitation in case 0).
• filter_layer (str) – Name of predefined sniffing filter criteria.
• src_filter (str) – Sniff only packet with defined source MAC.
• dst_filter (str) – Sniff only packet with defined destination MAC.

Returns:

None

Notes

This method introduces additional 1.5 seconds timeout after capture enabling. It’s required by Ixia sniffer to wait until capturing is started.

Examples:

env.tg[1].start_sniff(['eth0', ], filter_layer='IP', src_filter='00:00:00:01:01:01', dst_filter='00:00:00:22:22:22')

start_streams(stream_list)

Enable and start streams from the list simultaneously.

Parameters:stream_list (list[int]) – List of stream IDs. Returns:None

stop_sniff(ifaces, force=False, drop_packets=False, sniff_packet_count=1000)

Stops sniffing on specified interfaces and returns captured data.

Parameters:

• ifaces (list) – List of interfaces where capturing has to be stopped.
• force (bool) – Stop capturing even if time or packet count criteria isn’t achieved.
• drop_packets (bool) – Don’t return sniffed data (used in case you need just read statistics).
• sniff_packet_count (int) – Default number of packets to return (used to avoid test hanging in case storm).

Returns:

Dictionary where key = interface name, value = list of sniffed packets.

Return type:

dict

stop_streams(stream_list=None)

Disable streams from the list.

Parameters:stream_list (list[int]) – Stream IDs to stop. In case stream_list is not set all running streams will be stopped. Returns:None

taf.testlib.ui_dcrp_shell

ui_dcrp_shell.py

DCRP Shell UI specific functionality

class taf.testlib.ui_dcrp_shell.UiDcrpShell(dcrp_domain)

Bases: taf.testlib.ui_helpers.UiHelperMixin

UI class for DCRP domain.

Notes

When UiInterface class from ui_wrapper has abstraction methods, this class should also inherit from it.

__getattr__(item)

Override all not implemented in this class UI methods’ calls by calling them on all UI instances in parallel.

Parameters:item (str) – Name of called and not found in class item. Raises:AttributeError – error if UIs don’t have called method.

__init__(dcrp_domain)

Initiate UiDcrpShell class.

Parameters:dcrp_domain (SwitchDcrpDomain instance) – DCRP Domain instance

_get_ui_instance(node_id)

Get UI instance of specified node.

Parameters:node_id (int | str) – ID of node for getting it’s UI instance Raises:UIException – Error if wrong node_id was given Returns:UI instance of the given node Return type:Device UI instance

check_dcrpd_service(instance=None, services=None)

Check status of DCRP services on the specified node.

Parameters:

• instance (UI instance) – UI instance to restart DCRP services with
• services (list[str]) – List of services’ names

Returns:

Dictionary with service names as keys and bool status as values

Return type:

dict

check_isis_nodes_discovery(**kwargs)

class_logger = <logging.LoggerAdapter object>

configure_dcrpd(node_id, mesh_ports=None, cpu_mac_address=None, file_name=None)

Configure mesh ports for using by DCRP services.

Edit DCRP service configuration file (“/etc/dcrpd.conf” by default) by adding given ports as mesh ports. Set given MAC address for given ports and bring them UP.

Parameters:

• node_id (int | str) – node ID for configuring on
• mesh_ports (dict[dict]) – Dictionary with mesh port names as keys and dictionary with additional parameters, such as port MAC address, as values.
• cpu_mac_address (str) – MAC address to be set to CPU port. Format: “FF:FF:FF:FF:FF:FF”
• file_name (str) – DCRP configuration file name. If omitted self.dcrp_conf_file will be used.

configure_mlag(uplink_port, lag_mac, team_name=None, file_name=None)

Configure DCRP MLAG on specified node.

Edit DCRP service configuration file (“/etc/dcrpd.conf” by default) by adding given port(s) as uplink ports. Set given MAC address as lag mac address.

Parameters:

• uplink_port (list[str]) –

  List of ports to configure them as uplink ports. Ports should be in format “node_id port_id”. Examples:

  ["0013 10", "0014 20"]
  

• lag_mac (str) – MAC address to configure as MLAG MAC address. Same for each node
• team_name (str) – MLAG interface name, skip if None
• file_name (str) – DCRP configuration file name. If omitted self.dcrp_conf_file is used.

create_vlan_ports(ports=None, vlans=None, tagged='Tagged')

Wrapper for create_vlan_ports UI method. Method also adds VLAN to CPU and mesh ports.

Parameters:

• ports (list[str]) –

  List of ports or None. Ports should be in format “node_id port_id”. Examples:

  ["0013 10", "0014 20"]  # Mandatory parameter.
  

• vlans (list[int] | set(int)) – list of VLAN IDs. Mandatory parameter.
• tagged (str) – port tagging attribute

Raises:

UIException – Error if not all mandatory parameters are specified.

get_isis_neighbors(**kwargs)

get_isis_topology(**kwargs)

get_macs(**kwargs)

get_node_hostname(**kwargs)

static get_ports_map(ports_list)

Convert list of ports from format [“0013 10”, “0013 13”, “0014 15”, “0015 15”] to {“0013”: [10, 13], “0014”: [15], “0015”: [15]]

Parameters:ports_list (list[str]) – List of port to be converted in format [“5555 1”, “7777 2”] Returns:Dictionary with node IDs as keys and lists of port IDs as values Return type:dict

get_table_platform()

Get ‘Platform’ table.

get_table_ports(ports=None, all_params=False, ip_addr=False)

Wrapper for get_table_ports UI method.

Parameters:

• ports (list[str]) –

  List of ports or None. Ports should be in format “node_id port_id”. Example:

  ["0013 10", "0014 20"]
  

• all_params (bool) – get additional port properties
• ip_addr (bool) – Get IP address

static get_tg_ports_of_node(env, node_id, links_count=1)

Get tg ports connected to specified node.

Parameters:

• env (object) – Environment
• node_id (str) – Node id. E.g.: ‘4615’
• links_count (int) – Required links count for specific node

Returns:

TG instances and related tg/switch ports connected between each other.

E.g.: ‘tg’: {1: <dev_obj1>, 2: <dev_obj1>}, ‘tg_ports’: {1: (1,6,5), 2:(1,6,6)}, ‘sw_ports’: {1: 13, 2: 21}}

Return type:

dict

modify_ports(ports, expected_rcs=frozenset({0}), **kwargs)

Wrapper for modify_ports UI method.

Parameters:

• ports (list[str]) –

  List of ports or None. Ports should be in format “node_id port_id”. Examples:

  ["0013 10", "0014 20"]
  

• expected_rcs (int | list | set | frozenset) – expected return code

static parse_isis_table_neighbor(neighbor_table)

Parse ‘show isis neighbor’ table.

Parameters:neighbor_table (list[str] | iter()) – List of ‘show isis neighbor’ raw output Returns:A dictionary containing the system_id, interface, level, state, hold_time, snpa of each neighbor. Return type:iter()

static parse_isis_table_topology(topology_table)

Parse ‘show isis topology’ table.

Parameters:topology_table (list[str] | iter()) – List of ‘show isis topology’ raw output Returns:

A dictionary containing the vertex, type, metric, next_hop, interface and

parent values for each destination node

Return type:iter()

restart_dcrpd_service()

Wrapper for self.start_dcrpd_service method with True “restart” parameter.

set_age_time_out(**kwargs)

set_all_ports_admin_disabled(**kwargs)

start_dcrpd_service(**kwargs)

stop_dcrpd_service(**kwargs)

update_remote_config(node_id, parameters, file_name, clean=False)

Edit or add parameters in remote configuration file which contains “key = value” pairs.

Parameters:

• node_id (int | str) – node ID for configuring on
• parameters (dict) – Dictionary with key: value pair for editing or adding to remote configuration file
• file_name (str) – Full name (path + name) of remote configuration file
• clean (bool) – Empty configuration file before editing.

wait_all_ports_admin_disabled(**kwargs)

wait_for_port_value_to_change(ports, port_parameter, value, interval=1, timeout=30)

Wrapper for waiting for port value to be changed.

Parameters:

• ports (list[str]) –

  List of ports or None. Ports should be in format “node_id port_id”. Examples:

  ["0013 10", "0014 20"]
  

• port_parameter (str) – Parameter name to be checked
• value (int | str) – Parameter value to be checked
• interval (int) – How often parameter should be checked (seconds)
• timeout (int) – Time for checking value

Raises:

StandardError –

Returns:

None

wait_on_dcrpd_service(instance=None, services=None, timeout=45)

Wait for DCRP services are launched.

Parameters:

• instance (UI instance) – UI instance to wait DCRP services on
• services (list[str]) – List of services’ names
• timeout (int) – Timeout for waiting

taf.testlib.ui_dcrp_shell.in_parallel(func)

Decorator function. Runs decorated function in parallel for all nodes.

Decorated function must receive named parameter ‘instance’.

taf.testlib.ui_helpers

ui_helpers.py

UiHelper class for mixin for switch.ui object

class taf.testlib.ui_helpers.UiHelperMixin

Bases: object

Mixing class for switch.ui.

add_entry_to_fdb_ufd(vlan_id, fdb_entries)

Add static MAC to FDB.

Parameters:

• vlan_id (int) – vlan id to be used
• fdb_entries (list[tuple(int, str)]) – port and mac details that neeed to be added to FDB

Returns:

None

build_and_create_ufd_network_file(port_type, ports, bind_carrier='')

Creating network file for uplink and downlink ports.

Parameters:

• port_type (str) – type of interface (uplink/downlink)
• ports (list[int | str]) – ports to assign to UFD group
• bind_carrier (str | list[int]) – which uplink ports are bound to specified downlink ports

Returns:

None

clear_lag_table()

Removes all ports from LAG and clears LAG table.

compose_unique_mac_addr(pckt_type='unicast', prefix=None, ports=None)

Creates unique mac addresses for given ports. Each addres is concatenation of packet type speecific prefix, switch own IP address and port number.

Parameters:

• pckt_type (str) – type of packet: ‘unicast’, ‘multicast’ or ‘broadcast’
• prefix (str) – custom packet prefix consisting of two octets,eg ‘01:80’, ignored if pck_type == ‘broadcast’
• ports (list) – list of port names or numbers

Returns:

map of port names and mac adresses

delete_static_macs_from_port(port)

Deletes all static MAC addresses from port.

Parameters:port (str | int) – port

get_and_validate_statistics_ufd(ingress_ports, egress_ports, tg_instance, sniff_params, time_out=30)

Get the port statistics and validate the counters.

Parameters:

• ingress_ports (list[tuple(int, built-in function)]) – List of ingress ports that need to be validated
• egress_ports (list[tuple(int, built-in function)]) – List of egress ports that need to be validated
• tg_instance (instance object) – traffic generator instance object
• sniff_params (dict{(int, int, int) – list[tuple(str, str, bool)]}): parms used for validating packets in the TG
• time_out (int) – Time out required for the counters to get updated

Returns:

None

is_entry_added_to_ports2vlan_table(port_id, vlan_id, tagged='Tagged', pvid=None, table_ports2vlan=None)

Check if entry is added to Ports2Vlan table.

Parameters:

• port_id (int) – port number
• vlan_id (int) – vlan number
• tagged (str) – tagged or untagged
• pvid (bool) – true or false, if inputted vlan is pvid
• table_ports2vlan (list[dict]) – table of ports2vlan

Returns:

True or False

Return type:

bool

Examples:

is_entry_added_to_ports2vlan_table(port_id=port_id, vlan_id=vlan_id, pvid=True, tagged=tagged)

is_entry_added_to_vlan_table(vlan_id=1)

Check if entry is added to VLAN table.

Parameters:vlan_id (int) – vlan number from where packet was sent (integer) Returns:True or False Return type:bool

Examples:

is_entry_added_to_vlan_table(vlan_id=vlan_id)

is_lag_added(lag_id)

Check if lag has been added to LAG table.

Parameters:lag_id (str|int) – id of lag Returns:bool

is_port_added_to_lag(port, lag_id)

Check if port added to LAG.

Parameters:

• port (int) – port
• lag_id (str|int) – id of lag

Returns:

bool

set_admin_mode_for_slave_ports(admin_mode='Down')

Set adminMode for logical ports.

Parameters:admin_mode (str) – Ports adminMode Returns:True or raise exception

Examples:

assert ui_helpers.set_admin_mode_for_slave_ports(admin_mode="Up")

start_traffic_ufd(tg_instance, port_list)

Start sending traffic to the ports.

Parameters:

• tg_instance (instance object) – TG instance object
• port_list (list[dict]) – List of interfaces to which traffic needs to be send

Returns:

list of stream ids generated

Return type:

list[int]

ui_raises(method, *args, **values)

UI raises.

Parameters:method (str) – method to call

wait_for_port_status(lag_id, state, value, interval)

Wait for LAG/port state to become value.

Parameters:

• lag_id (int | str) – LAG/port id
• state (str) – state
• value (int | str) – expected value
• interval (int) – timeout

Raises:

SwitchException –

Returns:

None

wait_for_port_value_to_change(ports, port_parameter, value, interval=1, timeout=30)

Wait until value is changed in port table.

Parameters:

• ports (list[int | str]) – list of ports
• port_parameter (str) – port parameter
• value (int | str) – checking value
• timeout (int) – timeout

Raises:

StandardError –

Returns:

None

wait_for_state_lag_state(lag=None, port=1, state='Selected', timeout=30)

Wait until port state in RSTP table.

wait_until_ops_state(port=1, state='Up', timeout=30)

Obsoleted function. Use wait_until_value_is_changed with proper option instead.

wait_until_stp_param(mode='STP', port=1, param='rootGuard', value='Disabled', timeout=30, instance=0)

Wait until port role in xSTP table.

taf.testlib.ui_onpss_jsonrpc

taf.testlib.ui_ons_cli

taf.testlib.ui_ons_xmlrpc

taf.testlib.ui_wrapper

taf.testlib.virtual_env

taf.testlib.xmlrpc_proxy

xmlrpc_proxy.py

Implementation of xmlrpclib.ServerProxy class with timeout option

class taf.testlib.xmlrpc_proxy.CliMarshaller(encoding=None, allow_none=False)

Bases: xmlrpc.client.Marshaller

Convert INT-64 value to XML-RPC chunk.

Notes

No direct calls supposed.

dispatch = {<class 'list'>: <function Marshaller.dump_array>, '_arbitrary_instance': <function Marshaller.dump_instance>, <class 'int'>: <function CliMarshaller.dump_int>, <class 'datetime.datetime'>: <function Marshaller.dump_datetime>, <class 'bool'>: <function Marshaller.dump_bool>, <class 'xmlrpc.client.Binary'>: <function Marshaller.dump_instance>, <class 'bytes'>: <function Marshaller.dump_bytes>, <class 'float'>: <function Marshaller.dump_double>, <class 'xmlrpc.client.DateTime'>: <function Marshaller.dump_instance>, <class 'bytearray'>: <function Marshaller.dump_bytes>, <class 'str'>: <function Marshaller.dump_unicode>, <class 'dict'>: <function Marshaller.dump_struct>, <class 'tuple'>: <function Marshaller.dump_array>, <class 'NoneType'>: <function Marshaller.dump_nil>}

dump_i8(value, write)

Override base class method.

Notes

Allow processing INT-64 values in CLI.If passed value could not be converted by original method, try to convert it using ‘dump_i8’ method.

dump_int(value, write)

Override base class method.

Notes

Allow processing INT-64 values in CLI.If passed value could not be converted by original method, try to convert it using ‘dump_i8’ method.

dump_int_orig(value, write)

dump_long(value, write)

Override base class method.

Notes

Allow processing INT-64 values in CLI.If passed value could not be converted by original method, try to convert it using ‘dump_i8’ method.

dump_long_orig(value, write)

class taf.testlib.xmlrpc_proxy.TimeoutHTTPConnection(host, timeout=10)

Bases: http.client.HTTPConnection

Timeout HTTP connection class definition.

__init__(host, timeout=10)

Initialize TimeoutHTTPConnection class.

class taf.testlib.xmlrpc_proxy.TimeoutServerProxy(uri, timeout=180, *args, **kwargs)

Bases: xmlrpc.client.ServerProxy

xmlrpclib.ServerProxy class with additional timeout option

__init__(uri, timeout=180, *args, **kwargs)

Initialize TimeoutServerProxy class.

class taf.testlib.xmlrpc_proxy.TimeoutTransport(timeout=10, *args, **kwargs)

Bases: xmlrpc.client.Transport

Timeout Transport class definition.

__init__(timeout=10, *args, **kwargs)

Initialize TimeoutTransport class.

make_connection(host)

Configure connection.

class taf.testlib.xmlrpc_proxy._Method(send, name)

Bases: testlib.xmlrpc_proxy._Method

_Method class definition.

__call__(*args)

Configuring calls.

class_logger = <logging.LoggerAdapter object>

taf.testlib.xmltodict

exception taf.testlib.xmltodict.ParsingInterrupted

Bases: Exception

taf.testlib.xmltodict.parse(xml_input, encoding='utf-8', expat=<module 'xml.parsers.expat' from '/home/docs/.pyenv/versions/3.5.7/lib/python3.5/xml/parsers/expat.py'>, *args, **kwargs)

Parse the given XML input and convert it into a dictionary.

xml_input can either be a string or a file-like object.

If xml_attribs is True, element attributes are put in the dictionary among regular child elements, using @ as a prefix to avoid collisions. If set to False, they are just ignored.

Simple example:

>>> doc = xmltodict.parse("""

… <a prop=”x”> … <b>1</b> … <b>2</b> … </a> … “”“) >>> doc[‘a’][‘@property’] u’x’ >>> doc[‘a’][‘b’] [u‘1’, u‘2’]

If item_depth is 0, the function returns a dictionary for the root element (default behavior). Otherwise, it calls item_callback every time an item at the specified depth is found and returns None in the end (streaming mode).

The callback function receives two parameters: the path from the document root to the item (name-attribs pairs), and the item (dict). If the callback’s return value is false-ish, parsing will be stopped with the ParsingInterrupted exception.

Streaming example:

>>> def handle(path, item):

… print ‘path:%s item:%s’ % (path, item) … return True … >>> xmltodict.parse(“”” … <a prop=”x”> … <b>1</b> … <b>2</b> … </a>”“”, item_depth=2, item_callback=handle) path:[(u’a’, {u’prop’: u’x’}), (u’b’, None)] item:1 path:[(u’a’, {u’prop’: u’x’}), (u’b’, None)] item:2

The optional argument postprocessor is a function that takes path, key and value as positional arguments and returns a new (key, value) pair where both key and value may have changed. Usage example:

>>> def postprocessor(path, key, value):

… try: … return key + ‘:int’, int(value) … except (ValueError, TypeError): … return key, value >>> xmltodict.parse(‘<a><b>1</b><b>2</b><b>x</b></a>’, … postprocessor=postprocessor) OrderedDict([(u’a’, OrderedDict([(u’b:int’, [1, 2]), (u’b’, u’x’)]))])

You can pass an alternate version of expat (such as defusedexpat) by using the expat parameter. E.g:

>>> import defusedexpat
>>> xmltodict.parse('<a>hello</a>', expat=defusedexpat.pyexpat)
OrderedDict([(u'a', u'hello')])

taf.testlib.xmltodict.unparse(item, output=None, encoding='utf-8', **kwargs)