ncclient:NETCONF Python client

# See the License for the specific language governing permissions and

# limitations under the License.

def _abbreviate(uri):

    if uri.startswith('urn:ietf:params:netconf:'):

        splitted = uri.split(':')

            return [ ':' + splitted[5], ':' + splitted[5] + ':' + splitted[6] ]

        elif ':base:' in uri:

            return [ ':base', ':base' + ':'+ splitted[5] ]

def schemes(url_uri):

    """Given a URI that has a *scheme* query string (i.e. *:url* capability

        """

        return key in self

#: :class:`Capabilities` object representing the capabilities currently supported by NCClient

CAPABILITIES = Capabilities([

    'urn:ietf:params:netconf:capability:url:1.0?scheme=http,ftp,file,https,sftp',

    'urn:ietf:params:netconf:capability:validate:1.0',

    'urn:ietf:params:netconf:capability:xpath:1.0',

    #'urn:ietf:params:netconf:capability:notification:1.0', # TODO

    #'urn:ietf:params:netconf:capability:interleave:1.0' # theoretically already supported

])

def connect_ssh(*args, **kwds):

    """Connect to NETCONF server over SSH. See :meth:`SSHSession.connect()

    session = transport.SSHSession(capabilities.CAPABILITIES)

    session.load_known_hosts()

    session.connect(*args, **kwds)

#: Same as :meth:`connect_ssh`

connect = connect_ssh

    It is also a context manager, so a :class:`Manager` instance can be used

    with the *with* statement. The session is closed when the context ends. """

    def __init__(self, session):

        self._session = session

    def __enter__(self):

        return self

        self.close()

        return False

    def locked(self, target):

        """Returns a context manager for the *with* statement.

from errors import OperationError, TimeoutExpiredError, MissingCapabilityError

from rpc import RPC, RPCReply, RPCError

from retrieve import Get, GetConfig, GetReply

from edit import EditConfig, CopyConfig, DeleteConfig, Validate, Commit, DiscardChanges

from session import CloseSession, KillSession

from lock import Lock, Unlock, LockContext

    'get': Get,

    'validate': Validate,

    'commit': Commit,

    'lock': Lock,

    'unlock': Unlock,

}

__all__ = [

    'DeleteConfig',

    'Lock',

    'Unlock',

    'LockContext',

    'CloseSession',

    'KillSession',

class EditConfig(RPC):

    "*<edit-config>* RPC"

    SPEC = {'tag': 'edit-config', 'subtree': []}

class DeleteConfig(RPC):

    "*<delete-config>* RPC"

    SPEC = {'tag': 'delete-config', 'subtree': []}

class CopyConfig(RPC):

    "*<copy-config>* RPC"

    SPEC = {'tag': 'copy-config', 'subtree': []}

class Validate(RPC):

    "*<validate>* RPC. Depends on the *:validate* capability."

    DEPENDS = [':validate']

class Commit(RPC):

    "*<commit>* RPC. Depends on the *:candidate* capability."

    DEPENDS = [':candidate']

from copy import deepcopy

class Lock(RPC):

    "*<lock>* RPC"

    SPEC = {

class Unlock(RPC):

    "*<unlock>* RPC"

    SPEC = {

class LockContext:

    """

    A context manager for the :class:`Lock` / :class:`Unlock` pair of RPC's.

class GetReply(RPCReply):

    """Adds attributes for the *<data>* element to :class:`RPCReply`, which

    pertains to the :class:`Get` and :class:`GetConfig` operations."""

    def _parsing_hook(self, root):

        self._data = None

        if not self._errors:

    @property

    def data_ele(self):

class Get(RPC):

    "The *<get>* RPC"

    SPEC = {'tag': 'get', 'subtree': []}

class GetConfig(RPC):

    "The *<get-config>* RPC"

    SPEC = {'tag': 'get-config', 'subtree': []}

            return

        root = self._root = xml_.xml2ele(self._raw) # <rpc-reply> element

        # per rfc 4741 an <ok/> tag is sent when there are no errors or warnings

        if ok is not None:

            logger.debug('parsed [%s]' % ok.tag)

        else: # create RPCError objects from <rpc-error> elements

            if error is not None:

                logger.debug('parsed [%s]' % error.tag)

                for err in root.getiterator(error.tag):

        spec = {

            'tag': 'rpc',

            'attrib': {

                'message-id': self._id

                },

            'subtree': [ opspec ]

    def request(self, *args, **kwds):

        """Subclasses implement this method. Here, the operation is constructed

        in :ref:`dtree`, and the result of :meth:`_request` returned."""

    def _delivery_hook(self):

        """Subclasses can implement this method. Will be called after

class CloseSession(RPC):

    "*<close-session>* RPC. The connection to NETCONF server is also closed."

    SPEC = { 'tag': 'close-session' }

    def _delivery_hook(self):

        self.session.close()

class KillSession(RPC):

# See the License for the specific language governing permissions and

# limitations under the License.

        "Given a list of capability URI's returns <hello> message XML string"

        spec = {

            'tag': 'hello',

            'subtree': [{

                'tag': 'capabilities',

                'subtree': # this is fun :-)

### Namespace-related

#: Base NETCONF namespace

#: ... and this is BASE_NS according to Cisco devices tested

#: namespace for Tail-f data model

TAILF_AAA_1_1 = 'http://tail-f.com/ns/aaa/1.1'

#: namespace for Tail-f data model

TAILF_EXECD_1_1 = 'http://tail-f.com/ns/execd/1.1'

#: namespace for Cisco data model

try:

    register_namespace = ET.register_namespace

        # cElementTree uses ElementTree's _namespace_map, so that's ok

        ElementTree._namespace_map[uri] = prefix

unqualify = lambda tag: tag[tag.rfind('}')+1:]

class DictTree:

iselement = ET.iselement

def find(ele, tag, nslist=[]):

    """If *nslist* is empty, same as :meth:`xml.etree.ElementTree.Element.find`.

    If it is not, *tag* is interpreted as an unqualified name and qualified