Browse Source

add ignore ports to test case... if a _getmany request is too big and

we can split it, do so..  This happened w/ a switch when using SNMPv3..
need to write a test case for this...
ssh-lenovo
John-Mark Gurney 5 years ago
parent
commit
d8683f62b0
2 changed files with 29 additions and 12 deletions
  1. +1
    -1
      test_data.py
  2. +28
    -11
      vlanmang.py

+ 1
- 1
test_data.py View File

@@ -26,4 +26,4 @@ distributionswitch = {
},
}

distswitch = vlanmang.SwitchConfig('192.168.0.58', 'private', distributionswitch)
distswitch = vlanmang.SwitchConfig('192.168.0.58', 'private', distributionswitch, [ 'lag2' ])

+ 28
- 11
vlanmang.py View File

@@ -192,21 +192,31 @@ def getuntagged(data, lookupfun):
class SNMPSwitch(object):
'''A class for manipulating switches via standard SNMP MIBs.'''

def __init__(self, host, community):
def __init__(self, host, auth):
self._eng = SnmpEngine()
self._cd = CommunityData(community, mpModel=0)
if isinstance(auth, str):
self._cd = CommunityData(auth, mpModel=0)
else:
self._cd = auth
self._targ = UdpTransportTarget((host, 161))

def _getmany(self, *oids):
oids = [ ObjectIdentity(*oid) for oid in oids ]
[ oid.resolveWithMib(_mvc) for oid in oids ]
woids = [ ObjectIdentity(*oid) for oid in oids ]
[ oid.resolveWithMib(_mvc) for oid in woids ]

errorInd, errorStatus, errorIndex, varBinds = \
next(getCmd(self._eng, self._cd, self._targ, ContextData(), *(ObjectType(oid) for oid in oids)))
next(getCmd(self._eng, self._cd, self._targ, ContextData(), *(ObjectType(oid) for oid in woids)))

if errorInd: # pragma: no cover
raise ValueError(errorIndication)
elif errorStatus: # pragma: no cover
elif errorStatus:
if str(errorStatus) == 'tooBig' and len(oids) > 1:
# split the request in two
pivot = len(oids) / 2
a = self._getmany(*oids[:pivot])
b = self._getmany(*oids[pivot:])
return a + b

raise ValueError('%s at %s' %
(errorStatus.prettyPrint(), errorIndex and
varBinds[int(errorIndex)-1][0] or '?'))
@@ -346,6 +356,11 @@ if __name__ == '__main__': # pragma: no cover
import sys

changes, switch = checkchanges('data')

if not changes:
print 'No changes to apply.'
sys.exit(0)

pprint.pprint(changes)

res = raw_input('Apply the changes? (type yes to apply): ')
@@ -355,18 +370,20 @@ if __name__ == '__main__': # pragma: no cover

print 'applying...'
failed = []
for verb, vlan, arg, oldarg in changes:
print '%s: %s %s' % (verb, vlan, `arg`)
for verb, arg1, arg2, oldarg in changes:
print '%s: %s %s' % (verb, arg1, `arg2`)
try:
fun = getattr(switch, verb)
fun(arg1, arg2)
pass
except Exception as e:
print 'failed'
failed.append((verb, vlan, arg, e))
failed.append((verb, arg1, arg2, e))

if failed:
print '%d failed to apply, they are:' % len(failed)
for verb, vlan, arg, e in failed:
print '%s: %s %s: %s' % verb, vlan, arg, `e`
for verb, arg1, arg2, e in failed:
print '%s: %s %s: %s' % (verb, arg1, arg2, `e`)

class _TestMisc(unittest.TestCase):
def setUp(self):


Loading…
Cancel
Save