|
|
@@ -52,6 +52,16 @@ class SwitchConfig(object): |
|
|
|
def vlanconf(self): |
|
|
|
return self._vlanconf |
|
|
|
|
|
|
|
def intstobits(*ints): |
|
|
|
v = 0 |
|
|
|
for i in ints: |
|
|
|
v |= 1 << i |
|
|
|
|
|
|
|
r = list(bin(v)[2:-1]) |
|
|
|
r.reverse() |
|
|
|
|
|
|
|
return ''.join(r) |
|
|
|
|
|
|
|
def checkchanges(module): |
|
|
|
mod = importlib.import_module(module) |
|
|
|
mods = [ i for i in mod.__dict__.itervalues() if isinstance(i, SwitchConfig) ] |
|
|
@@ -80,6 +90,9 @@ def checkchanges(module): |
|
|
|
|
|
|
|
return res |
|
|
|
|
|
|
|
def getidxs(lst, lookupfun): |
|
|
|
return [ lookupfun(i) if isinstance(i, str) else i for i in lst ] |
|
|
|
|
|
|
|
def getpvidmapping(data, lookupfun): |
|
|
|
'''Return a mapping from vlan based table to a port: vlan |
|
|
|
dictionary.''' |
|
|
@@ -93,6 +106,21 @@ def getpvidmapping(data, lookupfun): |
|
|
|
|
|
|
|
return dict(res) |
|
|
|
|
|
|
|
def getegress(data, lookupfun): |
|
|
|
r = {} |
|
|
|
for id in data: |
|
|
|
r[id] = intstobits(*(getidxs(data[id]['u'], lookupfun) + |
|
|
|
getidxs(data[id].get('t', []), lookupfun))) |
|
|
|
|
|
|
|
return r |
|
|
|
|
|
|
|
def getuntagged(data, lookupfun): |
|
|
|
r = {} |
|
|
|
for id in data: |
|
|
|
r[id] = intstobits(*getidxs(data[id]['u'], lookupfun)) |
|
|
|
|
|
|
|
return r |
|
|
|
|
|
|
|
def getportlist(data, lookupfun): |
|
|
|
'''Return a set of all the ports indexes in data.''' |
|
|
|
|
|
|
@@ -236,35 +264,58 @@ class _TestMisc(unittest.TestCase): |
|
|
|
|
|
|
|
self._test_data = test_data |
|
|
|
|
|
|
|
def test_pvid(self): |
|
|
|
def test_intstobits(self): |
|
|
|
self.assertEqual(intstobits(1, 5, 10), '1000100001') |
|
|
|
self.assertEqual(intstobits(3, 4, 9), '001100001') |
|
|
|
|
|
|
|
def test_pvidegressuntagged(self): |
|
|
|
data = { |
|
|
|
1: { |
|
|
|
'u': [ 1, 5, 10 ] + range(13, 20) |
|
|
|
'u': [ 1, 5, 10 ] + range(13, 20), |
|
|
|
't': [ 'lag2', 6, 7 ] |
|
|
|
}, |
|
|
|
10: { |
|
|
|
'u': [ 2, 3, 6, 7, 8, 'lag2' ], |
|
|
|
}, |
|
|
|
13: { |
|
|
|
'u': [ 4, 9 ], |
|
|
|
't': [ 'lag2', 6, 7 ] |
|
|
|
}, |
|
|
|
} |
|
|
|
lookup = { |
|
|
|
'lag2': 30 |
|
|
|
} |
|
|
|
lufun = lookup.__getitem__ |
|
|
|
|
|
|
|
check = dict(itertools.chain(enumerate([ 1, 10, 10, 13, 1, 10, |
|
|
|
10, 10, 13, 1 ], 1), enumerate([ 1 ] * 7, 13), |
|
|
|
[ (30, 10) ])) |
|
|
|
|
|
|
|
# That a pvid mapping |
|
|
|
res = getpvidmapping(data, lookup.__getitem__) |
|
|
|
res = getpvidmapping(data, lufun) |
|
|
|
|
|
|
|
# is correct |
|
|
|
self.assertEqual(res, check) |
|
|
|
|
|
|
|
self.assertEqual(getportlist(data, lookup.__getitem__), |
|
|
|
self.assertEqual(getportlist(data, lufun), |
|
|
|
set(xrange(1, 11)) | set(xrange(13, 20)) | set([30])) |
|
|
|
|
|
|
|
checkegress = { |
|
|
|
1: '1000111001001111111' + '0' * (30 - 20) + '1', |
|
|
|
10: '01100111' + '0' * (30 - 9) + '1', |
|
|
|
13: '000101101' + '0' * (30 - 10) + '1', |
|
|
|
} |
|
|
|
|
|
|
|
self.assertEqual(getegress(data, lufun), checkegress) |
|
|
|
|
|
|
|
checkuntagged = { |
|
|
|
1: '1000100001001111111', |
|
|
|
10: '01100111' + '0' * (30 - 9) + '1', |
|
|
|
13: '000100001', |
|
|
|
} |
|
|
|
self.assertEqual(getuntagged(data, lufun), checkuntagged) |
|
|
|
|
|
|
|
@unittest.skip('foo') |
|
|
|
@mock.patch('vlanmang.SNMPSwitch.getpvid') |
|
|
|
@mock.patch('vlanmang.SNMPSwitch.getportmapping') |
|
|
|
@mock.patch('importlib.import_module') |
|
|
@@ -301,6 +352,9 @@ class _TestMisc(unittest.TestCase): |
|
|
|
[ ('setpvid', 20, 1, 283), |
|
|
|
('setpvid', 21, 1, 283), |
|
|
|
('setpvid', 30, 1, 5), |
|
|
|
('setegress', 1, '0' * 19 + '11' + '0' * 8 + '1', ''), |
|
|
|
('setuntagged', 1, '0' * 19 + '11' + '0' * 8 + '1', ''), |
|
|
|
('setegress', 5, '1' * 8 + '0' * 11 + '11' + '0' * 8 + '1', ''), |
|
|
|
] |
|
|
|
|
|
|
|
self.assertEqual(set(res), set(validres)) |
|
|
|