|
|
@@ -231,8 +231,8 @@ def getpvidmapping(data, lookupfun): |
|
|
|
return dict(res) |
|
|
|
|
|
|
|
def getegress(data, lookupfun): |
|
|
|
'''Return a dictionary, keyed by VLAN id with a string of the ports |
|
|
|
that need to be enagled for egress. This include both tagged and |
|
|
|
'''Return a dictionary, keyed by VLAN id with a bit string of ports |
|
|
|
that need to be enabled for egress. This include both tagged and |
|
|
|
untagged traffic.''' |
|
|
|
|
|
|
|
r = {} |
|
|
@@ -243,6 +243,9 @@ def getegress(data, lookupfun): |
|
|
|
return r |
|
|
|
|
|
|
|
def getuntagged(data, lookupfun): |
|
|
|
'''Return a dictionary, keyed by VLAN id with a bit string of ports |
|
|
|
that need to be enabled for untagged egress.''' |
|
|
|
|
|
|
|
r = {} |
|
|
|
for id in data: |
|
|
|
r[id] = _intstobits(*getidxs(data[id].get('u', []), lookupfun)) |
|
|
@@ -265,7 +268,8 @@ class SNMPSwitch(object): |
|
|
|
[ 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 woids))) |
|
|
|
next(getCmd(self._eng, self._cd, self._targ, |
|
|
|
ContextData(), *(ObjectType(oid) for oid in woids))) |
|
|
|
|
|
|
|
if errorInd: # pragma: no cover |
|
|
|
raise ValueError(errorIndication) |
|
|
@@ -302,7 +306,8 @@ class SNMPSwitch(object): |
|
|
|
value = OctetString(value) |
|
|
|
|
|
|
|
errorInd, errorStatus, errorIndex, varBinds = \ |
|
|
|
next(setCmd(self._eng, self._cd, self._targ, ContextData(), ObjectType(oid, value))) |
|
|
|
next(setCmd(self._eng, self._cd, self._targ, |
|
|
|
ContextData(), ObjectType(oid, value))) |
|
|
|
|
|
|
|
if errorInd: # pragma: no cover |
|
|
|
raise ValueError(errorIndication) |
|
|
@@ -331,25 +336,30 @@ class SNMPSwitch(object): |
|
|
|
if errorInd: # pragma: no cover |
|
|
|
raise ValueError(errorInd) |
|
|
|
elif errorStatus: # pragma: no cover |
|
|
|
raise ValueError('%s at %s' % (errorStatus.prettyPrint(), errorIndex and varBinds[int(errorIndex)-1][0] or '?')) |
|
|
|
raise ValueError('%s at %s' % |
|
|
|
(errorStatus.prettyPrint(), errorIndex and |
|
|
|
varBinds[int(errorIndex)-1][0] or '?')) |
|
|
|
else: |
|
|
|
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.''' |
|
|
|
and the value is the name from the IF-MIB::ifName entry.''' |
|
|
|
|
|
|
|
return { x[0][-1]: str(x[1]) for x in self._walk('IF-MIB', 'ifName') } |
|
|
|
return { 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.''' |
|
|
|
|
|
|
|
return [ x[0][-1] for x in self._walk('IF-MIB', 'ifName') if str(x[1]) == name ][0] |
|
|
|
return [ x[0][-1] for x in self._walk('IF-MIB', 'ifName') if |
|
|
|
str(x[1]) == name ][0] |
|
|
|
|
|
|
|
def getvlanname(self, vlan): |
|
|
|
'''Return the name for the vlan.''' |
|
|
|
'''Return the name for the vlan. This returns the value in |
|
|
|
Q-BRIDGE-MIB:dot1qVlanStaticName.''' |
|
|
|
|
|
|
|
v = self._get(('Q-BRIDGE-MIB', 'dot1qVlanStaticName', vlan)) |
|
|
|
|
|
|
@@ -369,25 +379,36 @@ class SNMPSwitch(object): |
|
|
|
def getvlans(self): |
|
|
|
'''Return an iterator with all the vlan ids.''' |
|
|
|
|
|
|
|
return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB', 'dot1qVlanStatus')) |
|
|
|
return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB', |
|
|
|
'dot1qVlanStatus')) |
|
|
|
|
|
|
|
def staticvlans(self): |
|
|
|
'''Return an iterator of the staticly defined/configured |
|
|
|
vlans. This sometimes excludes special built in vlans, |
|
|
|
like vlan 1.''' |
|
|
|
|
|
|
|
return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB', 'dot1qVlanStaticName')) |
|
|
|
return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB', |
|
|
|
'dot1qVlanStaticName')) |
|
|
|
|
|
|
|
def getpvid(self): |
|
|
|
'''Returns a dictionary w/ the interface index as the key, |
|
|
|
and the pvid of the interface.''' |
|
|
|
|
|
|
|
return { x[0][-1]: int(x[1]) for x in self._walk('Q-BRIDGE-MIB', 'dot1qPvid') } |
|
|
|
return { x[0][-1]: int(x[1]) for x in self._walk('Q-BRIDGE-MIB', |
|
|
|
'dot1qPvid') } |
|
|
|
|
|
|
|
def setpvid(self, port, vlan): |
|
|
|
'''Set the port's Pvid to vlan. This means that any packet |
|
|
|
received by the port that is untagged, will be routed the |
|
|
|
the vlan.''' |
|
|
|
|
|
|
|
self._set(('Q-BRIDGE-MIB', 'dot1qPvid', int(port)), Gauge32(vlan)) |
|
|
|
|
|
|
|
def getegress(self, *vlans): |
|
|
|
'''Get a dictionary keyed by the specified VLANs, where each |
|
|
|
value is a bit string that preresents what ports that |
|
|
|
particular VLAN will be transmitted on.''' |
|
|
|
|
|
|
|
r = { x[-1]: _octstrtobits(y) for x, y in |
|
|
|
self._getmany(*(('Q-BRIDGE-MIB', |
|
|
|
'dot1qVlanStaticEgressPorts', x) for x in vlans)) } |
|
|
@@ -395,11 +416,21 @@ class SNMPSwitch(object): |
|
|
|
return r |
|
|
|
|
|
|
|
def setegress(self, vlan, ports): |
|
|
|
'''Set the ports which the specified VLAN will have packets |
|
|
|
transmitted as either tagged, if unset in untagged, or |
|
|
|
untagged, if set in untagged, to bit bit string specified |
|
|
|
by ports.''' |
|
|
|
|
|
|
|
value = OctetString.fromBinaryString(ports) |
|
|
|
self._set(('Q-BRIDGE-MIB', 'dot1qVlanStaticEgressPorts', |
|
|
|
int(vlan)), value) |
|
|
|
|
|
|
|
def getuntagged(self, *vlans): |
|
|
|
'''Get a dictionary keyed by the specified VLANs, where each |
|
|
|
value is a bit string that preresents what ports that |
|
|
|
particular VLAN will be transmitted on as an untagged |
|
|
|
packet.''' |
|
|
|
|
|
|
|
r = { x[-1]: _octstrtobits(y) for x, y in |
|
|
|
self._getmany(*(('Q-BRIDGE-MIB', |
|
|
|
'dot1qVlanStaticUntaggedPorts', x) for x in vlans)) } |
|
|
@@ -407,6 +438,9 @@ class SNMPSwitch(object): |
|
|
|
return r |
|
|
|
|
|
|
|
def setuntagged(self, vlan, ports): |
|
|
|
'''Set the ports which the specified VLAN will have packets |
|
|
|
transmitted as untagged to the bit string specified by ports.''' |
|
|
|
|
|
|
|
value = OctetString.fromBinaryString(ports) |
|
|
|
self._set(('Q-BRIDGE-MIB', 'dot1qVlanStaticUntaggedPorts', |
|
|
|
int(vlan)), value) |
|
|
@@ -509,7 +543,8 @@ class _TestMisc(unittest.TestCase): |
|
|
|
self.assertEqual(res, check) |
|
|
|
|
|
|
|
self.assertEqual(swconf.getportlist(lufun), |
|
|
|
set(xrange(1, 11)) | set(xrange(13, 20)) | set(lookup.values())) |
|
|
|
set(xrange(1, 11)) | set(xrange(13, 20)) | |
|
|
|
set(lookup.values())) |
|
|
|
|
|
|
|
checkegress = { |
|
|
|
1: '1000111001001111111' + '0' * (30 - 20) + '1', |
|
|
@@ -584,9 +619,12 @@ class _TestMisc(unittest.TestCase): |
|
|
|
[ ('setpvid', 20, 1, 283), |
|
|
|
('setpvid', 21, 1, 283), |
|
|
|
('setpvid', 30, 1, 5), |
|
|
|
('setegress', 1, '0' * 19 + '11' + '0' * 8 + '1', '1' * 10), |
|
|
|
('setuntagged', 1, '0' * 19 + '11' + '0' * 8 + '1', '1' * 10), |
|
|
|
('setegress', 5, '1' * 8 + '0' * 11 + '11' + '0' * 8 + '1', '1' * 10), |
|
|
|
('setegress', 1, '0' * 19 + '11' + '0' * 8 + '1', |
|
|
|
'1' * 10), |
|
|
|
('setuntagged', 1, '0' * 19 + '11' + '0' * 8 + '1', |
|
|
|
'1' * 10), |
|
|
|
('setegress', 5, '1' * 8 + '0' * 11 + '11' + '0' * 8 + |
|
|
|
'1', '1' * 10), |
|
|
|
] |
|
|
|
|
|
|
|
self.assertEqual(set(res), set(validres)) |
|
|
@@ -631,7 +669,8 @@ class _TestSNMPSwitch(unittest.TestCase): |
|
|
|
else: |
|
|
|
#import pdb; pdb.set_trace() |
|
|
|
[ oid.resolveWithMib(_mvc) for oid in oids ] |
|
|
|
oids = [ ObjectType(x[0], OctetString(lookup[x[0][-1]])) for x in oids ] |
|
|
|
oids = [ ObjectType(x[0], |
|
|
|
OctetString(lookup[x[0][-1]])) for x in oids ] |
|
|
|
[ oid.resolveWithMib(_mvc) for oid in oids ] |
|
|
|
res = ( None, None, None, oids ) |
|
|
|
|
|
|
@@ -643,7 +682,8 @@ class _TestSNMPSwitch(unittest.TestCase): |
|
|
|
res = switch.getegress(*xrange(1, 10)) |
|
|
|
|
|
|
|
# will still return the complete set of results |
|
|
|
self.assertEqual(res, { x: _octstrtobits(lookup[x]) for x in xrange(1, 10) }) |
|
|
|
self.assertEqual(res, { x: _octstrtobits(lookup[x]) for x in |
|
|
|
xrange(1, 10) }) |
|
|
|
|
|
|
|
_skipSwitchTests = True |
|
|
|
|
|
|
@@ -741,12 +781,14 @@ class _TestSwitch(unittest.TestCase): |
|
|
|
testport = 3 |
|
|
|
|
|
|
|
egressports = switch.getegress(testvlan) |
|
|
|
self.assertEqual(egressports[testvlan], '00100000' + '0' * 8 * 4) |
|
|
|
self.assertEqual(egressports[testvlan], '00100000' + |
|
|
|
'0' * 8 * 4) |
|
|
|
|
|
|
|
switch.setuntagged(testvlan, '00100') |
|
|
|
|
|
|
|
untaggedports = switch.getuntagged(testvlan) |
|
|
|
self.assertEqual(untaggedports[testvlan], '00100000' + '0' * 8 * 4) |
|
|
|
self.assertEqual(untaggedports[testvlan], '00100000' + |
|
|
|
'0' * 8 * 4) |
|
|
|
|
|
|
|
switch.setpvid(testport, testvlan) |
|
|
|
|
|
|
|