unicon package¶

Subpackages¶

Submodules¶

unicon.logs module¶

Logging module for Unicon

class unicon.logs.ColorFormatter(fmt='\r\n%(asctime)s: %%UNICON-%(levelname)s: %(message)s', date_fmt='%Y-%m-%dT%H:%M:%S')¶

Bases: logging.Formatter

For formatting the console logs with colors for log levels.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{‘ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record)¶

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class unicon.logs.ExpectStreamFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)¶

Bases: logging.Formatter

For formatting the console logs with colors for expect internal log levels.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{‘ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record)¶

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class unicon.logs.TrimLineFormatter(fmt='\r\n%(asctime)s: %%UNICON-%(levelname)s: %(message)s')¶

Bases: logging.Formatter

Remove extra ‘rn’ in the log.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{‘ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record)¶

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class unicon.logs.UniconFileHandler(filename, mode='a', encoding=None, delay=False, errors=None)¶

Bases: logging.FileHandler, unicon.logs.UniconStreamHandler

Handler for logging message to file

Open the specified file and use it as the stream for logging.

emit(record)¶

Emit a record.

If the stream was not opened because ‘delay’ was specified in the constructor, open it before calling the superclass’s emit.

class unicon.logs.UniconScreenHandler(stream=None)¶

Bases: unicon.logs.UniconStreamHandler

Initialize the handler.

If stream is not specified, sys.stderr is used.

property stream¶

class unicon.logs.UniconStreamHandler(stream=None)¶

Bases: logging.StreamHandler

Initialize the handler.

If stream is not specified, sys.stderr is used.

emit(record)¶

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

end_stream()¶

streaming_state = False¶

class unicon.logs.Unicon_pyATS_TaskLog_Adapter¶

Bases: unicon.logs.UniconStreamHandler

Initialize the handler.

If stream is not specified, sys.stderr is used.

property stream¶

unicon.logs.color_str(color, data)¶

unicon.logs.remove_unicon_log_messages(log_text)¶: %UNICON message lines

unicon.patterns module¶

A basic database of patterns

class unicon.patterns.UniconCorePatterns¶: Bases: object

unicon.settings module¶

class unicon.settings.Settings¶

Bases: unicon.bases.settings.BaseSettings

GRACEFUL_DISCONNECT_WAIT_SEC = 1¶

POST_DISCONNECT_WAIT_SEC = 10¶

SUBCONNECTION_SETTINGS_NAME = '_subconnection_settings'¶

show()¶

update(settings)¶: Pass a dict object to update settings.

unicon.type_checkers module¶

This module implements data type checks for most common types and unicon data structures. It uses the descriptor feature to implement the type checking

class unicon.type_checkers.BooleanOrNoneType(name)¶: Bases: unicon.type_checkers.BooleanType

class unicon.type_checkers.BooleanType(name)¶: Bases: unicon.type_checkers.TypeDescriptor

class unicon.type_checkers.IntegerType(name)¶

Bases: unicon.type_checkers.TypeDescriptor

check for integer type

class unicon.type_checkers.PositiveIntegerOrNoneType(name)¶

Bases: unicon.type_checkers.PositiveIntegerType

check if number is a positive integer

class unicon.type_checkers.PositiveIntegerType(name)¶

Bases: unicon.type_checkers.IntegerType

check if number is a positive integer

class unicon.type_checkers.StringOrNoneType(name)¶

Bases: unicon.type_checkers.TypeDescriptor

check for string type or None

class unicon.type_checkers.StringType(name)¶

Bases: unicon.type_checkers.TypeDescriptor

check for string type

class unicon.type_checkers.TypeDescriptor(name, dtype, none=False)¶: Bases: object

unicon.utils module¶

All unicon utilities here go here ….

class unicon.utils.AttributeDict(d={})¶: Bases: dict

class unicon.utils.Utils¶

Bases: object

clear_line(terminal_server, port, username='lab', password='lab', enable_password='lab', timeout=100)¶

Utils method to perform clear line operation on the terminal server

Parameters

terminal_server (str) – ip address to the terminal server
port (int) – the line number to clear, usually this is the last two digits of the port number of the telnet connection.
username (str) – terminal server login username
password (str) – terminal server login password
enable_password (str) – terminal server enable password
timeout (int) – timeout value for clear line

Returns

None

get_ats_log()¶

get_redundancy_details(connection, timeout=None, who='my')¶

output_block_extract(data, block)¶: Parse the show redundancy state output blocks.

remove_ansi_escape_codes(ansi_string)¶: Remove ANSI escape codes.

remove_backspace(string)¶

Remove backspace characters and any chars before the backspace.

Returns the modified string.

remove_backspace_ansi_escape(string)¶

truncate_trailing_prompt(con_state, result, hostname=None, result_match=None)¶

