|
|
@@ -56,17 +56,29 @@ def checkchanges(module): |
|
|
|
mod = importlib.import_module(module) |
|
|
|
mods = [ i for i in mod.__dict__.itervalues() if isinstance(i, SwitchConfig) ] |
|
|
|
|
|
|
|
res = [] |
|
|
|
|
|
|
|
for i in mods: |
|
|
|
switch = SNMPSwitch(i.host, i.community) |
|
|
|
portmapping = switch.getportmapping() |
|
|
|
invportmap = { y: x for x, y in portmapping.iteritems() } |
|
|
|
lufun = invportmap.__getitem__ |
|
|
|
|
|
|
|
portlist = getportlist(i._vlanconf, invportmap.__getitem__) |
|
|
|
portlist = getportlist(i._vlanconf, lufun) |
|
|
|
|
|
|
|
ports = set(portmapping.iterkeys()) |
|
|
|
|
|
|
|
if ports != portlist: |
|
|
|
raise ValueError('missing or extra ports found: %s' % `ports.symmetric_difference(portlist)`) |
|
|
|
raise ValueError('missing or extra ports found: %s' % |
|
|
|
`ports.symmetric_difference(portlist)`) |
|
|
|
|
|
|
|
pvidmap = getpvidmapping(i._vlanconf, lufun) |
|
|
|
switchpvid = switch.getpvid() |
|
|
|
|
|
|
|
res.extend(('setpvid', idx, vlan, switchpvid[idx]) for idx, vlan in |
|
|
|
pvidmap.iteritems() if switchpvid[idx] != vlan) |
|
|
|
|
|
|
|
return res |
|
|
|
|
|
|
|
def getpvidmapping(data, lookupfun): |
|
|
|
'''Return a mapping from vlan based table to a port: vlan |
|
|
@@ -253,28 +265,47 @@ class _TestMisc(unittest.TestCase): |
|
|
|
self.assertEqual(getportlist(data, lookup.__getitem__), |
|
|
|
set(xrange(1, 11)) | set(xrange(13, 20)) | set([30])) |
|
|
|
|
|
|
|
@mock.patch('vlanmang.SNMPSwitch.getpvid') |
|
|
|
@mock.patch('vlanmang.SNMPSwitch.getportmapping') |
|
|
|
@mock.patch('importlib.import_module') |
|
|
|
def test_checkchanges(self, imprt, portmapping): |
|
|
|
def tmp(*args, **kwargs): |
|
|
|
return self._test_data |
|
|
|
imprt.side_effect = tmp |
|
|
|
def test_checkchanges(self, imprt, portmapping, gpvid): |
|
|
|
# that import returns the test data |
|
|
|
imprt.side_effect = itertools.repeat(self._test_data) |
|
|
|
|
|
|
|
# that getportmapping returns the following dict |
|
|
|
ports = { x: 'g%d' % x for x in xrange(1, 24) } |
|
|
|
ports[30] = 'lag1' |
|
|
|
ports[31] = 'lag2' |
|
|
|
portmapping.side_effect = [ ports, ports ] |
|
|
|
portmapping.side_effect = itertools.repeat(ports) |
|
|
|
|
|
|
|
# that the switch's pvid returns |
|
|
|
spvid = { x: 283 for x in xrange(1, 24) } |
|
|
|
spvid[30] = 5 |
|
|
|
gpvid.side_effect = itertools.repeat(spvid) |
|
|
|
|
|
|
|
# the the extra port is caught |
|
|
|
self.assertRaises(ValueError, checkchanges, 'data') |
|
|
|
|
|
|
|
# that the functions were called |
|
|
|
imprt.assert_called_with('data') |
|
|
|
portmapping.assert_called() |
|
|
|
|
|
|
|
# XXX - check that an ignore statement is honored |
|
|
|
|
|
|
|
# delete the extra port |
|
|
|
del ports[31] |
|
|
|
|
|
|
|
checkchanges('data') |
|
|
|
res = checkchanges('data') |
|
|
|
|
|
|
|
validres = [ ('setpvid', x, 5, 283) for x in xrange(1, 9) ] + \ |
|
|
|
[ ('setpvid', 20, 1, 283), |
|
|
|
('setpvid', 21, 1, 283), |
|
|
|
('setpvid', 30, 1, 5), |
|
|
|
] |
|
|
|
|
|
|
|
self.assertEqual(set(res), set(validres)) |
|
|
|
|
|
|
|
_skipSwitchTests = False |
|
|
|
_skipSwitchTests = True |
|
|
|
|
|
|
|
class _TestSwitch(unittest.TestCase): |
|
|
|
def setUp(self): |
|
|
|