pyats.topology package¶

Module:

Topology

Authors:

Siming Yuan (siyuan), CSG - Ottawa

Description:

This module provides two functionalities:

testbed description, conceptually similar to ATS Tcl CONFIG file.
testbed topology information conceptually similar to ATS Tcl’s MAP file.

In Python ATS infrastructure, the above two are combined together and featured in one YAML file. The YAML file is then read to create a Topology object.

The following graph depics what a typical topology looks like:

+--------------------------------------------------------------------------+
| Testbed Object                                                           |
|                                                                          |
| +-----------------------------+          +-----------------------------+ |
| | Device Object - myRouterA   |          | Device Object - myRouterB   | |
| |                             |          |                             | |
| |         device interfaces   |          |          device interfaces  | |
| | +----------+ +----------+   |          |   +----------+ +----------+ | |
| | | intf Obj | | intf Obj |   |          |   |  intf Obj| | intf Obj | | |
| | | Eth1/1   | | Eth1/2 *-----------*----------*  Eth1/1| | Eth1/2   | | |
| | +----------+ + ---------+   |     |    |   +----------+ +----------+ | |
| +-----------------------------+     |    +-----------------------------+ |
|                                     |                                    |
|                               +-----*----+                               |
|                               | Link Obj |                               |
|                               |rtrA-rtrB |                               |
|                               +----------+                               |
+--------------------------------------------------------------------------+

Notes

ipaddress module is called py2-ipaddress in PyPI, a backport of Python3’s standard ipaddress module, documentation at: http://docs.python.org/3.3/library/ipaddress

Subpackages¶

pyats.topology.loader package
- Submodules

Submodules¶

class pyats.topology.bases.TopologyDict(*args, **kwargs)¶

Bases: pyats.datastructures.attrdict.AttrDict

TopologyDict class

Provides a dictionary/AttrDict like datastructure to contain TopologyObjects whilst allowing the user to, in addition to accessing objects by object name, and also access these objects via their object aliases.

Note

the alias-access functionality here is for access-only, and does not operate on set functionalty. eg: when adding key/obj to this container type, the given key is used, regardless of whether it is the obj name or the obj alias.

Example

>>> td = TopologyDict()
>>> td['a'] = TopologyObject('a', alias = 'b')
>>> 'b' in td
True
>>> td['b'] is td['a']
True
>>> td.is_alias('b')
True

AttrDict object init

How this works:

All python objects internally store their attributes in a dictionary that is named __dict__.
There is no requirement that the internal dictionary __dict__ would need to be “just a plain dict”, so we can assign any subclass of dict() to the internal dictionary.
In our case we simply assign the AttrDict() instance we are instantiating (as we are in __init__).
By calling super()’s __init__() method we made sure that it (already) behaves exactly like a dictionary, since that function calls all the dictionary instantiation code.

Special Note:

Known to cause memory leak in Python < 2.7.3 and Python3 < 3.2.3

property aliases¶

Property aliases

Returns the set of all object aliases contained within this dictionary.

Note

if a value obj does not have an alias attribute, then its key is used as default, since [key] access works on both keys and aliases.

Returns
Return type: set() of alises

is_alias(key)¶

Test For alias

Tests wether a given key is an alias to the objects, or is a key.

Returns
Return type: bool()

property names¶

Property names

Returns the set of all names (keys) contained within this dictionary. Same in reality as keys() api, but a property.

Returns
Return type: set() of alises

class pyats.topology.bases.TopologyObject(name, alias=None, **kwargs)¶

Bases: object

TopologyObject Base class

Provides the base topology object class to inherit from. Every topology object should inherit from this class, and thus carry a name and an alias.

This class is to be used in conjunection with TopologyDict, which allows object querying by alias name.

built in class __init__

Instantiates the base topology object.

Parameters

(str) (alias) –
(str) –

Example

>>> obja = TopologyObject(name = 'obja')
>>> objb = TopologyObject(name = 'objb', alias = 'alias_objb')

class pyats.topology.commands.ValidateTestbedFile(*args, **kwargs)¶

