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(":")

        if ":capability:" in uri:

        elif ":base:" in uri:

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

    return []

def schemes(url_uri):

        return len(self._dict)

    def __iter__(self):

    def __repr__(self):

        return repr(self._dict.keys())

    def add(self, uri):

        "Add a capability."

        self._dict[uri] = _abbreviate(uri)

    def remove(self, uri):

        "Remove a capability."

        if key in self._dict:

import transport

import logging

logger = logging.getLogger('ncclient.manager')

CAPABILITIES = [

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

    "urn:liberouter:params:netconf:capability:power-control:1.0"

    "urn:ietf:params:netconf:capability:interleave:1.0"

]

OPERATIONS = {

    "get": operations.Get,

    "poweroff_machine": operations.PoweroffMachine,

    "reboot_machine": operations.RebootMachine

}

def connect_ssh(*args, **kwds):

    First, ``publickey`` authentication is attempted. If a specific *key_filename* is specified, it

    will be loaded and authentication attempted using it. If *allow_agent* is :const:`True` and an

    an encrypted key file is encountered, the *password* argument will be used as a decryption

    passphrase.

    :param host: hostname or address on which to connect

    :type host: `string`

    :raises: :exc:`~ncclient.transport.AuthenticationError`

    :rtype: `Manager`

    session = transport.SSHSession(capabilities.Capabilities(CAPABILITIES))

    session.load_known_hosts()

    session.connect(*args, **kwds)

connect = connect_ssh

"Same as :func:`connect_ssh`, since SSH is the default (and currently, the only) transport."

class Manager(object):

    def __init__(self, session):

        self._session = session

        self._async_mode = False

        self._timeout = None

    def __enter__(self):

        return self

        self.close_session()

        return False

    def locked(self, target):

        return operations.LockContext(self._session, target)

    @property

    def client_capabilities(self):

        return self._session._client_capabilities

        self._async_mode = mode

    def set_raise_mode(self, mode):

        self._raise_mode = mode

    async_mode = property(fget=lambda self: self._async_mode, fset=set_async_mode)

    "*<edit-config>* RPC"

        node = new_ele("edit-config")

        node.append(util.datastore_or_url("target", target, self._assert))

        if error_option is not None:

            self._assert(":confirmed-commit")

            sub_ele(node, "confirmed")

            if timeout is not None:

                sub_ele(node, "confirm-timeout").text = timeout

        return self._request(node)

    DEPENDS = ["urn:liberouter:params:netconf:capability:power-control:1.0"]

        return self._request(new_ele(qualify("poweroff-machine", PC_URN)))

class RebootMachine(RPC):

    DEPENDS = ["urn:liberouter:params:netconf:capability:power-control:1.0"]

        return self._request(new_ele(qualify("reboot-machine", PC_URN)))

    def request(self, target):

        node = new_ele("lock")

        return self._request(node)

    def request(self, target):

        node = new_ele("unlock")

        return self._request(node)

    def to_dict(self):

        return dict([ (attr[1:], getattr(self, attr)) for attr in RPCError.tag_to_attr.values() ])

    @property

    def xml(self):

        return self._raw

    @property

            if error is not None:

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

                    # Process a particular <rpc-error>

        self._parsed = True

    @property

    def xml(self):

            logger.debug('Async request, returning %r', self)

            return self

        else:

            self._event.wait(self._timeout)

            if self._event.isSet():

                if self._error:

            else:

                raise TimeoutExpiredError

        """Subclasses must implement this method. Typically only the request needs to be built as an

        `~xml.etree.ElementTree.Element` and everything else can be handed off to

        :meth:`_request`."""

        server, before making a request that requires it. A :exc:`MissingCapabilityError` will be

        raised if the capability is not available."""

        if capability not in self._session.server_capabilities:

    def deliver_reply(self, raw):

        # internal use

class SSHUnknownHostError(SSHError):

    def __init__(self, host, fingerprint):

        self.host = host

        self.fingerprint = fingerprint

    return xml if xml.startswith('<?xml') else '<?xml version="1.0" encoding="%s"?>%s' % (encoding, xml)

def to_ele(x):

    :arg x: the XML document or element

    :type x: `string` or `~xml.etree.ElementTree.Element`