Browse Source

add support for detecting missing vlans, and creating them if needed...

more weirdness on how to get the current set of vlans
ssh-lenovo
John-Mark Gurney 4 years ago
parent
commit
c00eb02867
2 changed files with 54 additions and 6 deletions
  1. +11
    -1
      NOTES.md
  2. +43
    -5
      vlanmang/__init__.py

+ 11
- 1
NOTES.md View File

@@ -25,7 +25,7 @@ Supported: authProtocol=usmHMACSHAAuthProtocol, privProtocol=usmDESPrivProtocol
NetGear GS724Tv4 NetGear GS724Tv4
---------------- ----------------


V6.3.1.4
### V6.3.1.4


Fixed user of admin. Auth key is the same as web admin password. Fixed user of admin. Auth key is the same as web admin password.


@@ -36,6 +36,11 @@ If you modify the egress ports, it will automatically ALSO set the bits for
untagged (despite them being "correct" before). Currently, you have to run the untagged (despite them being "correct" before). Currently, you have to run the
tool a second time to set the untagged config properly. tool a second time to set the untagged config properly.


### V6.3.1.19

Does not list VLANs in dot1qVlanStatus, must use dot1qVlanStaticName or
dot1qVlanCreationTime to get list of VLANs.

NetGear GS510TLP NetGear GS510TLP
---------------- ----------------


@@ -49,3 +54,8 @@ Supported: authProtocol=usmHMACSHAAuthProtocol, privProtocol=usmDESPrivProtocol
If you modify the egress ports, it will automatically ALSO set the bits for If you modify the egress ports, it will automatically ALSO set the bits for
untagged (despite them being "correct" before). Currently, you have to run the untagged (despite them being "correct" before). Currently, you have to run the
tool a second time to set the untagged config properly. tool a second time to set the untagged config properly.

NetGear GS108Tv2
----------------

Does not list VLAN 1 under dot1qVlanStaticName.

+ 43
- 5
vlanmang/__init__.py View File

@@ -158,6 +158,7 @@ class SwitchConfig(object):


# add in the ignore ports # add in the ignore ports
res.extend(self.ignports) res.extend(self.ignports)
#print(repr((res, self.ignports, lookupfun)))


# eliminate dups so that lookupfun isn't called as often # eliminate dups so that lookupfun isn't called as often
res = set(res) res = set(res)
@@ -239,8 +240,22 @@ def checkchanges(module):
vlans = i.vlanconf.keys() vlans = i.vlanconf.keys()
switch = SNMPSwitch(i.host, **i.authargs) switch = SNMPSwitch(i.host, **i.authargs)
switchpvid = switch.getpvid() switchpvid = switch.getpvid()

# check that vlans exist
missingvlans = set(vlans) - set(switch.getvlans())
#print(repr((vlans, missingvlans)))
if missingvlans:
# need to create vlans first
# XXX - if multiple switches need a vlan,
# do it in one pass
return [ (switch, name, 'createvlan', x, '', '') for
x in missingvlans ]


# Get switch port names for lookup
portmapping = switch.getportmapping() portmapping = switch.getportmapping()
invportmap = { y: x for x, y in portmapping.items() } invportmap = { y: x for x, y in portmapping.items() }
#print(repr(invportmap))
lufun = invportmap.__getitem__ lufun = invportmap.__getitem__


# get complete set of ports # get complete set of ports
@@ -250,8 +265,10 @@ def checkchanges(module):


# make sure switch agrees w/ them all # make sure switch agrees w/ them all
if ports != portlist: if ports != portlist:
raise ValueError('missing or extra ports found: %s' %
repr(ports.symmetric_difference(portlist)))
raise ValueError(
'missing or extra ports found on %s: %s' %
(repr(name),
repr(ports.symmetric_difference(portlist))))


# compare settings # compare settings
settings = list(i.settings) settings = list(i.settings)
@@ -532,8 +549,10 @@ class SNMPSwitch(object):
def getvlans(self): def getvlans(self):
'''Return an iterator with all the vlan ids.''' '''Return an iterator with all the vlan ids.'''


# GS724Tv4 ProSafe 24-port Gigabit Ethernet Smart Switch, 6.3.1.19, B1.0.0.4

return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB', return (x[0][-1] for x in self._walk('Q-BRIDGE-MIB',
'dot1qVlanStatus'))
'dot1qVlanCreationTime'))


def staticvlans(self): def staticvlans(self):
'''Return an iterator of the staticly defined/configured '''Return an iterator of the staticly defined/configured
@@ -823,13 +842,14 @@ class _TestMisc(unittest.TestCase):
privProtocol=usmDESPrivProtocol) privProtocol=usmDESPrivProtocol)


#@unittest.skip('foo') #@unittest.skip('foo')
@mock.patch('vlanmang.SNMPSwitch.getvlans')
@mock.patch('vlanmang.SNMPSwitch.getsettings') @mock.patch('vlanmang.SNMPSwitch.getsettings')
@mock.patch('vlanmang.SNMPSwitch.getuntagged') @mock.patch('vlanmang.SNMPSwitch.getuntagged')
@mock.patch('vlanmang.SNMPSwitch.getegress') @mock.patch('vlanmang.SNMPSwitch.getegress')
@mock.patch('vlanmang.SNMPSwitch.getpvid') @mock.patch('vlanmang.SNMPSwitch.getpvid')
@mock.patch('vlanmang.SNMPSwitch.getportmapping') @mock.patch('vlanmang.SNMPSwitch.getportmapping')
@mock.patch('importlib.import_module') @mock.patch('importlib.import_module')
def test_checkchanges(self, imprt, portmapping, gpvid, gegress, guntagged, gsettings):
def test_checkchanges(self, imprt, portmapping, gpvid, gegress, guntagged, gsettings, gvlans):
# that import returns the test data # that import returns the test data
imprt.side_effect = itertools.repeat(self._test_data) imprt.side_effect = itertools.repeat(self._test_data)


@@ -845,6 +865,24 @@ class _TestMisc(unittest.TestCase):
spvid[30] = 5 spvid[30] = 5
gpvid.side_effect = itertools.repeat(spvid) gpvid.side_effect = itertools.repeat(spvid)


# if the switch is missing vlan 283
gvlans.return_value = iter([ 1, 5 ])

res = checkchanges('data')
validres = [ ('createvlan', 283, '', '') ]

# make sure it needs to get created

# check initial params
self.assertTrue(all(isinstance(x[0], SNMPSwitch) for x in res))
self.assertTrue(all(x[1] == 'distswitch' for x in res))

res = [ x[2:] for x in res ]
self.assertEqual(res, validres)

# now that all vlans are present
gvlans.return_value = [ 1, 5, 283 ]

# the the extra port is caught # the the extra port is caught
self.assertRaises(ValueError, checkchanges, 'data') self.assertRaises(ValueError, checkchanges, 'data')


@@ -986,7 +1024,7 @@ class _TestSNMPSwitch(unittest.TestCase):
self.assertEqual(res, { x: _octstrtobits(lookup[x]) for x in self.assertEqual(res, { x: _octstrtobits(lookup[x]) for x in
range(1, 10) }) range(1, 10) })


_skipSwitchTests = False
_skipSwitchTests = True


class _TestSwitch(unittest.TestCase): class _TestSwitch(unittest.TestCase):
def setUp(self): def setUp(self):


Loading…
Cancel
Save