Bases: pyats.cli.base.Subcommand

Validates that a testbed file is loadable

description = '\nValidates the provided testbed YAML file for syntax and content completeness.\n '¶

help = 'validate your testbed file yaml syntax'¶

name = 'testbed'¶

run(args)¶: runs this subcommand

usage = '{prog} [file] [options]'¶

class pyats.topology.credentials.Credentials(local_creds={}, parent=None)¶

Bases: collections.abc.MutableMapping

This class defines a chained credential dictionary.

All mutations are defined on local credentials only. Lookups not satisfied against the local credentials are done against the parent’s credentials.

Parameters

local_creds (dict) – The credentials to look up first.
parent (object) – The parent has a credentials member that contains its credentials.

get(key, default=None)¶: Get the key, or default credential, or default value.

local_creds = {}¶

property parent¶

pop(k[, d]) → v, remove specified key and return the corresponding value.¶: If key is not found, d is returned if given, otherwise KeyError is raised.

popitem() → (k, v), remove and return some (key, value) pair¶: as a 2-tuple; but raise KeyError if D is empty.

class pyats.topology.device.Device(*args, **kwargs)¶

Bases: pyats.topology.bases.TopologyObject, pyats.topology.device.DeviceBase

Device class

Defines a device object in a testbed topology. A device is any piece of physical hardware that constitutes an important part of a testbed topology. A device can be either physical or virtual.

routers

switches

traffic generators

servers (that acts as part of the topology)

Example

>>> device = Device('myDevice')
>>> from pyats.topology import Testbed, Interface
>>> testbed = Testbed('myTestbed')
>>> interface = Interface('Ethernet1/1', 'ethernet')
>>> device = Device('myNewDevice',
...                 alias = 'myAlias',
...                 interfaces = [interface,],
...                 testbed = testbed)

Note

Device objects may be duplicated, but their name must be unique within a testbed object. Eg, device hostnames are unique within testbed objects.

Device class init method (built-in)

Parameters

(str) (type) –
(str) –
(str) –
(Testbed) (testbed) –
(dict) (kwargs) – information (how to connect to this device)
(dict) – information specific to this device
(dict) –
(dict) – specific to this device
(dict) – device
(list) (interfaces) –
(dict) –
(dict) –

CFG_TOPOLOGY_CLASS = 'topology.class.device'¶

ENTRY_POINT_NAME = 'device'¶

property default_connection_alias¶

Default alias for device connections

Returns the current default alias for device connections.

property testbed¶

Device testbed property getter

Returns the parent testbed object, or None.

class pyats.topology.device.DeviceBase(testbed=None, type=None, interfaces=None, custom=None, **kwargs)¶

Bases: object

This class defines the basic characteristics and behavior of a device.

Device base class init method (built-in)

Parameters

(str) (type) –
(Testbed) (testbed) –
(list) (interfaces) –
(dict) (kwargs) –
(dict) –

add_interface(interface)¶

Adds an interface to this device

This API adds an interface object to this device object

Parameters: (Interface) (interface) –
Returns
Return type: None

find_links(dest)¶

Find Link

API to find and return the set of matching links that matches the criteria destination, which could be either a device object, or an interface object.

Example

>>> device_a = Device('a')
>>> device_b = Device('b')
>>> link = Link('commonlink')
>>> device_a.add_interface(Interface('Eth1', 'eth', link=link))
>>> device_b.add_interface(Interface('Eth1', 'eth', link=link))
>>> assert link in device_a.find_links(device_b)
>>> assert link in device_a.find_links(device_b.interfaces['Eth1'])

Parameters: (obj) (dest) – interface object
Returns
Return type: set() of links matching the destination

property links¶

links

Returns the set of links currently connected to this device, via its interfaces.

Type: Property

property remote_devices¶

remote_devices

Returns the set of remote devices connected to this device, through its interface links.

Type: Property

property remote_interfaces¶

remote_interfaces

Returns the set of remote interfaces connected to this device, through its interface links.

Type: Property

remove_interface(interface)¶

Removes interface from device

