diff --git a/vlanmang.py b/vlanmang.py index 52b286b..94c796b 100644 --- a/vlanmang.py +++ b/vlanmang.py @@ -156,7 +156,7 @@ def checkchanges(module): if not _cmpbits(switchuntagged[i], untagged[i]): res.append(('setuntagged', i, untagged[i], switchuntagged[i])) - return res + return res, switch def getidxs(lst, lookupfun): return [ lookupfun(i) if isinstance(i, str) else i for i in lst ] @@ -314,6 +314,9 @@ class SNMPSwitch(object): return { x[0][-1]: int(x[1]) for x in self._walk('Q-BRIDGE-MIB', 'dot1qPvid') } + def setpvid(self, port, vlan): + self._set(('Q-BRIDGE-MIB', 'dot1qPvid', int(port)), Gauge32(vlan)) + def getegress(self, *vlans): r = { x[-1]: _octstrtobits(y) for x, y in self._getmany(*(('Q-BRIDGE-MIB', @@ -321,6 +324,11 @@ class SNMPSwitch(object): return r + def setegress(self, vlan, ports): + value = OctetString.fromBinaryString(ports) + self._set(('Q-BRIDGE-MIB', 'dot1qVlanStaticEgressPorts', + int(vlan)), value) + def getuntagged(self, *vlans): r = { x[-1]: _octstrtobits(y) for x, y in self._getmany(*(('Q-BRIDGE-MIB', @@ -328,8 +336,37 @@ class SNMPSwitch(object): return r + def setuntagged(self, vlan, ports): + value = OctetString.fromBinaryString(ports) + self._set(('Q-BRIDGE-MIB', 'dot1qVlanStaticUntaggedPorts', + int(vlan)), value) + if __name__ == '__main__': # pragma: no cover - print `checkchanges('data')` + import pprint + import sys + + changes, switch = checkchanges('data') + pprint.pprint(changes) + + res = raw_input('Apply the changes? (type yes to apply): ') + if res != 'yes': + print 'not applying changes.' + sys.exit(1) + + print 'applying...' + failed = [] + for verb, vlan, arg, oldarg in changes: + print '%s: %s %s' % (verb, vlan, `arg`) + try: + pass + except Exception as e: + print 'failed' + failed.append((verb, vlan, arg, 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` class _TestMisc(unittest.TestCase): def setUp(self): @@ -462,7 +499,9 @@ class _TestMisc(unittest.TestCase): 283: '00000000111111111110011', } ] - res = checkchanges('data') + res, switch = checkchanges('data') + + self.assertIsInstance(switch, SNMPSwitch) validres = [ ('setpvid', x, 5, 283) for x in xrange(1, 9) ] + \ [ ('setpvid', 20, 1, 283), @@ -475,7 +514,7 @@ class _TestMisc(unittest.TestCase): self.assertEqual(set(res), set(validres)) -_skipSwitchTests = True +_skipSwitchTests = False class _TestSwitch(unittest.TestCase): def setUp(self): @@ -558,10 +597,30 @@ class _TestSwitch(unittest.TestCase): # Create test vlan switch.createvlan(testvlan, testname) + testport = None try: # make sure the test vlan was created self.assertIn(testvlan, set(switch.staticvlans())) self.assertEqual(testname, switch.getvlanname(testvlan)) + + switch.setegress(testvlan, '00100') + + pvidmap = switch.getpvid() + testport = 3 + + egressports = switch.getegress(testvlan) + self.assertEqual(egressports[testvlan], '00100000' + '0' * 8 * 4) + + switch.setuntagged(testvlan, '00100') + + untaggedports = switch.getuntagged(testvlan) + self.assertEqual(untaggedports[testvlan], '00100000' + '0' * 8 * 4) + + switch.setpvid(testport, testvlan) + + self.assertEqual(switch.getpvid()[testport], testvlan) finally: + if testport: + switch.setpvid(testport, pvidmap[3]) switch.deletevlan(testvlan)