API Reference¶

yang.connector module¶

yang.connector module defines a set of classes that connect to Data Model Interfaces (DMI)

Netconf is a wrapper around the ncclient package. Gnmi is a wrapper around the grpcio v1.12.1+ package with protobuf v3.6.0. Restconf implementation is coming next.

class yang.connector.Gnmi(*args, **kwargs)[source]¶

Bases: pyats.connections.bases.BaseConnection

Session handling for gNMI connections.

Can be used with pyATS same as yang.connector.Netconf is used or can be used as a standalone module.

Methods:

capabilities(): gNMI Capabilities.

set(dict): gNMI Set. Input is namespace, xpath/value pairs.

get(dict): gNMI Get mode=’STATE’. Input xpath/value pairs (value optional).

subscribe(dict): gNMI Subscribe. Input xpath/value pairs and format.

pyATS Examples:

>>> from pyats.topology import loader
>>> from yang.connector.gnmi import Gnmi
>>> testbed=loader.load('testbed.static.yaml')
>>> device=testbed.devices['uut']
>>> device.connect(alias='gnmi', via='yang2')
>>> #####################
>>> # Capabilities      #
>>> #####################
>>> resp=device.capabilities()
>>> resp.gNMI_version
'0.7.0'
>>> #####################
>>> # Get example       #
>>> #####################
>>> from yang.connector import proto
>>> request = proto.gnmi_pb2.GetRequest()
>>> request.type = proto.gnmi_pb2.GetRequest.DataType.Value('ALL')
>>> request.encoding = proto.gnmi_pb2.Encoding.Value('JSON_IETF')
>>> path = proto.gnmi_pb2.Path()
>>> path1, path2, path3, path4 = (
        proto.gnmi_pb2.PathElem(),
        proto.gnmi_pb2.PathElem(),
        proto.gnmi_pb2.PathElem(),
        proto.gnmi_pb2.PathElem()
    )
>>> path1.name, path2.name, path3.name, path4.name = (
        'syslog',
        'messages',
        'message',
        'node-name'
    )
>>> path.elem.extend([path1, path2, path3, path4])
>>> request.path.append(path)
>>> resp = device.gnmi.get(request)
>>> print(resp)

instantiates this pipeline instance and store arguments internally.

Parameters

(obj) (device) –
(str) (via) –
(str) –
(dict) (kwargs) –

Note

this api only supports keyword-only arguments for clarity’s sake.

capabilities()[source]¶

Gnmi Capabilities method.

Returns: gNMI Capabilities object
Return type: proto.gnmi_pb2.CapabilityResponse

configure(cmd)[source]¶

Helper method for backwards compatibility.

Parameters: cmd (proto.gnmi_pb2.SetRequest) – gNMI SetRequest object
Returns: gNMI SetResponse object
Return type: proto.gnmi_pb2.SetResponse

connect()[source]¶

Connect to device using gNMI and get capabilities.

Raises: gNMIException – No gNMI capabilities returned by device.

property connected¶: Return True if session is connected.

disconnect()[source]¶: Disconnect from SSH device.

execute(cmd)[source]¶

Helper method for backwards compatibility.

Parameters: cmd (proto.gnmi_pb2.GetRequest) – gNMI GetResponse object
Returns: gNMI GetResponse object
Return type: proto.gnmi_pb2.GetResponse

get(request)[source]¶

Gnmi GET method.

Parameters: request (proto.gnmi_pb2.GetRequest) – gNMI GetResponse object
Returns: gNMI GetResponse object
Return type: proto.gnmi_pb2.GetResponse

property gnmi¶

Helper method to keep backwards compatibility.

Returns: self
Return type: Gnmi

set(request)[source]¶

Gnmi SET method.

Parameters: request (proto.gnmi_pb2.SetRequest) – gNMI SetRequest object
Returns: gNMI SetResponse object
Return type: proto.gnmi_pb2.SetResponse

subscribe(request_iter)[source]¶

Gnmi Subscribe method.