This API removes the given interface object from device

Note that this API does not remove the Interface object, but merely removes it from belonging to this device.

Parameters: (str/Interface) (interface) – removed from device,
Returns
Return type: None

property testbed¶

Device testbed property getter

Returns the parent testbed object, or None.

Errors related to topology module

exception pyats.topology.exceptions.DuplicateDeviceAliasError(alias, testbed)¶

Bases: Exception

Raised when a given device alias is already known or part of a testbed

exception pyats.topology.exceptions.DuplicateDeviceError(device, testbed)¶

Bases: Exception

Raised when a given device is already known or part of a testbed

exception pyats.topology.exceptions.DuplicateInterfaceConnectionError(intf, link)¶

Bases: Exception

Raised when a device interface is non-unique in a link

exception pyats.topology.exceptions.DuplicateInterfaceError(intf, device)¶

Bases: Exception

Raised when a device interface is non-unique in a device

exception pyats.topology.exceptions.DuplicateLinkError(names, testbed)¶

Bases: Exception

Raised when a link name is not unique within a testbed

exception pyats.topology.exceptions.LinkError(intf, link)¶

Bases: Exception

Raised when a topology link error happens

exception pyats.topology.exceptions.MissingDeviceError(device)¶

Bases: Exception

Raised when a device is missing from topology section

exception pyats.topology.exceptions.MissingInterfaceTypeError(interface)¶

Bases: Exception

Raised when an interface is created without a type specified.

exception pyats.topology.exceptions.SubclassFactoryError(basecls, newcls)¶: Bases: Exception

exception pyats.topology.exceptions.UnknownDeviceError(device, testbed)¶

Bases: Exception

Raised when a specified device is unknown

exception pyats.topology.exceptions.UnknownInterfaceError(intf, device)¶

Bases: Exception

Raised when a device interface is unknown

exception pyats.topology.exceptions.YamlConfigError¶: Bases: Exception

class pyats.topology.factory.TopologyMetaClassFactory¶

Bases: pyats.datastructures.factory.MetaClassFactory

TopologyMetaClassFactory Metaclass

A metaclass factory specifically made for topology classes. Inherits from MetaClassFactory as to enable default behaviour of changing to another class if the ‘class’ argument is given. Without this argument, TopologyMetaClassFactory searches configuration and entry points for an available class instead, which is then passes to the parent MetaClassFactory to perform a class change.

class pyats.topology.interface.Interface(*args, **kwargs)¶

Bases: pyats.topology.bases.TopologyObject, pyats.topology.interface.InterfaceBase

Interface class

Defines an interface object. Interface objects ideally belong to a parent device object. Interface names are unique within a device, and must be instantiated with an interface name.

Example

>>> device = Device('myDevice')
>>> interface = Interface('Ethernet1/1',
...                       type = 'ethernet',
...                       device = device)

>>> link = Link('newlink')
>>> interface = Interface('Ethernet2/1',
...                       type = 'ethernet',
...                       alias = 'myinterface',
...                       link = link,
...                       device = device)

Interface class init (built-in)

Parameters

(string) (alias) –
(string) –
(string) –
(device) (device) –
(ipaddress.IPv4Interface) (ipv4) –
(ipaddress.IPv6Interface or list of ipaddress.IPv6Interface) (ipv6) – IPv6 address/mask of interface
(Link or string) (link) – connected to, or the name of the link where this interface is connected to.
(dict) (kwargs) – interface attributes

Returns

Return type

None

Raises

TypeError –

CFG_TOPOLOGY_CLASS = 'topology.class.interface'¶

ENTRY_POINT_NAME = 'interface'¶

class pyats.topology.interface.InterfaceBase(type, device=None, link=None, **kwargs)¶

Bases: object

This class defines the basic characteristics and behavior of an interface.

Interface class init (built-in)

Parameters

(string) (type) –
(device) (device) –
(Link or string) (link) – connected to, or the name of the link where this interface is connected to.
(dict) (kwargs) – interface attributes

Returns

Return type

None

Raises

TypeError –

property device¶