Design intent is to take a raw output (typically with the prepended echoed command already removed) and strip out the trailing prompt, returning only the wanted content.

If last result_match is passed in, it will be used first to truncate.

If the plugin designer sets up their state prompt correctly (Here, correctly means State prompt must start with (.*?), to match all data before prompt.), then it becomes possible to more readily determine what part of the output is wanted and which part belongs to the prompt. The first match group of state pattern considered as output.

If the state prompt has not been set up correctly (Here, not correctly means, if pattern does not find first match group), a best-effort algorithm is used to attempt to truncate the trailing prompt from the output(Only if hostname is passed), but this may not work correctly in all situations (especially for commands that themselves contain the hostname, premature output truncation may result).

If the hostname has not been set, and if the router prompt does not include the hostname then prompt removal is not possible and the output is returned unchanged.

Arguments: current state object , command output , hostname, result_match truncate_state_pattern(enable_state, result, dev.hostname, result_match)

unicon.utils.clear_line_handler(spawn, context)¶

unicon.utils.connection_refused_to_terminal(spawn)¶: handles connection refused scenarios

unicon.utils.invalid_login(spawn)¶

unicon.utils.learn_os_platform(con, output, os_map)¶: learn device os and platform from device output

unicon.utils.recursive_update(a, b)¶: Recursively updates on nested dictionary a with dictionary b.

unicon.utils.update_connection_kwargs(con)¶: update connection kwargs for learn_os redirect connection

unicon.sshutils module¶

class unicon.sshutils.SSHTunnel¶

Bases: object

Class to manage SSH tunnels via external ssh client

Attributes

proxy_conn: dict with Connection objects indexed by proxy hostname

tunnel_port: current tunnel port number

TUNNEL_RETRIES = 10¶

add_tunnel(proxy_conn=None, tunnel_port=0, tunnel_type='local', tunnel_protocol='tcp', target_address='127.0.0.1', target_port='22')¶

Add SSH tunnel via an existing proxy connection object.

Parameters

proxy_conn (Connection) – Unicon Connection object with active connection to the SSH tunnel host.
tunnel_port (int) – tunnel port to use, 0 will auto select tunnel port, default: 0
tunnel_type (str) – SSH tunnel type (local|remote), default: local
target_address (str) – target IP address, default: 127.0.0.1
target_port (str|int) – target TCP port, default: 22

Example

Import sshtunnel object from the module and use the method to add tunnels via the proxy connection object.

from unicon.sshutils import sshtunnel

tunnel_port = sshtunnel.add_tunnel(proxy_conn=proxy,
                                   target_address='127.0.0.1',
                                   target_port=10001)

auto_tunnel_add(device, via, tunnel_type=None, tunnel_protocol='tcp')¶

Add a SSH tunnel based on the connection for a device.

This method is intended to be used by modules that implement a pyATS connection class.

Parameters

device (Device) – pyATS Device object
via (str) – device connection name
tunnel_type (str) – type of SSH tunnel (remote|local) default: local
tunnel_protocol (string) – protocol to tunnel (tcp|udp*) default: tcp * UDP is not yet supported

Returns

tunnel port number

Example

tunnel_port = sshtunnel.auto_tunnel_add(device, via)
if tunnel_port:
    ip = connection_block[via].sshtunnel.tunnel_ip
    port = tunnel_port
else:
    ip = connection_block[via].ip
    port = connection_block[via].get('port')

connect_proxy_host(testbed, proxy_host)¶

Connect to the proxy_host defined in the testbed object.

This method adds the Connection object to the proxy_connections attribute, e.g. self.proxy_connections[proxy_host] = Connection(..)

This method is intended to be used by the auto_tunnel_add method.

Parameters

testbed (Testbed) – pyATS testbed object
proxy_host (str) – name of the proxy device

Example

# testbed.yaml
testbed:
  servers:
    js:
      address: 127.0.0.1
      credentials:
          ssh:
              username: cisco
              password: cisco
      custom:
        port: 2200
        ssh_options: -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null

>>> sshtunnel.connect_proxy_host(testbed=device.testbed, proxy_host=proxy_host)

remove_tunnel(proxy_conn=None, tunnel_type='local', tunnel_port=None)¶

Remove SSH tunnel via an existing proxy connection object.

Parameters

proxy_conn (Connection) – Unicon Connection object with active connection to the SSH tunnel host.
tunnel_type (str) – SSH tunnel type (local|remote)
tunnel_port (int) – tunnel port to cancel

exception unicon.sshutils.SSHTunnelError¶

Bases: Exception

Exception class for SSH tunnel class

unicon package¶

Subpackages¶

Submodules¶

unicon.logs module¶

unicon.patterns module¶

unicon.settings module¶

unicon.type_checkers module¶

unicon.utils module¶

unicon.sshutils module¶

Module contents¶