|
|
@@ -5,7 +5,9 @@ from pysnmp.hlapi import * |
|
|
|
from pysnmp.smi.builder import MibBuilder |
|
|
|
from pysnmp.smi.view import MibViewController |
|
|
|
|
|
|
|
import importlib |
|
|
|
import itertools |
|
|
|
import mock |
|
|
|
import random |
|
|
|
import unittest |
|
|
|
|
|
|
@@ -32,6 +34,32 @@ _mvc = MibViewController(_mbuilder) |
|
|
|
# LLDP: |
|
|
|
# 1.0.8802.1.1.2.1.4.1.1 aka LLDP-MIB, lldpRemTable |
|
|
|
|
|
|
|
class SwitchConfig(object): |
|
|
|
def __init__(self, host, community, vlanconf): |
|
|
|
self._host = host |
|
|
|
self._community = community |
|
|
|
self._vlanconf = vlanconf |
|
|
|
|
|
|
|
@property |
|
|
|
def host(self): |
|
|
|
return self._host |
|
|
|
|
|
|
|
@property |
|
|
|
def community(self): |
|
|
|
return self._community |
|
|
|
|
|
|
|
@property |
|
|
|
def vlanconf(self): |
|
|
|
return self._vlanconf |
|
|
|
|
|
|
|
def checkchanges(module): |
|
|
|
mod = importlib.import_module(module) |
|
|
|
mods = [ i for i in mod.__dict__.itervalues() if isinstance(i, SwitchConfig) ] |
|
|
|
|
|
|
|
for i in mods: |
|
|
|
switch = SNMPSwitch(i.host, i.community) |
|
|
|
portmapping = switch.getportmapping() |
|
|
|
|
|
|
|
def getpvidmapping(data, lookupfun): |
|
|
|
'''Return a mapping from vlan based table to a port: vlan |
|
|
|
dictionary.''' |
|
|
@@ -134,6 +162,12 @@ class SNMPSwitch(object): |
|
|
|
for varBind in varBinds: |
|
|
|
yield varBind |
|
|
|
|
|
|
|
def getportmapping(self): |
|
|
|
'''Return a port name mapping. Keys are the port index |
|
|
|
and the value is the name from the ifName entry.''' |
|
|
|
|
|
|
|
return dict((x[0][-1], str(x[1])) for x in self._walk('IF-MIB', 'ifName')) |
|
|
|
|
|
|
|
def findport(self, name): |
|
|
|
'''Look up a port name and return it's port index. This |
|
|
|
looks up via the ifName table in IF-MIB.''' |
|
|
@@ -170,6 +204,11 @@ class SNMPSwitch(object): |
|
|
|
return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB', 'dot1qVlanStaticName')) |
|
|
|
|
|
|
|
class _TestMisc(unittest.TestCase): |
|
|
|
def setUp(self): |
|
|
|
import test_data |
|
|
|
|
|
|
|
self._test_data = test_data |
|
|
|
|
|
|
|
def test_pvid(self): |
|
|
|
data = { |
|
|
|
1: { |
|
|
@@ -199,6 +238,19 @@ class _TestMisc(unittest.TestCase): |
|
|
|
self.assertEqual(getportlist(data, lookup.__getitem__), |
|
|
|
set(xrange(1, 11)) | set(xrange(13, 20)) | set([30])) |
|
|
|
|
|
|
|
@mock.patch('vlanmang.SNMPSwitch.getportmapping') |
|
|
|
@mock.patch('importlib.import_module') |
|
|
|
def test_checkchanges(self, imprt, portmapping): |
|
|
|
def tmp(*args, **kwargs): |
|
|
|
print `args` |
|
|
|
return self._test_data |
|
|
|
imprt.side_effect = tmp |
|
|
|
|
|
|
|
checkchanges('data') |
|
|
|
|
|
|
|
imprt.assert_called_with('data') |
|
|
|
portmapping.assert_called() |
|
|
|
|
|
|
|
_skipSwitchTests = True |
|
|
|
|
|
|
|
class _TestSwitch(unittest.TestCase): |
|
|
@@ -206,17 +258,27 @@ class _TestSwitch(unittest.TestCase): |
|
|
|
args = open('test.creds').read().split() |
|
|
|
self.switch = SNMPSwitch(*args) |
|
|
|
|
|
|
|
def test_unpacktable(self): |
|
|
|
pass |
|
|
|
switchmodel = self.switch._get(('ENTITY-MIB', |
|
|
|
'entPhysicalModelName', 1)) |
|
|
|
if switchmodel != 'GS108T smartSwitch' or \ |
|
|
|
_skipSwitchTests: # pragma: no cover |
|
|
|
self.skipTest('Need a GS108T switch to run these tests') |
|
|
|
|
|
|
|
@unittest.skipIf(_skipSwitchTests, 'slow') |
|
|
|
def test_misc(self): |
|
|
|
switch = self.switch |
|
|
|
|
|
|
|
self.assertEqual(switch.findport('g1'), 1) |
|
|
|
self.assertEqual(switch.findport('l1'), 14) |
|
|
|
|
|
|
|
@unittest.skipIf(_skipSwitchTests, 'slow') |
|
|
|
def test_portnames(self): |
|
|
|
switch = self.switch |
|
|
|
|
|
|
|
resp = dict((x, 'g%d' % x) for x in xrange(1, 9)) |
|
|
|
resp.update({ 13: 'cpu' }) |
|
|
|
resp.update((x, 'l%d' % (x - 13)) for x in xrange(14, 18)) |
|
|
|
|
|
|
|
self.assertEqual(switch.getportmapping(), resp) |
|
|
|
|
|
|
|
def test_vlan(self): |
|
|
|
switch = self.switch |
|
|
|
|
|
|
|