Interface device property getter

Returns the parent device object, or None.

property link¶

Interface link property getter

Returns the link obj this interface is connected to, or None

property remote_devices¶

remote devices

Returns the list of remote devices connected to this interface

Parameters: None –
Returns
Return type: set of Devices connected to this interface
Type: Property

property remote_interfaces¶

remote interfaces

Returns the list of remote interfaces connected to this interface via the interface link.

Parameters: None –
Returns
Return type: set of Interfaces connected to this interface
Type: Property

class pyats.topology.interface.myINTER(*args, **kwargs)¶

Bases: pyats.topology.interface.Interface

Interface class init (built-in)

Parameters

(string) (alias) –
(string) –
(string) –
(device) (device) –
(ipaddress.IPv4Interface) (ipv4) –
(ipaddress.IPv6Interface or list of ipaddress.IPv6Interface) (ipv6) – IPv6 address/mask of interface
(Link or string) (link) – connected to, or the name of the link where this interface is connected to.
(dict) (kwargs) – interface attributes

Returns

Return type

None

Raises

TypeError –

class pyats.topology.link.Link(*args, **kwargs)¶

Bases: pyats.topology.bases.TopologyObject, pyats.topology.link.LinkBase

Link class

Defines a link object in a testbed topology. A Link object is associated with one or more Interface objects, and links should be unique by name within a testbed object.

Note

interfaces contained within a Link object is referenced by weak references.

Example

>>> link = Link('newlink')
>>> interface = Interface('Ethernet2/1',
...                       type = 'ethernet',
...                       alias = 'myinterface',
...                       link = link,
...                       device = device)
>>> link = Link('newLink',
...             alias = 'myLink',
...             interfaces = [interface,])

Link class init method (built-in)

Parameters

(str) (alias) –
(str) –
(list) (interfaces) –
(dict) (kwargs) – link attributes

CFG_TOPOLOGY_CLASS = 'topology.class.link'¶

ENTRY_POINT_NAME = 'link'¶

class pyats.topology.link.LinkBase(interfaces=None, **kwargs)¶

Bases: object

This class defines the basic characteristics and behavior of a link.

Link class init method (built-in)

Parameters

(list) (interfaces) –
(dict) (kwargs) – link attributes

connect_interface(interface)¶

Connects an interface to this Link

API to add(associate) Interface(obj) to this Link

Parameters: (Interface) (interface) – to be connected to this Link

property connected_devices¶

Connected Devices Property

Property that returns the set of unique connected devices to this link.

Returns
Return type: set() of unique devices connected to this link

disconnect_interface(interface)¶

Disconnects an interface from this Link

API to remove(unassociate) Interface(obj) from this Link

Parameters: (Interface/str) (interfaces) – to be removed from this Link

pyats.topology.schema.ipv6_or_list_of_ipv6(value)¶

check_file

translates IPv6 address, or list of IPv6 addresses.

pyats.topology.schema.secret_string(path, value)¶

Parameters

path (list) – Location in loaded file of secret string to ask user to enter.
value (str or int) – Value of secret string specified in content.

class pyats.topology.testbed.Testbed(*args, **kwargs)¶

Bases: pyats.topology.bases.TopologyObject, pyats.topology.testbed.TestbedBase

Testbed class

Defines a testbed object which is the overall container where devices, device interfaces & links are contained.

Example

>>> testbed = Testbed('myTestbed')
>>> from pyats.topology import Device
>>> device = Device('myDevice')
>>> testbed.add_device(device)

Note

device names within each testbed must be unique.

Testbed class init method (built-in)

Parameters

(str) (testbed_file) –
(str) –
(list) (devices) –
(dict) (clean) – testbed (such as tftp/ftp/ntp)
(dict) –
(dict) –
(dict) – common to this testbed
(dict) – information common to this testbed
(dict) – to this testbed
(str) – this testbed object.

CFG_TOPOLOGY_CLASS = 'topology.class.testbed'¶

ENTRY_POINT_NAME = 'testbed'¶

configure(command, *, devices=None, ikwargs=[], **kwargs)¶

