noc.core.script.base

Module Contents

class noc.core.script.base.BaseScriptMetaclass

Bases: type

Process @match decorators

classmethod __new__(mcs, name, bases, attrs)
class noc.core.script.base.BaseScript(service, credentials, args=None, capabilities=None, version=None, parent=None, timeout=None, name=None, session=None, session_idle_timeout=None)

Bases: six.with_metaclass()

Service Activation script base class

name
TIMEOUT
SESSION_IDLE_TIMEOUT
DEFAULT_ACCESS_PREFERENCE = SC
cache = False
interface
requires = []
base_logger
_x_seq
session_lock
session_cli
session_mml
session_rtsp
reuse_cli_session = True
keep_cli_session = True
matchers
ScriptError
CLISyntaxError
CLIOperationError
NotSupportedError
UnexpectedResultError
hexbin
cli_protocols
mml_protocols
rtsp_protocols
always_prefer
rx_detect_sep
root

Get root script

motd

Return message of the day

_match_lines_cache
is_beefed
__call__(self, *args, **kwargs)
apply_matchers(self)

Process matchers and apply is_XXX properties :return:

clean_input(self, args)

Cleanup input parameters against interface

clean_output(self, result)

Clean script result against interface

run(self)

Run script

classmethod compile_match_filter(cls, *args, **kwargs)

Compile arguments into version check function Returns callable accepting self and version hash arguments

classmethod match(cls, *args, **kwargs)

execute method decorator

match_version(self, *args, **kwargs)

inline version for BaseScript.match

execute(self, **kwargs)

Default script behavior: Pass through _execute_chain and call appropriate handler

call_method(self, cli_handler=None, snmp_handler=None, fallback_handler=None, **kwargs)

Call function depending on access_preference :param cli_handler: String or callable to call on CLI access method :param snmp_handler: String or callable to call on SNMP access method :param fallback_handler: String or callable to call if no access method matched :param kwargs: :return:

execute_cli(self, **kwargs)

Process script using CLI :param kwargs: :return:

execute_snmp(self, **kwargs)

Process script using SNMP :param kwargs: :return:

cleaned_config(self, config)

Clean up config from all unnecessary trash

strip_first_lines(self, text, lines=1)

Strip first lines

expand_rangelist(self, s)

Expand expressions like “1,2,5-7” to [1, 2, 5, 6, 7]

expand_interface_range(self, s)

Convert interface range expression to a list of interfaces “Gi 1/1-3,Gi 1/7” -> [“Gi 1/1”, “Gi 1/2”, “Gi 1/3”, “Gi 1/7”] “1:1-3” -> [“1:1”, “1:2”, “1:3”] “1:1-1:3” -> [“1:1”, “1:2”, “1:3”] :param s: Comma-separated list :return:

macs_to_ranges(self, macs)

Converts list of macs to rangea :param macs: Iterable yielding mac addresses :returns: [(from, to), ..]

hexstring_to_mac(self, s)

Convert a 6-octet string to MAC address

get_cache(self, key1, key2)

Get cached result or raise KeyError

set_cache(self, key1, key2, value)

Set cached result

configure(self)

Returns configuration context

cached(self)

Return cached context managed. All nested CLI and SNMP GET/GETNEXT calls will be cached.

Usage:

with self.cached():
self.cli(“…..) self.scripts.script()
enter_config(self)

Enter configuration mote

leave_config(self)

Leave configuration mode

save_config(self, immediately=False)

Save current config

schedule_to_save(self)
set_motd(self, motd)

Match s against regular expression rx using re.search Raise UnexpectedResultError if regular expression is not matched. Returns match object. rx can be string or compiled regular expression

re_match(self, rx, s, flags=0)

Match s against regular expression rx using re.match Raise UnexpectedResultError if regular expression is not matched. Returns match object. rx can be string or compiled regular expression

classmethod match_lines(cls, rx, s)
find_re(self, iter, s)

Find first matching regular expression or raise Unexpected result error

hex_to_bin(self, s)

Convert hexadecimal string to boolean string. All non-hexadecimal characters are ignored :param s: Input string :return: Boolean string :rtype: str

push_prompt_pattern(self, pattern)
pop_prompt_pattern(self)
has_oid(self, oid)

Check object responses to oid

get_timeout(self)
cli(self, cmd, command_submit=None, bulk_lines=None, list_re=None, cached=False, file=None, ignore_errors=False, allow_empty_response=True, nowait=False, obj_parser=None, cmd_next=None, cmd_stop=None)

Execute CLI command and return result. Initiate cli session when necessary :param cmd: CLI command to execute :param command_submit: :param bulk_lines: :param list_re: :param cached: :param file: :param ignore_errors: :param allow_empty_response: Allow empty output. If False - ignore prompt and wait output :type allow_empty_response: bool :param nowait:

Execute CLI command and return a result. if list_re is None, return a string if list_re is regular expression object, return a list of dicts (group name -> value),

one dict per matched line
get_cli_stream(self)
close_cli_stream(self)
close_snmp(self)
mml(self, cmd, **kwargs)

Execute MML command and return result. Initiate MML session when necessary :param cmd: :param kwargs: :return:

get_mml_stream(self)
close_mml_stream(self)
rtsp(self, method, path, **kwargs)

Execute RTSP command and return result. Initiate RTSP session when necessary :param method: :param path: :param kwargs: :return:

get_rtsp_stream(self)
close_rtsp_stream(self)
close_current_session(self)
classmethod close_session(cls, session_id)

Explicit session closing :return:

get_access_preference(self)
get_always_preferred(self)

Return always preferred access method :return:

has_cli_access(self)
has_snmp_access(self)
has_cli_only_access(self)
has_snmp_only_access(self)
has_snmp(self)

Check whether equipment has SNMP enabled

has_snmp_v1(self)
has_snmp_v2c(self)
has_snmp_v3(self)
has_snmp_bulk(self)

Check whether equipment supports SNMP BULK

has_capability(self, capability, allow_zero=False)

Check whether equipment supports capability

ignored_exceptions(self, iterable)

Context manager to silently ignore specified exceptions

iter_pairs(self, g, offset=0)

Convert iterable g to a pairs i.e. [1, 2, 3, 4] -> [(1, 2), (3, 4)] :param g: Iterable :param offset: Skip first recirds :return:

to_reuse_cli_session(self)
to_keep_cli_session(self)
start_tracking(self)
stop_tracking(self)
push_cli_tracking(self, r, state)
push_snmp_tracking(self, oid, tlv)
iter_cli_tracking(self)

Yields command, packets for collected data :return:

iter_cli_fsm_tracking(self)
request_beef(self)

Download and return beef :return:

class noc.core.script.base.ScriptsHub(script)

Bases: object

Object representing Script.scripts structure. Returns initialized child script which can be used ans callable

class _CallWrapper(script_class, parent)

Bases: object

__call__(self, **kwargs)
__getattr__(self, item)
__contains__(self, item)

Check object has script name

class noc.core.script.base.PartialResult

Bases: object

__slots__ = ['result']
__int__(self, **kwargs)