|
|
@@ -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) |