Configure commands against devices in parallel.

Parameters

(str or list) (command) –
(list of Device) (devices) – if not provided, defaults to all devices in this testbed
(list of dict) (ikwargs) – configure service for each device
**kwargs (any additional keyword arguments to be passed to the) – configure service.

Returns

Dictionary with output {device.name

Return type

<configure output>}

Example

# configure against all devices testbed.configure(‘no ip domain-lookup’)

# configure for some devices # Note: devices is a list of Device objects devices = [testbed.devices[dev] for dev testbed.devices if testbed.devices[dev].os == ‘iosxe’] testbed.configure(‘ntp server 172.1.2.3’, devices=devices)

connect(*devices, vias={}, timeout=None, **kwargs)¶

Connect to testbed devices in parallel

convenience function to initiate connection to this testbed’s devices in asynchronous fashion.

Parameters

(list) (devices) – if not provided, defaults to all devices in this testbed (this is a *varargs argument)
(dict) (kwargs) – initiate connectivity with.
(int) (timeout) –
(dict) – connection class as initialization argument

Example

# connect to all devices in this testbed testbed.connect()

# connect to some devices in this testbed testbed.connect(testbed.devices[‘uut’],

testbed.devices[‘helper’])

# connect to some devices in this testbed # and provide unique vias testbed.connect(testbed.devices[‘uut’],

testbed.devices[‘helper’], vias = {‘uut’: ‘cli’,

‘helper’: ‘console’},

log_stdout = False)

Returns
Return type: None

destroy(*devices, vias={}, **kwargs)¶

Destroy connections to testbed devices in parallel

This is the API to destroy all connections object to this testbed’s devices in asynchronous fashion. It will first disconnect the connection, the remove the object from device connection manager.

Parameters

(list) (devices) – if not provided, defaults to all devices in this testbed (this is a *varargs argument)
(dict) (kwargs) – initiate connectivity with.
(dict) – connection class as initialization argument

Example

# destroy all connections in this testbed testbed.destroy()

# destroy some devices in this testbed testbed.disconnect(testbed.devices[‘uut’],

testbed.devices[‘helper’])

# destroy some devices in this testbed # and provide unique vias testbed.disconnect(testbed.devices[‘uut’],

testbed.devices[‘helper’], vias = {‘uut’: ‘cli’,

‘helper’: ‘console’})

Returns
Return type: None

disconnect(*devices, vias={}, timeout=None, **kwargs)¶

Disconnect all connections to testbed devices in parallel

This is the API to disconnect all connections object to this testbed’s devices in asynchronous fashion. It will only disconnect the connection, the object will remain in device connection

manager.

Parameters

(list) (devices) – if not provided, defaults to all devices in this testbed (this is a *varargs argument)
(dict) (kwargs) – initiate connectivity with.
(dict) – connection class as initialization argument

Example

# disconnect all connections in this testbed testbed.disconnect()

# disconnect to some devices in this testbed testbed.disconnect(testbed.devices[‘uut’],

testbed.devices[‘helper’])

# disconnect some devices in this testbed # and provide unique vias testbed.disconnect(testbed.devices[‘uut’],

testbed.devices[‘helper’], vias = {‘uut’: ‘cli’,

‘helper’: ‘console’})

Returns
Return type: None

execute(command, *, devices=None, ikwargs=[], **kwargs)¶

Execute commands against devices in parallel.

Parameters

(str or list) (command) –
(list of Device) (devices) – if not provided, defaults to all devices in this testbed
(list of dict) (ikwargs) – execute service for each device
**kwargs (any additional keyword arguments to be passed to the) – execute service.

Returns

Dictionary with output {device.name

Return type

<execute output>}

Example

# execute show version against all devices output = testbed.execute(‘show version’)

# execute show version against some devices # Note: devices is a list of Device objects devices = [testbed.devices[dev] for dev testbed.devices if testbed.devices[dev].os == ‘iosxe’] output = testbed.execute(‘show version’, devices=devices)

# Additional arguments for execution, e.g. timeout output = testbed.execute(‘show arp’, timeout=120)

