Skip to content

Instantly share code, notes, and snippets.

@jathanism

jathanism/01_napalm_compat.py

Last active January 31, 2017 15:39
Show Gist options
  • Save jathanism/a012b7103f63a92a7e3d to your computer and use it in GitHub Desktop.
Save jathanism/a012b7103f63a92a7e3d to your computer and use it in GitHub Desktop.
Download ZIP
A very basic test of integration of Trigger's NetDevices/Tacacsrc w/ Napalm. There are two prototypes: 03_troll_test.py is testing using asyncio/trollius or async I/O. 04_twist_test.py is using Twisted.
Raw
01_napalm_compat.py
"""
Integration of Trigger + NAPALM drivers
"""
from napalm import get_network_driver
def get_driver_for_netdevice(device):
vendor_name = device.vendor.name
if vendor_name == 'cisco':
vendor_name = 'ios'
driver = get_network_driver(vendor_name)
return driver
Raw
02_netdevices.py.diff
diff --git a/trigger/netdevices/__init__.py b/trigger/netdevices/__init__.py
index b9de772..1e0b5e9 100644
--- a/trigger/netdevices/__init__.py
+++ b/trigger/netdevices/__init__.py
@@ -251,6 +251,36 @@ class NetDevice(object):
# Set the correct line-ending per vendor
self.delimiter = self._set_delimiter()
+ self._setup_napalm()
+
+ def _setup_napalm(self):
+ from trigger.napalm_compat import get_driver_for_netdevice
+ self._driver = get_driver_for_netdevice(self)
+
+ from trigger.tacacsrc import Tacacsrc
+ creds = Tacacsrc().creds[settings.DEFAULT_REALM]
+
+ self._napalm_device = self._driver(
+ self.nodeName, creds.username, creds.password
+ )
+
+ def open(self):
+ """Open a connection to a me."""
+ self._napalm_device.open()
+
+ def run_command(self, command):
+ """
+ Run a command on the device using NAPALM underlying driver.
+
+ Currently hard-coded to Arista.
+ """
+ return self._napalm_device.device.enable(command)
+
+ def close(self):
+ """Close connection to me."""
+ self._napalm_device.close()
+
def _populate_data(self, data):
"""
Populate the custom attribute data
(END)
Raw
03_troll_test.py
#!/usr/bin/env python
"""
Simple test of using Trigger + asyncio + NAPALM
"""
import trollius as asyncio
from trollius import From, Return
from trigger.netdevices import NetDevices
@asyncio.coroutine
def execute_napalm(device, command):
"""Execute a command on a device using Naplam."""
def _run(device, command):
device.open()
result = device.run_command(command)
device.close()
return result
loop = asyncio.get_event_loop()
future = loop.run_in_executor(None, _run, device, command)
result = yield From(future)
raise Return(result)
@asyncio.coroutine
def map_command(devices, command, results):
"""Run a command on a list of devices."""
for device in devices:
results[device.nodeName] = yield From(execute_napalm(device, command))
raise Return(results)
def main(devices, command):
"""Do stuff and return results."""
loop = asyncio.get_event_loop()
results = {}
try:
loop.run_until_complete(map_command(devices, command, results))
finally:
loop.close()
return results
if __name__ == '__main__':
nd = NetDevices()
devices = nd.match(vendor='arista')
results = main(devices, 'show version')
# Results
"""
$ ipython -i ./troll_test.py
In [1]: results
Out[1]:
{'arista-sw1': [{'command': 'show version',
'encoding': 'json',
'result': {u'architecture': u'i386',
u'bootupTimestamp': 1453828173.42,
u'hardwareRevision': u'',
u'internalBuildId': u'f590eed4-1e66-43c6-8943-cee0390fbafe',
u'internalVersion': u'4.14.5F-2209869.4145F',
u'memFree': 77876,
u'memTotal': 996136,
u'modelName': u'vEOS',
u'serialNumber': u'',
u'systemMacAddress': u'08:00:27:6d:9b:ad',
u'version': u'4.14.5F'}}],
'switchname': [{'command': 'show version',
'encoding': 'json',
'result': {u'architecture': u'i386',
u'bootupTimestamp': 1453828173.42,
u'hardwareRevision': u'',
u'internalBuildId': u'f590eed4-1e66-43c6-8943-cee0390fbafe',
u'internalVersion': u'4.14.5F-2209869.4145F',
u'memFree': 77916,
u'memTotal': 996136,
u'modelName': u'vEOS',
u'serialNumber': u'',
u'systemMacAddress': u'08:00:27:6d:9b:ad',
u'version': u'4.14.5F'}}]}
"""
Raw
04_twist_test.py
#!/usr/bin/env python
"""
Simple test of using Trigger + Twisted + NAPALM
"""
from twisted.internet import task, defer, threads
from trigger.netdevices import NetDevices
@defer.inlineCallbacks
def execute_napalm(device, command):
"""Execute a command on a device using Napalm."""
def _run(device, command):
device.open()
result = device.run_command(command)
device.close()
return result
deferred = yield threads.deferToThread(_run, device, command)
defer.returnValue(deferred)
@defer.inlineCallbacks
def map_command(devices, command, results):
"""Run a command on a list of devices."""
for device in devices:
results[device.nodeName] = yield execute_napalm(device, command)
defer.returnValue(results)
def main(reactor, devices, command, results):
"""Do stuff and return results."""
results = map_command(devices, command, results)
return results
if __name__ == '__main__':
nd = NetDevices()
devices = nd.match(vendor='arista')
results = {}
task.react(main, [devices, 'show version', results])
# Results
"""
In [1]: results
Out[1]:
{'arista-sw1': [{'command': 'show version',
'encoding': 'json',
'result': {u'architecture': u'i386',
u'bootupTimestamp': 1453828173.43,
u'hardwareRevision': u'',
u'internalBuildId': u'f590eed4-1e66-43c6-8943-cee0390fbafe',
u'internalVersion': u'4.14.5F-2209869.4145F',
u'memFree': 28536,
u'memTotal': 996136,
u'modelName': u'vEOS',
u'serialNumber': u'',
u'systemMacAddress': u'08:00:27:6d:9b:ad',
u'version': u'4.14.5F'}}],
'switchname': [{'command': 'show version',
'encoding': 'json',
'result': {u'architecture': u'i386',
u'bootupTimestamp': 1453828173.43,
u'hardwareRevision': u'',
u'internalBuildId': u'f590eed4-1e66-43c6-8943-cee0390fbafe',
u'internalVersion': u'4.14.5F-2209869.4145F',
u'memFree': 28576,
u'memTotal': 996136,
u'modelName': u'vEOS',
u'serialNumber': u'',
u'systemMacAddress': u'08:00:27:6d:9b:ad',
u'version': u'4.14.5F'}}]}
"""
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment