add support for Lenovo CE0128TB.. minor bug fix for a NetGear

when ports are changed, set both egress and untagged... a switch
reset/sets the untagged bit upon egress change...

The Lenovo's SNMP get is broken w/ multiple OIDs, so add a hack
to detect that the switch is a Lenovo, and don't use getmany (well
call getmany_real w/ one OID at a time).

NOTES.md

@@ -7,6 +7,20 @@ Support auto creating missing VLANs with names from a table.
 Compatability
 =============
 
+Lenovo CE0128TB
+---------------
+
+V8.4.3.14
+
+SNMP get is broken when fetching multiple values at once.  This is worked
+around by doing individual gets, so this switch will be a bit slower.
+
+Note: The switch defaults the 10G port to stacking:
+https://web.archive.org/web/20230804012801/https://datacentersupport.lenovo.com/us/en/products/networking/campus-switch/switch-ce0128tb/solutions/ht510171-how-to-change-10-g-ports-on-a-lenovo-ce-switch-from-stack-to-ethernet
+
+The SNMP MIB for it is
+`LENOVO-INVENTORY-MIB::agentInventoryStackPortConfiguredStackMode`.
+
 NetGear GS724TS
 ---------------
 

README.md

@@ -70,12 +70,12 @@ mibdump.py --mib-source=mibdir <MIB name>
 You can specify the `--mib-source` multiple times, e.g. to include the
 NetSNMP definitions that are often located in /usr/share/snmp/mibs.
 
-Note: There are may be errors in the MIB file, like NetGear's
-fastpathswitching.my file has a definition for
-agentKeepalivePortLastLoopDetectedTime that has a default value that is
+Note: There are may be errors in the MIB file.  NetGear's
+`fastpathswitching.my` file has a definition for
+`agentKeepalivePortLastLoopDetectedTime` that has a default value that is
 too short.  It's a 4 byte octet string instead of an 8 byte octet string.
 If you modify the MIB files, you will need to rerun the `mibdump.py`
-command.
+command.  Lenovo's has the same error in `lenovoswitching.my`.
 
 
 Example

test_data.py

@@ -31,4 +31,4 @@ settings = [
 	('annumberset', 42),
 	('nochange', 100),
 ]
-distswitch = vlanmang.SwitchConfig('192.168.0.58', { 'community': 'private' }, distributionswitch, [ 'lag2' ], settings)
+distswitch = vlanmang.SwitchConfig('192.168.0.58', { 'getmanybroken': False, 'community': 'private' }, distributionswitch, [ 'lag2' ], settings)

vlanmang/__init__.py

@@ -51,6 +51,12 @@ __all__ = [
 	'SNMPSwitch',
 ]
 
+if False:
+	from pysnmp import debug
+
+	# use specific flags or 'all' for full debugging
+	debug.setLogger(debug.Debug('dsp', 'msgproc', 'secmod'))
+
 _mbuilder = MibBuilder()
 _mvc = MibViewController(_mbuilder)
 
@@ -290,9 +296,8 @@ def checkchanges(module):
 		switchuntagged = switch.getuntagged(*vlans)
 		untagged = getuntagged(i.vlanconf, lufun)
 		for i in vlans:
-			if not _cmpbits(switchegress[i], egress[i]):
+			if not _cmpbits(switchegress[i], egress[i]) or not _cmpbits(switchuntagged[i], untagged[i]):
 				res.append((switch, name, 'setegress', i, egress[i], switchegress[i]))
-			if not _cmpbits(switchuntagged[i], untagged[i]):
 				res.append((switch, name, 'setuntagged', i, untagged[i], switchuntagged[i]))
 
 	return res
@@ -344,10 +349,14 @@ def getuntagged(data, lookupfun):
 
 	return r
 
+_lenovo_ce0128t = (1, 3, 6, 1, 4, 1, 19046, 1, 7, 43)	# LENOVO-REF-MIB::ce0128t
+
 class SNMPSwitch(object):
 	'''A class for manipulating switches via standard SNMP MIBs.'''
 
-	def __init__(self, host, community=None, username=None, authKey=None, authProtocol=usmHMACSHAAuthProtocol, privKey=None, privProtocol=None):
+	def __init__(self, host, community=None, username=None, authKey=None,
+	    authProtocol=usmHMACSHAAuthProtocol, privKey=None,
+	    privProtocol=None, getmanybroken=None):
 		'''Create a instance to read data and program a switch via
 		SNMP.
 
@@ -399,10 +408,26 @@ class SNMPSwitch(object):
 
 		self._targ = UdpTransportTarget((host, 161), timeout=10)
 
+		self._getmanybroken = False
+
+		if getmanybroken is None:
+			if tuple(self._get(('SNMPv2-MIB',
+			    'sysObjectID', 0))) == _lenovo_ce0128t:
+				self._getmanybroken = True
+		else:
+			self._getmanybroken = bool(getmanybroken)
+
+
 	def __repr__(self):	# pragma: no cover
 		return '<SNMPSwitch: auth=%s, targ=%s>' % (repr(self._auth), repr(self._targ))
 
 	def _getmany(self, *oids):
+		if self._getmanybroken:
+			return [ self._getmany_real(x)[0] for x in oids ]
+		else:
+			return self._getmany_real(*oids)
+
+	def _getmany_real(self, *oids):
 		woids = [ ObjectIdentity(*oid) for oid in oids ]
 		[ oid.resolveWithMib(_mvc) for oid in woids ]
 
@@ -790,7 +815,8 @@ class _TestMisc(unittest.TestCase):
 
 		# That a switch passed a community string
 		commstr = 'foobar'
-		switch = SNMPSwitch(None, community=commstr)
+		switch = SNMPSwitch(None, community=commstr,
+		    getmanybroken=False)
 
 		# That getCmd returns a valid object
 		vb = [ [ None, None ] ]
@@ -823,7 +849,8 @@ class _TestMisc(unittest.TestCase):
 		# That a switch passed v3 auth data
 		username = 'someuser'
 		authKey = 'authKey'
-		switch = SNMPSwitch(None, username=username, authKey=authKey)
+		switch = SNMPSwitch(None, username=username, authKey=authKey,
+		    getmanybroken=False)
 
 		# That getCmd returns a valid object
 		vb = [ [ None, None ] ]
@@ -847,7 +874,7 @@ class _TestMisc(unittest.TestCase):
 		# that it can be called with a privKey
 		privKey = 'privKey'
 		switch = SNMPSwitch(None, username=username, authKey=authKey,
-		    privKey=privKey)
+		    privKey=privKey, getmanybroken=False)
 
 		# and that UsmUserData was called w/ the correct args
 		uud.assert_called_with(username, authKey, privKey,
@@ -859,7 +886,8 @@ class _TestMisc(unittest.TestCase):
 
 		# that it can be called with an alternate privProtocol
 		switch = SNMPSwitch(None, username=username, authKey=authKey,
-		    privKey=privKey, privProtocol=usmDESPrivProtocol)
+		    privKey=privKey, privProtocol=usmDESPrivProtocol,
+		    getmanybroken=False)
 
 		# and that UsmUserData was called w/ the correct args
 		uud.assert_called_with(username, authKey, privKey,
@@ -872,7 +900,7 @@ class _TestMisc(unittest.TestCase):
 		# that it can be called with an alternate authProtocol
 		switch = SNMPSwitch(None, username=username, authKey=authKey,
 		    authProtocol=usmHMACMD5AuthProtocol, privKey=privKey,
-		    privProtocol=usmDESPrivProtocol)
+		    privProtocol=usmDESPrivProtocol, getmanybroken=False)
 
 		# and that UsmUserData was called w/ the correct args
 		uud.assert_called_with(username, authKey, privKey,
@@ -907,6 +935,7 @@ class _TestMisc(unittest.TestCase):
 		gvlans.return_value = iter([ 1, 5 ])
 
 		res = checkchanges('data')
+
 		validres = [ ('createvlan', 283, '', '') ]
 
 		# make sure it needs to get created
@@ -982,6 +1011,7 @@ class _TestMisc(unittest.TestCase):
 		        '1' * 10),
 		      ('setegress', 5, '1' * 8 + '0' * 11 + '11' + '0' * 8 +
 		        '1', '1' * 10),
+		      ('setuntagged', 5, '1' * 8, '1' * 8 + '0' * 10),
 		    ]
 
 		self.assertEqual(set(res), set(validres))
@@ -1008,7 +1038,7 @@ class _TestSNMPSwitch(unittest.TestCase):
 	@mock.patch('vlanmang.getCmd')
 	def test_getmany_nosuchinstance(self, gc, cd):
 		# that a switch
-		switch = SNMPSwitch(None, community=None)
+		switch = SNMPSwitch(None, community=None, getmanybroken=False)
 
 		lookup = { x: chr(x) for x in range(1, 10) }
 
@@ -1021,7 +1051,7 @@ class _TestSNMPSwitch(unittest.TestCase):
 	@mock.patch('vlanmang.getCmd')
 	def test_getmany_nosuchobject(self, gc, cd):
 		# that a switch
-		switch = SNMPSwitch(None, community=None)
+		switch = SNMPSwitch(None, community=None, getmanybroken=False)
 
 		lookup = { x: chr(x) for x in range(1, 10) }
 
@@ -1034,7 +1064,7 @@ class _TestSNMPSwitch(unittest.TestCase):
 	@mock.patch('vlanmang.getCmd')
 	def test_getmany(self, gc, cd):
 		# that a switch
-		switch = SNMPSwitch(None, community=None)
+		switch = SNMPSwitch(None, community=None, getmanybroken=False)
 
 		lookup = { x: chr(x) for x in range(1, 10) }