parse(command, *, devices=None, ikwargs=[], **kwargs)¶

Parse commands from devices in parallel.

Parameters

(str) (command) –
(list of Device) (devices) – if not provided, defaults to all devices in this testbed
(list of dict) (ikwargs) – parse service for each device
**kwargs (any additional keyword arguments to be passed to the) – parse service.

Returns

Dictionary with output {device.name

Return type

<parse output>}

Example

# parse command from all devices in parallel output = testbed.parse(‘show version’)

squeeze(*wanted_obj_names, extend_devices_from_links=True)¶

Destructively reduces the topology to include only named devs/links.

API to force the testbed to destroy unwanted devices, links and the interfaces connected to them.

Parameters

wanted_obj_names (A series of positional str parameters.) – The device and/or link names to keep in the topology (aliases are accepted).
extend_devices_from_links (bool) – If True, then the list of wanted devices is extended with all devices connected to wanted links, whether or not they were in the user’s wanted device list. If False, then the list of wanted devices is not extended and all devices not specified by the user are pruned. Wanted links are also pruned if they are not connected to any of the user’s wanted devices. Interfaces are pruned from wanted links if they are not connected to a wanted device.

Example

This section contains initial configuration used by the following examples:

>>> from pyats.topology import Device,Link,Interface,Testbed
>>> t = Testbed('tb')
>>> t.add_device(Device('d1'))
>>> t.add_device(Device('d2'))
>>> t.add_device(Device('d3'))
>>> d1 = t.devices.d1
>>> d2 = t.devices.d2
>>> d3 = t.devices.d3
>>> l12 = Link('l12')
>>> l13 = Link('l13')
>>> l23 = Link('l23')
>>> d1.add_interface(Interface("d1.i1", link=l12, type="ethernet"))
>>> d1.add_interface(Interface("d1.i2", link=l13, type="ethernet"))
>>> d2.add_interface(Interface("d2.i1", link=l12, type="ethernet"))
>>> d2.add_interface(Interface("d2.i2", link=l23, type="ethernet"))
>>> d3.add_interface(Interface("d3.i1", link=l13, type="ethernet"))
>>> d3.add_interface(Interface("d3.i2", link=l23, type="ethernet"))
>>> [dev for dev in t.devices]
['d3', 'd2', 'd1']
>>> [link.name for link in t.links]
['l23', 'l12', 'l13']

The following example shows a link-extended squeeze.
Notice that other devices are added to the list of wanted devices
if they are connected to the wanted link::

    >>> t.squeeze('d1', 'l23', extend_devices_from_links=True)
    >>> [dev for dev in t.devices]
    ['d3', 'd2', 'd1']
    >>> [link.name for link in t.links]
    ['l23']


The following example shows a non-link-extended squeeze.
Notice that a wanted link is excluded if it is not connected
to wanted devices::

    >>> t.squeeze('d1', 'l23', extend_devices_from_links=False)
    >>> [dev for dev in t.devices]
    ['d1']
    >>> [link.name for link in t.links]
    []

Returns
Return type: None

class pyats.topology.testbed.TestbedBase(devices=None, credentials={}, servers=None, custom=None, **kwargs)¶

Bases: object

This class defines the basic characteristics and behavior of a testbed.

Testbed base class init method (built-in)

Parameters

(list) (devices) –
(dict) (kwargs) –
(dict) – testbed (such as tftp/ftp/ntp)
(dict) –
(dict) –

add_device(device)¶

Adds a device to this testbed

API to add a new device to this testbed

Parameters: (Device) (device) –
Returns
Return type: None

property links¶

links

returns the set of links that currently exists in this testbed

Type: Property

passwords_deprecation_suppress = False¶

remove_device(device)¶

Removes a device from this testbed

API to remove an existing device to this testbed.

Note that this API does not remove the Device object, but merely removes it from belonging to this testbed.

Parameters: (Device/str) (device) –
Returns
Return type: None

server_cred_deprecation_suppress = False¶

tacacs_deprecation_suppress = False¶