Parameters: request_iter (proto.gnmi_pb2.SubscribeRequest) – gNMI SubscribeRequest object
Returns: gNMI SubscribeResponse object
Return type: proto.gnmi_pb2.SubscribeResponse

class yang.connector.GnmiNotification(device, response, **request)[source]¶

Bases: threading.Thread

Thread listening for event notifications from the device.

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

process_opfields(response)[source]¶

Decode response and verify result.

Decoder callback returns desired format of response. Verify callback returns verification of expected results.

Parameters: response (proto.gnmi_pb2.Notification) – Contains updates that have changes since last timestamp.

property request¶

run()[source]¶: Check for inbound notifications.

stop()[source]¶

stopped()[source]¶

class yang.connector.Grpc(*args, **kwargs)[source]¶

Bases: pyats.connections.bases.BaseConnection

Session handling for Grpc outbound connections.

Can be used with pyATS same as yang.connector

EXAMPLE TESTBED

devices:

router-1:

connections:

a:: ip: 10.10.0.1 port: 23 protocol: telnet
grpc:: protocol: grpc class: yang.connector.Grpc overwrite_config_file: True (Optional, default: False) config_file: /Users/user/telemetry/router_1/config.conf (Optional, default: ./transporter.conf) output_file: /Users/user/telemetry/router_1/output.txt (Optional, default: ./mdt.json) telemetry_subscription_id: 501 (Optional, default: 11172017) transporter: telegraf (Optional, default: telegraf) transporter_ip: 192.168.0.253 (Optional, default will fetch local IP) transporter_port: 56789 (Optional, default is a dynamic port) autoconfigure: True (Optional, default: True)

credentials:

default:: username: user password: cisco123

os: iosxe

EXAMPLE USAGE

Python 3.11.5 (main, Sep 25 2023, 16:57:00) [Clang 14.0.0 (clang-1400.0.29.202)]

>>> from pyats.topology.loader import load
>>> testbed = load('/Users/user/testbed.yaml')
-------------------------------------------------------------------------------
>>> dev = testbed.devices['router-1.yaml']
>>> dev.connect(via='grpc', alias='grpc')

instantiates this pipeline instance and store arguments internally.

Parameters

(obj) (device) –
(str) (via) –
(str) –
(dict) (kwargs) –

Note

this api only supports keyword-only arguments for clarity’s sake.

class yang.connector.Netconf(*args, **kwargs)[source]¶

Bases: ncclient.manager.Manager, pyats.connections.bases.BaseConnection

Implementation of NetConf connection to devices (NX-OS, IOS-XR or IOS-XE), based on pyATS BaseConnection and ncclient.

YAML Example:

devices:
    asr22:
        type: 'ASR'
        tacacs:
            login_prompt: "login:"
            password_prompt: "Password:"
            username: "admin"
        passwords:
            tacacs: admin
            enable: admin
            line: admin
        connections:
            a:
                protocol: telnet
                ip: "1.2.3.4"
                port: 2004
            vty:
                protocol : telnet
                ip : "2.3.4.5"
            netconf:
                class: yang.connector.Netconf
                ip : "2.3.4.5"
                port: 830
                username: admin
                password: admin

Code Example:

>>> from pyats.topology import loader
>>> testbed = loader.load('/users/xxx/xxx/asr22.yaml')
>>> device = testbed.devices['asr22']
>>> device.connect(alias='nc', via='netconf')
>>> device.nc.connected
True
>>> netconf_request = """
...     <rpc message-id="101"
...      xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
...     <get>
...     <filter>
...     <native xmlns="http://cisco.com/ns/yang/ned/ios">
...     <version>
...     </version>
...     </native>
...     </filter>
...     </get>
...     </rpc>
...     """
>>> reply = device.nc.request(netconf_request)
>>> print(reply)
<?xml version="1.0" encoding="UTF-8"?>
<rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"
message-id="101"><data>
<native xmlns="http://cisco.com/ns/yang/ned/ios">
<version>16.3</version></native></data></rpc-reply>
>>> device.nc.disconnect()
>>> device.nc.connected
False
>>>

