[Python-modules-commits] [pyiosxr] 01/03: Import pyiosxr_0.21.orig.tar.gz
Vincent Bernat
bernat at moszumanska.debian.org
Tue Nov 1 18:49:37 UTC 2016
This is an automated email from the git hooks/post-receive script.
bernat pushed a commit to branch master
in repository pyiosxr.
commit 94c8f497d259c07e782b04c744bb052cce10a1ff
Author: Vincent Bernat <bernat at debian.org>
Date: Tue Nov 1 19:46:28 2016 +0100
Import pyiosxr_0.21.orig.tar.gz
---
PKG-INFO | 4 +-
pyIOSXR.egg-info/PKG-INFO | 4 +-
pyIOSXR.egg-info/requires.txt | 3 +-
pyIOSXR/__init__.py | 5 +
pyIOSXR/exceptions.py | 85 +++-
pyIOSXR/iosxr.py | 469 ++++++++++++------
requirements.txt | 3 +-
setup.py | 21 +-
test/test.py | 1063 ++++++++++++++++++++++++++---------------
9 files changed, 1107 insertions(+), 550 deletions(-)
diff --git a/PKG-INFO b/PKG-INFO
index 93eaa0b..54457de 100644
--- a/PKG-INFO
+++ b/PKG-INFO
@@ -1,12 +1,12 @@
Metadata-Version: 1.1
Name: pyIOSXR
-Version: 0.14
+Version: 0.21
Summary: Python API to interact with network devices running IOS-XR
Home-page: https://github.com/fooelisa/pyiosxr/
Author: Elisa Jasinska
Author-email: elisa at bigwaveit.org
License: UNKNOWN
-Download-URL: https://github.com/fooelisa/pyiosxr/tarball/0.14
+Download-URL: https://github.com/fooelisa/pyiosxr/tarball/0.21
Description: UNKNOWN
Keywords: IOS-XR,IOSXR,Cisco,networking
Platform: UNKNOWN
diff --git a/pyIOSXR.egg-info/PKG-INFO b/pyIOSXR.egg-info/PKG-INFO
index 93eaa0b..54457de 100644
--- a/pyIOSXR.egg-info/PKG-INFO
+++ b/pyIOSXR.egg-info/PKG-INFO
@@ -1,12 +1,12 @@
Metadata-Version: 1.1
Name: pyIOSXR
-Version: 0.14
+Version: 0.21
Summary: Python API to interact with network devices running IOS-XR
Home-page: https://github.com/fooelisa/pyiosxr/
Author: Elisa Jasinska
Author-email: elisa at bigwaveit.org
License: UNKNOWN
-Download-URL: https://github.com/fooelisa/pyiosxr/tarball/0.14
+Download-URL: https://github.com/fooelisa/pyiosxr/tarball/0.21
Description: UNKNOWN
Keywords: IOS-XR,IOSXR,Cisco,networking
Platform: UNKNOWN
diff --git a/pyIOSXR.egg-info/requires.txt b/pyIOSXR.egg-info/requires.txt
index 808fb07..64230ca 100644
--- a/pyIOSXR.egg-info/requires.txt
+++ b/pyIOSXR.egg-info/requires.txt
@@ -1 +1,2 @@
-pexpect
+netmiko>=0.5.2
+lxml>=3.2.4
diff --git a/pyIOSXR/__init__.py b/pyIOSXR/__init__.py
index 0c5f704..a4f7af2 100644
--- a/pyIOSXR/__init__.py
+++ b/pyIOSXR/__init__.py
@@ -1,4 +1,9 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""A module to interact with Cisco devices running IOS-XR."""
+
# Copyright 2015 Netflix. All rights reserved.
+# Copyright 2016 BigWaveIT. All rights reserved.
#
# The contents of this file are licensed under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with the
diff --git a/pyIOSXR/exceptions.py b/pyIOSXR/exceptions.py
index d19abf5..2b83b56 100644
--- a/pyIOSXR/exceptions.py
+++ b/pyIOSXR/exceptions.py
@@ -1,4 +1,9 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""Exceptions for pyiosxr, a module to interact with Cisco devices running IOS-XR."""
+
# Copyright 2015 Netflix. All rights reserved.
+# Copyright 2016 BigWaveIT. All rights reserved.
#
# The contents of this file are licensed under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with the
@@ -12,20 +17,86 @@
# License for the specific language governing permissions and limitations under
# the License.
-class UnknownError(Exception):
+
+class IOSXRException(Exception):
+
+
+ def __init__(self, msg=None, dev=None):
+
+ super(Exception, self).__init__(msg)
+ if dev:
+ self._xr = dev
+ # release the XML agent
+ self._xr._xml_agent_acquired = False
+
+
+class ConnectError(IOSXRException):
+ """Exception while openning the connection."""
+
+ pass
+
+
+class CommitError(IOSXRException):
+
+ """Raised when unable to commit. Mostly due to ERROR 0x41866c00"""
+
pass
-class InvalidInputError(Exception):
+
+class LockError(IOSXRException):
+ """Throw this exception when unable to lock the config DB."""
+
pass
-class XMLCLIError(Exception):
+
+class UnlockError(IOSXRException):
+ """Throw this exception when unable to unlock the config DB."""
+
pass
-class TimeoutError(Exception):
+
+class CompareConfigError(IOSXRException):
+ """Throw this exception when unable to compare config."""
+
pass
-class EOFError(Exception):
+
+class UnknownError(IOSXRException):
+ """UnknownError Exception."""
+
pass
-class IteratorIDError(Exception):
- pass
+
+class InvalidInputError(IOSXRException):
+ """InvalidInputError Exception."""
+
+ pass
+
+
+class XMLCLIError(IOSXRException):
+ """XMLCLIError Exception."""
+
+ pass
+
+
+class InvalidXMLResponse(IOSXRException):
+ """Raised when unable to process properly the XML reply from the device."""
+
+ pass
+
+class TimeoutError(IOSXRException):
+ """TimeoutError Exception."""
+
+ pass
+
+
+class EOFError(IOSXRException):
+ """EOFError Exception."""
+
+ pass
+
+
+class IteratorIDError(IOSXRException):
+ """IteratorIDError Exception."""
+
+ pass
diff --git a/pyIOSXR/iosxr.py b/pyIOSXR/iosxr.py
index 4bfd3a9..b55dfaa 100644
--- a/pyIOSXR/iosxr.py
+++ b/pyIOSXR/iosxr.py
@@ -1,4 +1,6 @@
+# -*- coding: utf-8 -*-
# Copyright 2015 Netflix. All rights reserved.
+# Copyright 2016 BigWaveIT. All rights reserved.
#
# The contents of this file are licensed under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with the
@@ -12,202 +14,384 @@
# License for the specific language governing permissions and limitations under
# the License.
+"""Contains the main IOS-XR driver class."""
+
+# stdlib
import re
-import sys
+import time
import difflib
-import pexpect
-from exceptions import XMLCLIError, InvalidInputError, TimeoutError, EOFError, IteratorIDError
-import xml.etree.ElementTree as ET
+# third party lib
+from lxml import etree as ET
+from netmiko import ConnectHandler
+from netmiko.ssh_exception import NetMikoTimeoutException
+from netmiko.ssh_exception import NetMikoAuthenticationException
+# local modules
+from exceptions import LockError
+from exceptions import UnlockError
+from exceptions import XMLCLIError
+from exceptions import CommitError
+from exceptions import ConnectError
+from exceptions import TimeoutError
+from exceptions import IteratorIDError
+from exceptions import InvalidInputError
+from exceptions import CompareConfigError
+from exceptions import InvalidXMLResponse
-# Build and execute xml requests.
-def __execute_rpc__(device, rpc_command, timeout):
- rpc_command = '<?xml version="1.0" encoding="UTF-8"?><Request MajorVersion="1" MinorVersion="0">'+rpc_command+'</Request>'
- try:
- device.sendline(rpc_command)
- index = device.expect_exact(["</Response>","ERROR: 0xa240fe00"], timeout = timeout)
- if index == 1:
- raise XMLCLIError('The XML document is not well-formed')
- except pexpect.TIMEOUT as e:
- raise TimeoutError("pexpect timeout error")
- except pexpect.EOF as e:
- raise EOFError("pexpect EOF error")
-
- #remove leading XML-agent prompt
- response_assembled = device.before+device.match
- response = re.sub('^[^<]*', '', response_assembled)
-
- root = ET.fromstring(response)
- if 'IteratorID' in root.attrib:
- raise IteratorIDError("Non supported IteratorID in Response object. \
-Turn iteration off on your XML agent by configuring 'xml agent [tty | ssl] iteration off'. \
-For more information refer to http://www.cisco.com/c/en/us/td/docs/ios_xr_sw/iosxr_r4-1/xml/programming/guide/xl41apidoc.pdf, \
-7-99.Turn iteration off on your XML agent.")
-
- childs = [x.tag for x in list(root)]
-
- result_summary = root.find('ResultSummary')
-
- if result_summary is not None and int(result_summary.get('ErrorCount', 0)) > 0:
- if 'CLI' in childs:
- error_msg = root.find('CLI').get('ErrorMsg') or ''
- elif 'Commit' in childs:
- error_msg = root.find('Commit').get('ErrorMsg') or ''
- else:
- error_msg = root.get('ErrorMsg') or ''
+# ~~~ all three functions below should be deprecated and completely removed in the next releases ~~~
+####################################################################################################
+# anyway they are supposed to be private module functions
- error_msg += '\nOriginal call was: %s' % rpc_command
- raise XMLCLIError(error_msg)
- if 'CLI' in childs:
- cli_childs = [x.tag for x in list(root.find('CLI'))]
- if 'Configuration' in cli_childs:
- output = root.find('CLI').find('Configuration').text
- if output is None:
- output = ''
- elif 'Invalid input detected' in output:
- raise InvalidInputError('Invalid input entered:\n%s' % output)
+def __execute_rpc__(device, rpc_command, timeout):
+ return device._execute_rpc(rpc_command)
- return root
-# Ecexute show commands not in config context.
def __execute_show__(device, show_command, timeout):
- rpc_command = '<CLI><Exec>'+show_command+'</Exec></CLI>'
- response = __execute_rpc__(device, rpc_command, timeout)
- return response.find('CLI').find('Exec').text.lstrip()
+ return device._execute_show(show_command)
+
-# Ecexute show commands not in config context.
def __execute_config_show__(device, show_command, timeout):
- rpc_command = '<CLI><Configuration>'+show_command+'</Configuration></CLI>'
- response = __execute_rpc__(device, rpc_command, timeout)
- return response.find('CLI').find('Configuration').text.lstrip()
+ return device._execute_config_show(show_command)
+####################################################################################################
+
+class IOSXR(object):
-class IOSXR:
+ _ITERATOR_ID_ERROR_MSG = (
+ 'Non supported IteratorID in Response object.'
+ 'Turn iteration off on your XML agent by configuring "xml agent [tty | ssl] iteration off".'
+ 'For more information refer to'
+ 'http://www.cisco.com/c/en/us/td/docs/ios_xr_sw/iosxr_r4-1/xml/programming/guide/xl41apidoc.pdf, 7-99.'
+ 'Please turn iteration off for the XML agent.'
+ )
+ """
+ Establishes a connection with the IOS-XR device via SSH and facilitates the communication through the XML agent.
+ """
def __init__(self, hostname, username, password, port=22, timeout=60, logfile=None, lock=True):
"""
- A device running IOS-XR.
+ IOS-XR device constructor.
- :param hostname: (str) IP or FQDN of the device you want to connect to
+ :param hostname: (str) IP or FQDN of the target device
:param username: (str) Username
:param password: (str) Password
:param port: (int) SSH Port (default: 22)
:param timeout: (int) Timeout (default: 60 sec)
:param logfile: File-like object to save device communication to or None to disable logging
- :param lock: (bool) Auto-lock config upon open() if set to True, connect without locking if False (default: True)
+ :param lock: (bool) Auto-lock config upon open() if set to True, connect without locking if False
+ (default: True)
"""
self.hostname = str(hostname)
self.username = str(username)
self.password = str(password)
- self.port = int(port)
- self.timeout = int(timeout)
- self.logfile = logfile
+ self.port = int(port)
+ self.timeout = int(timeout)
+ self.logfile = logfile
self.lock_on_connect = lock
- self.locked = False
+ self.locked = False
+ self._cli_prompt = None
+ self._xml_agent_acquired = False
def __getattr__(self, item):
"""
- Ok, David came up with this kind of dynamic method. It takes
- calls with show commands encoded in the name. I'll replacs the
- underscores for spaces and issues the show command... pretty neat!
+ Dynamic getter to translate generic show commands.
+
+ David came up with this dynamic method. It takes
+ calls with show commands encoded in the name. I'll replace the
+ underscores for spaces and issues the show command on the device...
+ pretty neat!
non keyword params for show command:
all non keyword arguments is added to the command to allow dynamic parameters:
- eks: .show_interface("GigabitEthernet0/0/0/0")
+ eg: .show_interface("GigabitEthernet0/0/0/0")
keyword params for show command:
config=True/False : set True to run show command in config mode
- eks: .show_configuration_merge(config=True)
+ eg: .show_configuration_merge(config=True)
"""
- def wrapper(*args, **kwargs):
+ def _getattr(*args, **kwargs):
+
cmd = item.replace('_', ' ')
for arg in args:
cmd += " %s" % arg
if kwargs.get("config"):
- response = __execute_config_show__(self.device, cmd, self.timeout)
+ response = self._execute_config_show(cmd)
else:
- response = __execute_show__(self.device, cmd, self.timeout)
+ response = self._execute_show(cmd)
- match = re.search(".*(!! IOS XR Configuration.*)</Exec>",response,re.DOTALL)
+ match = re.search(".*(!! IOS XR Configuration.*)</Exec>", response, re.DOTALL)
if match is not None:
response = match.group(1)
return response
if item.startswith('show'):
- return wrapper
+ return _getattr
else:
raise AttributeError("type object '%s' has no attribute '%s'" % (self.__class__.__name__, item))
def make_rpc_call(self, rpc_command):
"""
- Allow a user to query a device directly using XML-requests
+ Allow a user to query a device directly using XML-requests.
+
+ :param rpc_command: (str) rpc command such as:
+ <Get><Operational><LLDP><NodeTable></NodeTable></LLDP></Operational></Get>
"""
- result = __execute_rpc__(self.device, rpc_command, self.timeout)
+ result = self._execute_rpc(rpc_command)
return ET.tostring(result)
def open(self):
"""
- Opens a connection to an IOS-XR device.
+ Open a connection to an IOS-XR device.
+
+ Connects to the device using SSH and drops into XML mode.
"""
- device = pexpect.spawn('ssh -o ConnectTimeout={} -p {} {}@{}'.format(self.timeout, self.port, self.username, self.hostname), logfile=self.logfile)
try:
- index = device.expect(['\(yes\/no\)\?', 'password:', '#', pexpect.EOF], timeout = self.timeout)
- if index == 0:
- device.sendline('yes')
- index = device.expect(['\(yes\/no\)\?', 'password:', '#', pexpect.EOF], timeout = self.timeout)
- if index == 1:
- device.sendline(self.password)
- elif index == 3:
- pass
- if index != 2:
- device.expect('#', timeout = self.timeout)
- device.sendline('xml')
- index = device.expect(['XML>', 'ERROR: 0x24319600'], timeout = self.timeout)
- if index == 1:
- raise XMLCLIError('XML TTY agent has not been started. Please configure \'xml agent tty\'.')
- except pexpect.TIMEOUT as e:
- raise TimeoutError("pexpect timeout error")
- except pexpect.EOF as e:
- raise EOFError("pexpect EOF error")
- self.device = device
+ self.device = ConnectHandler(device_type='cisco_xr',
+ ip=self.hostname,
+ port=self.port,
+ username=self.username,
+ password=self.password)
+ self.device.timeout = self.timeout
+ except NetMikoTimeoutException as t_err:
+ raise ConnectError(t_err.message)
+ except NetMikoAuthenticationException as au_err:
+ raise ConnectError(au_err.message)
+
+ self._cli_prompt = self.device.find_prompt()
+
+ self._enter_xml_mode()
+
+ def _enter_xml_mode(self):
+
+ try:
+ out = self._send_command('xml') # enter in XML mode
+ except TimeoutError as terr:
+ raise ConnectError('Cannot connect to the XML agent. Enabled?', self)
+
if self.lock_on_connect:
self.lock()
+ def _timeout_exceeded(self, start, msg='Timeout exceeded!'):
+
+ if time.time() - start > self.timeout:
+ # it timeout exceeded, throw TimeoutError
+ raise TimeoutError(msg, self)
+
+ return False
+
+ def _send_command(self, command, delay_factor=.1, receive=False, start=None, expect_string=r'XML>'):
+
+ output = ''
+
+ if not receive:
+ start = time.time()
+ # because the XML agent is able to process only one single request over the same SSH session at a time
+ # first come first served
+ while self._xml_agent_acquired and not self._timeout_exceeded(start, 'Waiting to acquire the XML agent!'):
+ # will wait here till the XML agent is ready to receive new requests
+ # if stays too much, _timeout_exceeded will raise TimeoutError
+ time.sleep(delay_factor) # rest a bit
+ self._xml_agent_acquired = True
+ try:
+ output = self.device.send_command_expect(command,
+ expect_string=expect_string,
+ strip_prompt=False,
+ strip_command=False,
+ delay_factor=delay_factor,
+ max_loops=1500)
+ except IOError as ioe:
+ if self._timeout_exceeded(start):
+ time.sleep(delay_factor) # go sleep a bit, you still got time
+ return self._send_command(command, receive=True, start=start) # let's try receiving more
+ else:
+ output = self._netmiko_recv() # try to read some more
+
+ if '0xa3679e00' in output:
+ # when multiple parallel request are made, the device throws the error:
+ # ERROR: 0xa3679e00 'XML Service Library' detected the 'fatal' condition
+ # 'Multiple concurrent requests are not allowed over the same session.
+ # A request is already in progress on this session.'
+ # we could use a mechanism similar to NETCONF and push the requests in queue and serve them sequentially
+ # BUT we are not able to assign unique IDs and identify the request-reply map
+ # so will throw an error that does not help too much :(
+ raise XMLCLIError('XML agent cannot process parallel requests!', self)
+
+ if not output.strip().endswith('XML>'):
+ if '0x44318c06' in output or (self._cli_prompt and expect_string != self._cli_prompt and \
+ (output.startswith(self._cli_prompt) or output.endswith(self._cli_prompt))):
+ # sometimes the device throws a stupid error like:
+ # ERROR: 0x44318c06 'XML-TTY' detected the 'warning' condition
+ # 'A Light Weight Messaging library communication function returned an error': No such device or address
+ # and the XML agent connection is closed, but the SSH connection is fortunately maintained
+ # OR sometimes, the device simply exits from the XML mode without any clue
+ # In both cases, we need to re-enter in XML mode...
+ # so, whenever the CLI promt is detected, will re-enter in XML mode
+ # unless the expected string is the prompt
+ self._xml_agent_acquired = False # release the channel
+ self._enter_xml_mode()
+ # however, the command could not be executed properly, so we need to raise the XMLCLIError exception
+ raise XMLCLIError('Could not properly execute the command. Re-entering XML mode...', self)
+ if not output.strip(): # empty output, means that the device did not start delivering the output
+ if not self._timeout_exceeded(start):
+ time.sleep(delay_factor) # go sleep a bit, you still got time
+ return self._send_command(command, receive=True, start=start) # let's try receiving more
+ raise XMLCLIError(output.strip(), self)
+
+ self._xml_agent_acquired = False # release the XML agent
+ return str(output.replace('XML>', '').strip())
+
+ def _netmiko_recv(self, max_loops=1500):
+
+ output = ''
+
+ for tmp_output in self.device.receive_data_generator():
+ output += tmp_output
+
+ return output
+
+ # previous module function __execute_rpc__
+ def _execute_rpc(self, command_xml, delay_factor=.1):
+
+ xml_rpc_command = '<?xml version="1.0" encoding="UTF-8"?><Request MajorVersion="1" MinorVersion="0">' \
+ + command_xml + '</Request>'
+
+ response = self._send_command(xml_rpc_command, delay_factor=delay_factor)
+
+ try:
+ root = ET.fromstring(response)
+ except ET.XMLSyntaxError as xml_err:
+ if 'IteratorID="' in response:
+ raise IteratorIDError(self._ITERATOR_ID_ERROR_MSG, self)
+ raise InvalidXMLResponse('Unable to process the XML Response from the device!', self)
+
+ if 'IteratorID' in root.attrib:
+ raise IteratorIDError(self._ITERATOR_ID_ERROR_MSG, self)
+
+ childs = [x.tag for x in list(root)]
+
+ result_summary = root.find('ResultSummary')
+
+ if result_summary is not None and int(result_summary.get('ErrorCount', 0)) > 0:
+
+ if 'CLI' in childs:
+ error_msg = root.find('CLI').get('ErrorMsg') or ''
+ elif 'Commit' in childs:
+ error_msg = root.find('Commit').get('ErrorMsg') or ''
+ error_code = root.find('Commit').get('ErrorCode') or ''
+ if error_code == '0x41866c00':
+ # yet another pointless IOS-XR error:
+ # if the config DB was changed by another process,
+ # while the current SSH connection is established and alive,
+ # we won't be able to commit and the device will throw the following error:
+ # 'CfgMgr' detected the 'warning' condition
+ # 'One or more commits have occurred from other configuration sessions since this session started
+ # or since the last commit was made from this session.'
+ # dumb.
+ # in this case we need to re-open the connection with the XML agent
+ self.discard_config() # discard candidate config
+ try:
+ # exiting from the XML mode
+ self._send_command('exit', expect_string=self._cli_prompt)
+ except XMLCLIError:
+ pass # because does not end with `XML>`
+ self._enter_xml_mode() # re-entering XML mode
+ raise CommitError(
+ error_msg + '\nPlease reload the changes and try committing again!',
+ self
+ )
+ elif error_code == '0x41864e00' or error_code == '0x43682c00':
+ # raises this error when the commit buffer is empty
+ raise CommitError('The target configuration buffer is empty.')
+
+ else:
+ error_msg = root.get('ErrorMsg') or ''
+
+ error_msg += '\nOriginal call was: %s' % xml_rpc_command
+ raise XMLCLIError(error_msg, self)
+
+ if 'CLI' in childs:
+ cli_childs = [x.tag for x in list(root.find('CLI'))]
+ if 'Configuration' in cli_childs:
+ output = root.find('CLI').find('Configuration').text
+ elif 'Exec' in cli_childs:
+ output = root.find('CLI').find('Exec').text
+ if output is None:
+ output = ''
+ elif 'Invalid input detected' in output:
+ raise InvalidInputError('Invalid input entered:\n%s' % output, self)
+
+ return root
+
+ # previous module function __execute_show__
+ def _execute_show(self, show_command):
+ """
+ Executes an operational show-type command.
+ """
+ rpc_command = '<CLI><Exec>'+show_command+'</Exec></CLI>'
+ response = self._execute_rpc(rpc_command)
+ raw_response = response.xpath('.//CLI/Exec')[0].text
+ return raw_response.strip() if raw_response else ''
+
+ # previous module function __execute_config_show__
+ def _execute_config_show(self, show_command, delay_factor=.1):
+ """
+ Executes a configuration show-type command.
+ """
+ rpc_command = '<CLI><Configuration>'+show_command+'</Configuration></CLI>'
+ response = self._execute_rpc(rpc_command, delay_factor=delay_factor)
+ raw_response = response.xpath('.//CLI/Configuration')[0].text
+ return raw_response.strip() if raw_response else ''
+
def close(self):
"""
- Closes the connection to the IOS-XR device.
+ Close the connection to the IOS-XR device.
+
+ Clean up after you are done and explicitly close the router connection.
"""
if self.lock_on_connect or self.locked:
self.unlock()
- self.device.close()
+ self._xml_agent_acquired = False
+ self.device.remote_conn.close()
def lock(self):
"""
- Locks the IOS-XR device config.
+ Lock the config database.
+
+ Use if Locking/Unlocking is not performaed automatically by lock=False
"""
if not self.locked:
rpc_command = '<Lock/>'
- response = __execute_rpc__(self.device, rpc_command, self.timeout)
+ try:
+ self._execute_rpc(rpc_command)
+ except XMLCLIError:
+ raise LockError('Unable to enter in configure exclusive mode!', self)
self.locked = True
def unlock(self):
"""
- Unlocks the IOS-XR device config.
+ Unlock the IOS-XR device config.
+
+ Use if Locking/Unlocking is not performaed automatically by lock=False
"""
if self.locked:
rpc_command = '<Unlock/>'
- response = __execute_rpc__(self.device, rpc_command, self.timeout)
+ try:
+ self._execute_rpc(rpc_command)
+ except XMLCLIError:
+ raise UnlockError('Unable to unlock the config!', self)
self.locked = False
def load_candidate_config(self, filename=None, config=None):
"""
- Populates the attribute candidate_config with the desired
+ Load candidate confguration.
+
+ Populate the attribute candidate_config with the desired
configuration and loads it into the router. You can populate it from
a file or from a string. If you send both a filename and a string
containing the configuration, the file takes precedence.
@@ -227,27 +411,27 @@ class IOSXR:
rpc_command = '<CLI><Configuration>'+configuration+'</Configuration></CLI>'
try:
- __execute_rpc__(self.device, rpc_command, self.timeout)
+ self._execute_rpc(rpc_command)
except InvalidInputError as e:
self.discard_config()
- raise InvalidInputError(e.message)
+ raise InvalidInputError(e.message, self)
def get_candidate_config(self, merge=False, formal=False):
"""
- Retrieve the configuration loaded as candidate config in your configuration session
+ Retrieve the configuration loaded as candidate config in your configuration session.
:param merge: Merge candidate config with running config to return
the complete configuration including all changed
:param formal: Return configuration in IOS-XR formal config format
"""
- command="show configuration"
+ command = "show configuration"
if merge:
- command+=" merge"
+ command += " merge"
if formal:
- command+=" formal"
- response = __execute_config_show__(self.device, command, self.timeout)
+ command += " formal"
+ response = self._execute_config_show(command)
- match = re.search(".*(!! IOS XR Configuration.*)$",response,re.DOTALL)
+ match = re.search(".*(!! IOS XR Configuration.*)$", response, re.DOTALL)
if match is not None:
response = match.group(1)
@@ -255,34 +439,37 @@ class IOSXR:
def compare_config(self):
"""
- Compares executed candidate config with the running config and
- returns a diff, assuming the loaded config will be merged with the
+ Compare configuration to be merged with the one on the device.
+
+ Compare executed candidate config with the running config and
+ return a diff, assuming the loaded config will be merged with the
existing one.
:return: Config diff.
"""
- show_merge = __execute_config_show__(self.device, 'show configuration merge', self.timeout)
- show_run = __execute_config_show__(self.device, 'show running-config', self.timeout)
+ _show_merge = self._execute_config_show('show configuration merge')
+ _show_run = self._execute_config_show('show running-config')
- diff = difflib.unified_diff(show_run.splitlines(1)[2:-2],show_merge.splitlines(1)[2:-2],n=0)
- diff = ''.join([x.replace('\r', '') for x in diff])
- return diff
+ diff = difflib.unified_diff(_show_run.splitlines(1)[2:-2], _show_merge.splitlines(1)[2:-2])
+ return ''.join([x.replace('\r', '') for x in diff])
def compare_replace_config(self):
"""
- Compares executed candidate config with the running config and
- returns a diff, assuming the entire config will be replaced.
+ Compare configuration to be replaced with the one on the device.
+
+ Compare executed candidate config with the running config and
+ return a diff, assuming the entire config will be replaced.
:return: Config diff.
"""
- diff = __execute_config_show__(self.device, 'show configuration changes diff', self.timeout)
+
+ diff = self._execute_config_show('show configuration changes diff')
return ''.join(diff.splitlines(1)[2:-2])
def commit_config(self, label=None, comment=None, confirmed=None):
"""
- Commits the candidate config to the device, by merging it with the
- existing one.
+ Commit the candidate config.
:param label: Commit comment, displayed in the commit entry on the device.
:param comment: Commit label, displayed instead of the commit ID on the device.
@@ -296,15 +483,15 @@ class IOSXR:
if confirmed:
if 30 <= int(confirmed) <= 300:
rpc_command += ' Confirmed="%d"' % int(confirmed)
- else: raise InvalidInputError('confirmed needs to be between 30 and 300')
+ else:
+ raise InvalidInputError('confirmed needs to be between 30 and 300 seconds', self)
rpc_command += '/>'
- response = __execute_rpc__(self.device, rpc_command, self.timeout)
+ self._execute_rpc(rpc_command)
def commit_replace_config(self, label=None, comment=None, confirmed=None):
"""
- Commits the candidate config to the device, by replacing the existing
- one.
+ Commit the candidate config to the device, by replacing the existing one.
:param comment: User comment saved on this commit on the device
:param label: User label saved on this commit on the device
@@ -318,21 +505,25 @@ class IOSXR:
if confirmed:
if 30 <= int(confirmed) <= 300:
rpc_command += ' Confirmed="%d"' % int(confirmed)
- else: raise InvalidInputError('confirmed needs to be between 30 and 300')
+ else:
+ raise InvalidInputError('confirmed needs to be between 30 and 300 seconds', self)
rpc_command += '/>'
- response = __execute_rpc__(self.device, rpc_command, self.timeout)
+ self._execute_rpc(rpc_command)
def discard_config(self):
"""
- Clears uncommited changes in the current session.
+ Clear uncommited changes in the current session.
+
+ Clear previously loaded configuration on the device without committing it.
"""
rpc_command = '<Clear/>'
- response = __execute_rpc__(self.device, rpc_command, self.timeout)
+ self._execute_rpc(rpc_command)
- def rollback(self):
+ def rollback(self, rb_id=1):
"""
- Used after a commit, the configuration will be reverted to the
- previous committed state.
+ Rollback the last committed configuration.
+
+ :param rb_id: Rollback a specific number of steps. Default: 1
"""
- rpc_command = '<Unlock/><Rollback><Previous>1</Previous></Rollback><Lock/>'
- response = __execute_rpc__(self.device, rpc_command, self.timeout)
+ rpc_command = '<Unlock/><Rollback><Previous>{rb_id}</Previous></Rollback><Lock/>'.format(rb_id=rb_id)
+ self._execute_rpc(rpc_command)
diff --git a/requirements.txt b/requirements.txt
index 808fb07..ff86190 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1 +1,2 @@
-pexpect
+netmiko >= 0.5.2
+lxml>=3.2.4
diff --git a/setup.py b/setup.py
index e674e3a..6ad489d 100644
--- a/setup.py
+++ b/setup.py
@@ -1,4 +1,9 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""A module to interact with Cisco devices running IOS-XR."""
+
# Copyright 2015 Netflix. All rights reserved.
+# Copyright 2016 BigWaveIT. All rights reserved.
#
# The contents of this file are licensed under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with the
@@ -23,7 +28,7 @@ install_reqs = parse_requirements('requirements.txt', session=uuid.uuid1())
# e.g. ['django==1.5.1', 'mezzanine==1.4.6']
reqs = [str(ir.req) for ir in install_reqs]
-version = '0.14'
+version = '0.21'
setup(
name='pyIOSXR',
@@ -32,11 +37,11 @@ setup(
packages=find_packages(),
install_requires=reqs,
include_package_data=True,
- description = 'Python API to interact with network devices running IOS-XR',
- author = 'Elisa Jasinska',
- author_email = 'elisa at bigwaveit.org',
- url = 'https://github.com/fooelisa/pyiosxr/', # use the URL to the github repo
- download_url = 'https://github.com/fooelisa/pyiosxr/tarball/%s' % version,
- keywords = ['IOS-XR', 'IOSXR', 'Cisco', 'networking'],
- classifiers = [],
+ description='Python API to interact with network devices running IOS-XR',
+ author='Elisa Jasinska',
+ author_email='elisa at bigwaveit.org',
+ url='https://github.com/fooelisa/pyiosxr/',
+ download_url='https://github.com/fooelisa/pyiosxr/tarball/%s' % version,
+ keywords=['IOS-XR', 'IOSXR', 'Cisco', 'networking'],
+ classifiers=[],
)
diff --git a/test/test.py b/test/test.py
index bfbff9f..7ef6b99 100755
--- a/test/test.py
+++ b/test/test.py
@@ -1,467 +1,750 @@
#!/usr/bin/env python
+# coding=utf-8
+"""Unit tests for pyiosxr, a module to interact with Cisco devices running IOS-XR."""
+import os
import sys
-import mock
import unittest
-import xml.etree.ElementTree as ET
+from lxml import etree as ET
-import pexpect
+# ~~~ import pyIOSXR modules ~~~
from pyIOSXR import IOSXR
-from pyIOSXR.exceptions import XMLCLIError, InvalidInputError, TimeoutError, EOFError, IteratorIDError
+# private functions
+from pyIOSXR.iosxr import __execute_rpc__
+from pyIOSXR.iosxr import __execute_show__
+from pyIOSXR.iosxr import __execute_config_show__
+# exceptions
+from pyIOSXR.exceptions import LockError
+from pyIOSXR.exceptions import UnlockError
+from pyIOSXR.exceptions import XMLCLIError
+from pyIOSXR.exceptions import CommitError
+from pyIOSXR.exceptions import ConnectError
+from pyIOSXR.exceptions import TimeoutError
+from pyIOSXR.exceptions import IteratorIDError
+from pyIOSXR.exceptions import InvalidInputError
+from pyIOSXR.exceptions import CompareConfigError
+from pyIOSXR.exceptions import InvalidXMLResponse
+
+
+class _MockedNetMikoDevice(object):
+
+ """
+ Defines the minimum attributes necessary to mock a SSH connection using netmiko.
+ """
+
+ def __init__(self):
+
+ class _MockedParamikoTransport(object):
+ def close(self):
+ pass
+ self.remote_conn = _MockedParamikoTransport()
+
+ @staticmethod
+ def get_mock_file(command, format='xml'):
+ filename = \
+ command.replace('<?xml version="1.0" encoding="UTF-8"?><Request MajorVersion="1" MinorVersion="0">', '')\
+ .replace('</Request>', '')\
+ .replace('<', '')\
+ .replace('>', '_')\
+ .replace('/', '')\
+ .replace('\n', '')\
+ .replace('.', '_')\
+ .replace(' ', '_')\
+ .replace('"', '_')\
+ .replace('=', '_')\
+ .replace('$', '')\
+ .replace('!', '')[:150]
+ curr_dir = os.path.dirname(os.path.abspath(__file__))
+ filename = '{filename}.{fmt}'.format(
+ filename=filename,
+ fmt=format
+ )
+ fullpath = os.path.join(curr_dir, 'mock', filename)
+ return open(fullpath).read()
+
+ def find_prompt(self):
+ return self.get_mock_file('\n', format='txt')
+
+ def send_command(self,
+ command_string,
+ delay_factor=.1,
+ max_loops=150,
+ strip_prompt=True,
+ strip_command=True):
+ return self.get_mock_file(command_string)
+
+ def receive_data_generator(self):
+ return ['', ''] # to have an iteration inside private method _netmiko_recv
+
+ def send_command_expect(self,
+ command_string,
+ expect_string=None,
+ delay_factor=.2,
+ max_loops=500,
+ auto_find_prompt=True,
+ strip_prompt=True,
+ strip_command=True):
+ # for the moment returns the output from send_command only
+ # this may change in time
+ return self.send_command(command_string)
+
+
+class _MockedIOSXRDevice(IOSXR):
+
+ """
+ Overrides only the very basic methods from the main device driver, that cannot be mocked.
+ """
+
+ def open(self):
+ self.device = _MockedNetMikoDevice()
+ self._cli_prompt = self.device.find_prompt()
+ self._enter_xml_mode()
+
+
+class TestIOSXRDevice(unittest.TestCase):
+
+ """
+ Tests IOS-XR basic functions.
+ """
+
+ HOSTNAME = 'localhost'
+ USERNAME = 'vagrant'
+ PASSWORD = 'vagrant'
+ PORT = 12205
... 1023 lines suppressed ...
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/python-modules/packages/pyiosxr.git
More information about the Python-modules-commits
mailing list