From 74cda5ac0d11376a1da3efca2a6e37540fee3a94 Mon Sep 17 00:00:00 2001 From: John-Mark Gurney Date: Mon, 23 Sep 2019 16:10:39 -0700 Subject: [PATCH] add and update docs, wrap lines at 80 cols.. --- vlanmang.py | 82 ++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 62 insertions(+), 20 deletions(-) diff --git a/vlanmang.py b/vlanmang.py index 8e78dac..7d10923 100644 --- a/vlanmang.py +++ b/vlanmang.py @@ -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)