timeout¶

Timeout value in seconds which is used by paramiko channel. By default this value is 30 seconds.

Type: int

client_capabilities¶

Object ncclient.capabilities.Capabilities representing the client’s capabilities.

Type: object

server_capabilities¶

Object ncclient.capabilities.Capabilities representing the server’s capabilities, and it has a list of data models the server supports.

Type: object

async_mode¶

Specify whether operations are executed asynchronously (True) or synchronously (False). The default value is False.

Type: boolean

__init__ instantiates a single connection instance.

configure(msg)[source]¶

High-level api: configure is a common method of console, vty and ssh sessions, however it is not supported by this Netconf class. This is just a placeholder in case someone mistakenly calls config method in a netconf session. An Exception is thrown out with explanation.

Parameters: msg (str) – Any config CLI need to be sent out.
Raises: Exception – configure is not a supported method of this Netconf class.

configure_logging()[source]¶

connect()[source]¶

High-level api: opens the NetConf connection and exchanges capabilities. Since topology YAML file is parsed by BaseConnection, the following parameters can be specified in your YAML file.

Parameters

host (string) – Hostname or IP address to connect to.
port (int, optional) – By default port is 830, but some devices use the default SSH port of 22 so this may need to be specified.
timeout (int, optional) – An optional keyed argument to set timeout value in seconds. By default this value is 30 seconds.
username (string) – The username to use for SSH authentication.
password (string) – The password used if using password authentication, or the passphrase to use for unlocking keys that require it.
key_filename (string) – a filename where a the private key to be used can be found.
allow_agent (boolean) – Enables querying SSH agent (if found) for keys. The default value is True.
hostkey_verify (boolean) – Enables hostkey verification from ~/.ssh/known_hosts. The default value is False.
look_for_keys (boolean) – Enables looking in the usual locations for ssh keys (e.g. ~/.ssh/id_*). The default value is True.
ssh_config (string) – Enables parsing of an OpenSSH configuration file, if set to its path, e.g. ~/.ssh/config or to True. If the value is True, ncclient uses ~/.ssh/config. The default value is None.

Raises

Exception – If the YAML file does not have correct connections section, or establishing transport to ip:port is failed, ssh authentication is failed, or other transport failures.

Note

There is no return from this method. If something goes wrong, an exception will be raised.

YAML Example:

devices:
    asr22:
        type: 'ASR'
        tacacs:
            login_prompt: "login:"
            password_prompt: "Password:"
            username: "admin"
        passwords:
            tacacs: admin
            enable: admin
            line: admin
        connections:
            a:
                protocol: telnet
                ip: "1.2.3.4"
                port: 2004
            vty:
                protocol : telnet
                ip : "2.3.4.5"
            netconf:
                class: yang.connector.Netconf
                ip : "2.3.4.5"
                port: 830
                username: admin
                password: admin

Code Example:

>>> from pyats.topology import loader
>>> testbed = loader.load('/users/xxx/xxx/asr22.yaml')
>>> device = testbed.devices['asr22']
>>> device.connect(alias='nc', via='netconf')
>>>

Expected Results:

>>> device.nc.connected
True
>>> for iter in device.nc.server_capabilities:
...     print(iter)
...
urn:ietf:params:xml:ns:yang:smiv2:RFC-1215?module=RFC-1215
urn:ietf:params:xml:ns:yang:smiv2:SNMPv2-TC?module=SNMPv2-TC
...
>>>

disconnect()[source]¶: High-level api: closes the NetConf connection.

execute(operation, *args, **kwargs)[source]¶: High-level api: The fact that most connection classes implement execute method lead us to add this method here as well. Supported operations are get, get_config, get_schema, dispatch, edit_config, copy_config, validate, commit, discard_changes, delete_config, lock, unlock, close_session, kill_session, poweroff_machine and reboot_machine. Refer to ncclient document for more details.

notify_wait(steps)[source]¶: Activate notification listener and check results

request(msg, timeout=30, return_obj=False)[source]¶

High-level api: sends message through NetConf session and returns with a reply. Exception is thrown out either the reply is in wrong format or timout. Users can modify timeout value (in seconds) by passing parameter timeout. Users may want to set a larger timeout when making a large query.

Parameters

msg (str) – Any message need to be sent out in XML format. The message can be in wrong format if it is a negative test case. Because ncclient tracks same message-id in both rpc and rpc-reply, missing message-id in your rpc may cause exception when receiving rpc-reply. Most other wrong format rpc’s can be sent without exception.
timeout (int, optional) – An optional keyed argument to set timeout value in seconds. Its default value is 30 seconds. If timeout is less than 30, timeout will use the default of 30 seconds.
return_obj (boolean, optional) – Normally a string is returned as a reply. In other cases, we may want to return a RPCReply object, so we can access some attributes, e.g., reply.ok or reply.elapsed.

Returns

The reply from the device in string. If something goes wrong, an exception will be raised. If return_obj=True, the reply is a RPCReply object.

Return type

str or RPCReply

Raises

Exception – If NetConf is not connected, or there is a timeout when receiving reply.

Code Example:

>>> from pyats.topology import loader
>>> testbed = loader.load('/users/xxx/xxx/asr_20_22.yaml')
>>> device = testbed.devices['asr22']
>>> device.connect(alias='nc', via='netconf')
>>> netconf_request = """
...     <rpc message-id="101"
...      xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
...     <get>
...     <filter>
...     <native xmlns="http://cisco.com/ns/yang/Cisco-IOS-XE-native">
...     <version>
...     </version>
...     </native>
...     </filter>
...     </get>
...     </rpc>
...     """
>>> reply = device.nc.request(netconf_request)
>>>

Expected Results:

>>> print(reply)
<?xml version="1.0" encoding="UTF-8"?>
<rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"
message-id="101"><data>
<native xmlns="http://cisco.com/ns/yang/ned/ios">
<version>16.3</version></native></data></rpc-reply>
>>>

property session¶

return the SSH session object.

Returns: The SSH session that was created by ncclient.transport.SSHSession.
Return type: object
Type: High-level api

subscribe(request)[source]¶: Creates a notification listener and mark it as active

class yang.connector.NetconfEnxr(*args, **kwargs)[source]¶

Bases: object

Subclass using POSIX pipes to Communicate NETCONF messaging.

class LockContext(target, cls)[source]¶: Bases: object

chunk = re.compile('(\n#+\\d+\n)')¶

commit(**kwargs)[source]¶

configure(msg)[source]¶

High-level api: configure is a common method of console, vty and ssh sessions, however it is not supported by this NetconfEnxr class. This is just a placeholder in case someone mistakenly calls config method in a netconf session. An Exception is thrown out with explanation.

Parameters: msg (str) – Any config CLI need to be sent out.
Raises: Exception – configure is not a supported method of this Netconf class.

connect(timeout=None)[source]¶: Connect to ENXR pipe.

property connected¶: Check for active connection.

discard_changes(**kwargs)[source]¶

disconnect()[source]¶: Disconnect from ENXR pipe.

dispatch(rpc_command=None, **kwargs)[source]¶

edit_config(target=None, config=None, **kwargs)[source]¶: Send edit-config.

get(filter=None, **kwargs)[source]¶

get_config(source=None, filter=None, **kwargs)[source]¶: Send get-config.

get_rpc(elements)[source]¶: Return string representation of lxml element with rpc.

lock(target=None, **kwargs)[source]¶

locked(target)[source]¶: Return a lock context manager for EnXR

recv_data()[source]¶: Retrieve data from process pipe.

request(rpc)[source]¶

rpc_pipe_err = '\n <rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">\n <rpc-error>\n <error-type>transport</error-type>\n <error-tag>resource-denied</error-tag>\n <error-severity>error</error-severity>\n <error-message>No pipe data returned</error-message>\n </rpc-error>\n </rpc-reply>'¶

send_cmd(rpc)[source]¶: Send a message to process pipe.

unlock(target=None, **kwargs)[source]¶