From 9dc6e969837d9bb68258e07b68c4a33c03e21a91 Mon Sep 17 00:00:00 2001 From: Lex Li Date: Mon, 26 Feb 2024 01:18:44 -0500 Subject: [PATCH] Switched to local test agent. --- pysnmp/entity/config.py | 862 +++++++++++------- tests/agent_context.py | 126 +++ .../test_custom_asn1_mib_search_path.py | 34 +- .../manager/cmdgen/test_usm_none_none.py | 62 +- .../asyncio/manager/cmdgen/test_v1_get.py | 33 +- .../asyncio/manager/cmdgen/test_v1_next.py | 33 +- .../asyncio/manager/cmdgen/test_v1_set.py | 29 +- .../asyncio/manager/cmdgen/test_v2c_bulk.py | 29 +- .../asyncio/manager/cmdgen/test_v2c_get.py | 30 +- .../asyncio/manager/cmdgen/test_v2c_next.py | 33 +- .../asyncio/manager/cmdgen/test_v2c_set.py | 31 +- 11 files changed, 848 insertions(+), 454 deletions(-) create mode 100644 tests/agent_context.py diff --git a/pysnmp/entity/config.py b/pysnmp/entity/config.py index 16828d102..1ca5dd616 100644 --- a/pysnmp/entity/config.py +++ b/pysnmp/entity/config.py @@ -6,6 +6,7 @@ # from pyasn1.compat.octets import null from pysnmp.carrier.asyncio.dgram import udp, udp6, unix +from pysnmp.entity.engine import SnmpEngine from pysnmp.proto.secmod.rfc3414.auth import hmacmd5, hmacsha, noauth from pysnmp.proto.secmod.rfc3414.priv import des, nopriv from pysnmp.proto.secmod.rfc3826.priv import aes @@ -38,8 +39,12 @@ usmDESPrivProtocol = des.Des.serviceID usm3DESEDEPrivProtocol = des3.Des3.serviceID usmAesCfb128Protocol = aes.Aes.serviceID -usmAesBlumenthalCfb192Protocol = aes192.AesBlumenthal192.serviceID # semi-standard but not widely used -usmAesBlumenthalCfb256Protocol = aes256.AesBlumenthal256.serviceID # semi-standard but not widely used +usmAesBlumenthalCfb192Protocol = ( + aes192.AesBlumenthal192.serviceID +) # semi-standard but not widely used +usmAesBlumenthalCfb256Protocol = ( + aes256.AesBlumenthal256.serviceID +) # semi-standard but not widely used usmAesCfb192Protocol = aes192.Aes192.serviceID # non-standard but used by many vendors usmAesCfb256Protocol = aes256.Aes256.serviceID # non-standard but used by many vendors usmNoPrivProtocol = nopriv.NoPriv.serviceID @@ -50,39 +55,60 @@ usmKeyTypeLocalized = 2 # Auth services -authServices = {hmacmd5.HmacMd5.serviceID: hmacmd5.HmacMd5(), - hmacsha.HmacSha.serviceID: hmacsha.HmacSha(), - hmacsha2.HmacSha2.sha224ServiceID: hmacsha2.HmacSha2(hmacsha2.HmacSha2.sha224ServiceID), - hmacsha2.HmacSha2.sha256ServiceID: hmacsha2.HmacSha2(hmacsha2.HmacSha2.sha256ServiceID), - hmacsha2.HmacSha2.sha384ServiceID: hmacsha2.HmacSha2(hmacsha2.HmacSha2.sha384ServiceID), - hmacsha2.HmacSha2.sha512ServiceID: hmacsha2.HmacSha2(hmacsha2.HmacSha2.sha512ServiceID), - noauth.NoAuth.serviceID: noauth.NoAuth()} +authServices = { + hmacmd5.HmacMd5.serviceID: hmacmd5.HmacMd5(), + hmacsha.HmacSha.serviceID: hmacsha.HmacSha(), + hmacsha2.HmacSha2.sha224ServiceID: hmacsha2.HmacSha2( + hmacsha2.HmacSha2.sha224ServiceID + ), + hmacsha2.HmacSha2.sha256ServiceID: hmacsha2.HmacSha2( + hmacsha2.HmacSha2.sha256ServiceID + ), + hmacsha2.HmacSha2.sha384ServiceID: hmacsha2.HmacSha2( + hmacsha2.HmacSha2.sha384ServiceID + ), + hmacsha2.HmacSha2.sha512ServiceID: hmacsha2.HmacSha2( + hmacsha2.HmacSha2.sha512ServiceID + ), + noauth.NoAuth.serviceID: noauth.NoAuth(), +} # Privacy services -privServices = {des.Des.serviceID: des.Des(), - des3.Des3.serviceID: des3.Des3(), - aes.Aes.serviceID: aes.Aes(), - aes192.AesBlumenthal192.serviceID: aes192.AesBlumenthal192(), - aes256.AesBlumenthal256.serviceID: aes256.AesBlumenthal256(), - aes192.Aes192.serviceID: aes192.Aes192(), # non-standard - aes256.Aes256.serviceID: aes256.Aes256(), # non-standard - nopriv.NoPriv.serviceID: nopriv.NoPriv()} +privServices = { + des.Des.serviceID: des.Des(), + des3.Des3.serviceID: des3.Des3(), + aes.Aes.serviceID: aes.Aes(), + aes192.AesBlumenthal192.serviceID: aes192.AesBlumenthal192(), + aes256.AesBlumenthal256.serviceID: aes256.AesBlumenthal256(), + aes192.Aes192.serviceID: aes192.Aes192(), # non-standard + aes256.Aes256.serviceID: aes256.Aes256(), # non-standard + nopriv.NoPriv.serviceID: nopriv.NoPriv(), +} def __cookV1SystemInfo(snmpEngine, communityIndex): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - snmpEngineID, = mibBuilder.importSymbols('__SNMP-FRAMEWORK-MIB', 'snmpEngineID') - snmpCommunityEntry, = mibBuilder.importSymbols('SNMP-COMMUNITY-MIB', 'snmpCommunityEntry') + (snmpEngineID,) = mibBuilder.importSymbols("__SNMP-FRAMEWORK-MIB", "snmpEngineID") + (snmpCommunityEntry,) = mibBuilder.importSymbols( + "SNMP-COMMUNITY-MIB", "snmpCommunityEntry" + ) tblIdx = snmpCommunityEntry.getInstIdFromIndices(communityIndex) return snmpCommunityEntry, tblIdx, snmpEngineID -def addV1System(snmpEngine, communityIndex, communityName, - contextEngineId=None, contextName=None, - transportTag=None, securityName=None): - (snmpCommunityEntry, tblIdx, - snmpEngineID) = __cookV1SystemInfo(snmpEngine, communityIndex) +def addV1System( + snmpEngine: SnmpEngine, + communityIndex: str, + communityName: str, + contextEngineId=None, + contextName=None, + transportTag=None, + securityName=None, +): + (snmpCommunityEntry, tblIdx, snmpEngineID) = __cookV1SystemInfo( + snmpEngine, communityIndex + ) if contextEngineId is None: contextEngineId = snmpEngineID.syntax @@ -95,68 +121,85 @@ def addV1System(snmpEngine, communityIndex, communityName, securityName = securityName is not None and securityName or communityIndex snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpCommunityEntry.name + (8,) + tblIdx, 'destroy'),) + ((snmpCommunityEntry.name + (8,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpCommunityEntry.name + (1,) + tblIdx, communityIndex), - (snmpCommunityEntry.name + (2,) + tblIdx, communityName), - (snmpCommunityEntry.name + (3,) + tblIdx, securityName), - (snmpCommunityEntry.name + (4,) + tblIdx, contextEngineId), - (snmpCommunityEntry.name + (5,) + tblIdx, contextName), - (snmpCommunityEntry.name + (6,) + tblIdx, transportTag), - (snmpCommunityEntry.name + (7,) + tblIdx, 'nonVolatile'), - (snmpCommunityEntry.name + (8,) + tblIdx, 'createAndGo')) + ( + (snmpCommunityEntry.name + (1,) + tblIdx, communityIndex), + (snmpCommunityEntry.name + (2,) + tblIdx, communityName), + (snmpCommunityEntry.name + (3,) + tblIdx, securityName), + (snmpCommunityEntry.name + (4,) + tblIdx, contextEngineId), + (snmpCommunityEntry.name + (5,) + tblIdx, contextName), + (snmpCommunityEntry.name + (6,) + tblIdx, transportTag), + (snmpCommunityEntry.name + (7,) + tblIdx, "nonVolatile"), + (snmpCommunityEntry.name + (8,) + tblIdx, "createAndGo"), + ) ) debug.logger & debug.flagSM and debug.logger( - 'addV1System: added new table entry ' + "addV1System: added new table entry " 'communityIndex "%s" communityName "%s" securityName "%s" ' 'contextEngineId "%s" contextName "%s" transportTag ' - '"%s"' % (communityIndex, communityName, securityName, - contextEngineId, contextName, transportTag)) + '"%s"' + % ( + communityIndex, + communityName, + securityName, + contextEngineId, + contextName, + transportTag, + ) + ) def delV1System(snmpEngine, communityIndex): - (snmpCommunityEntry, tblIdx, - snmpEngineID) = __cookV1SystemInfo(snmpEngine, communityIndex) + (snmpCommunityEntry, tblIdx, snmpEngineID) = __cookV1SystemInfo( + snmpEngine, communityIndex + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpCommunityEntry.name + (8,) + tblIdx, 'destroy'),) + ((snmpCommunityEntry.name + (8,) + tblIdx, "destroy"),) ) debug.logger & debug.flagSM and debug.logger( - 'delV1System: deleted table entry by communityIndex ' - '"%s"' % (communityIndex,)) + "delV1System: deleted table entry by communityIndex " '"%s"' % (communityIndex,) + ) def __cookV3UserInfo(snmpEngine, securityName, securityEngineId): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - snmpEngineID, = mibBuilder.importSymbols('__SNMP-FRAMEWORK-MIB', 'snmpEngineID') + (snmpEngineID,) = mibBuilder.importSymbols("__SNMP-FRAMEWORK-MIB", "snmpEngineID") if securityEngineId is None: securityEngineId = snmpEngineID.syntax else: securityEngineId = snmpEngineID.syntax.clone(securityEngineId) - usmUserEntry, = mibBuilder.importSymbols('SNMP-USER-BASED-SM-MIB', 'usmUserEntry') + (usmUserEntry,) = mibBuilder.importSymbols("SNMP-USER-BASED-SM-MIB", "usmUserEntry") tblIdx1 = usmUserEntry.getInstIdFromIndices(securityEngineId, securityName) - pysnmpUsmSecretEntry, = mibBuilder.importSymbols('PYSNMP-USM-MIB', 'pysnmpUsmSecretEntry') + (pysnmpUsmSecretEntry,) = mibBuilder.importSymbols( + "PYSNMP-USM-MIB", "pysnmpUsmSecretEntry" + ) tblIdx2 = pysnmpUsmSecretEntry.getInstIdFromIndices(securityName) return securityEngineId, usmUserEntry, tblIdx1, pysnmpUsmSecretEntry, tblIdx2 -def addV3User(snmpEngine, userName, - authProtocol=usmNoAuthProtocol, authKey=None, - privProtocol=usmNoPrivProtocol, privKey=None, - securityEngineId=None, - securityName=None, - authKeyType=usmKeyTypePassphrase, - privKeyType=usmKeyTypePassphrase, - # deprecated parameter - contextEngineId=None): - +def addV3User( + snmpEngine, + userName, + authProtocol=usmNoAuthProtocol, + authKey=None, + privProtocol=usmNoPrivProtocol, + privKey=None, + securityEngineId=None, + securityName=None, + authKeyType=usmKeyTypePassphrase, + privKeyType=usmKeyTypePassphrase, + # deprecated parameter + contextEngineId=None, +): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder if securityName is None: @@ -165,35 +208,45 @@ def addV3User(snmpEngine, userName, if securityEngineId is None: # backward compatibility securityEngineId = contextEngineId - (securityEngineId, usmUserEntry, tblIdx1, - pysnmpUsmSecretEntry, tblIdx2) = __cookV3UserInfo( - snmpEngine, securityName, securityEngineId) + ( + securityEngineId, + usmUserEntry, + tblIdx1, + pysnmpUsmSecretEntry, + tblIdx2, + ) = __cookV3UserInfo(snmpEngine, securityName, securityEngineId) # Load augmenting table before creating new row in base one - pysnmpUsmKeyEntry, = mibBuilder.importSymbols('PYSNMP-USM-MIB', 'pysnmpUsmKeyEntry') + (pysnmpUsmKeyEntry,) = mibBuilder.importSymbols( + "PYSNMP-USM-MIB", "pysnmpUsmKeyEntry" + ) # Load clone-from (may not be needed) - zeroDotZero, = mibBuilder.importSymbols('SNMPv2-SMI', 'zeroDotZero') + (zeroDotZero,) = mibBuilder.importSymbols("SNMPv2-SMI", "zeroDotZero") snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((usmUserEntry.name + (13,) + tblIdx1, 'destroy'),) + ((usmUserEntry.name + (13,) + tblIdx1, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((usmUserEntry.name + (2,) + tblIdx1, userName), - (usmUserEntry.name + (3,) + tblIdx1, securityName), - (usmUserEntry.name + (4,) + tblIdx1, zeroDotZero.name), - (usmUserEntry.name + (5,) + tblIdx1, authProtocol), - (usmUserEntry.name + (8,) + tblIdx1, privProtocol), - (usmUserEntry.name + (13,) + tblIdx1, 'createAndGo')) + ( + (usmUserEntry.name + (2,) + tblIdx1, userName), + (usmUserEntry.name + (3,) + tblIdx1, securityName), + (usmUserEntry.name + (4,) + tblIdx1, zeroDotZero.name), + (usmUserEntry.name + (5,) + tblIdx1, authProtocol), + (usmUserEntry.name + (8,) + tblIdx1, privProtocol), + (usmUserEntry.name + (13,) + tblIdx1, "createAndGo"), + ) ) if authProtocol not in authServices: - raise error.PySnmpError(f'Unknown auth protocol {authProtocol}') + raise error.PySnmpError(f"Unknown auth protocol {authProtocol}") if privProtocol not in privServices: - raise error.PySnmpError(f'Unknown privacy protocol {privProtocol}') + raise error.PySnmpError(f"Unknown privacy protocol {privProtocol}") - pysnmpUsmKeyType, = mibBuilder.importSymbols('__PYSNMP-USM-MIB', 'pysnmpUsmKeyType') + (pysnmpUsmKeyType,) = mibBuilder.importSymbols( + "__PYSNMP-USM-MIB", "pysnmpUsmKeyType" + ) authKeyType = pysnmpUsmKeyType.syntax.clone(authKeyType) @@ -204,9 +257,7 @@ def addV3User(snmpEngine, userName, masterAuthKey = localAuthKey = authKey if authKeyType < usmKeyTypeMaster: # pass phrase is given - masterAuthKey = authServices[authProtocol].hashPassphrase( - authKey or null - ) + masterAuthKey = authServices[authProtocol].hashPassphrase(authKey or null) if authKeyType < usmKeyTypeLocalized: # pass phrase or master key is given localAuthKey = authServices[authProtocol].localizeKey( @@ -252,35 +303,43 @@ def addV3User(snmpEngine, userName, ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((pysnmpUsmSecretEntry.name + (4,) + tblIdx2, 'destroy'),) + ((pysnmpUsmSecretEntry.name + (4,) + tblIdx2, "destroy"),) ) # Commit plain-text pass-phrases if we have them snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((pysnmpUsmSecretEntry.name + (4,) + tblIdx2, 'createAndGo'),) + ((pysnmpUsmSecretEntry.name + (4,) + tblIdx2, "createAndGo"),) ) if authKeyType < usmKeyTypeMaster: snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((pysnmpUsmSecretEntry.name + (1,) + tblIdx2, userName), - (pysnmpUsmSecretEntry.name + (2,) + tblIdx2, authKey)) + ( + (pysnmpUsmSecretEntry.name + (1,) + tblIdx2, userName), + (pysnmpUsmSecretEntry.name + (2,) + tblIdx2, authKey), + ) ) if privKeyType < usmKeyTypeMaster: snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((pysnmpUsmSecretEntry.name + (1,) + tblIdx2, userName), - (pysnmpUsmSecretEntry.name + (3,) + tblIdx2, privKey)) + ( + (pysnmpUsmSecretEntry.name + (1,) + tblIdx2, userName), + (pysnmpUsmSecretEntry.name + (3,) + tblIdx2, privKey), + ) ) debug.logger & debug.flagSM and debug.logger( - 'addV3User: added new table entries ' + "addV3User: added new table entries " 'userName "%s" securityName "%s" authProtocol %s ' 'privProtocol %s localAuthKey "%s" localPrivKey "%s" ' 'masterAuthKey "%s" masterPrivKey "%s" authKey "%s" ' 'privKey "%s" by index securityName "%s" securityEngineId ' - '"%s"' % ( - userName, securityName, authProtocol, privProtocol, + '"%s"' + % ( + userName, + securityName, + authProtocol, + privProtocol, localAuthKey and localAuthKey.prettyPrint(), localPrivKey and localPrivKey.prettyPrint(), masterAuthKey and masterAuthKey.prettyPrint(), @@ -288,48 +347,54 @@ def addV3User(snmpEngine, userName, authKey and authKey.prettyPrint(), privKey and privKey.prettyPrint(), securityName, - securityEngineId.prettyPrint())) + securityEngineId.prettyPrint(), + ) + ) -def delV3User(snmpEngine, - userName, - securityEngineId=None, - # deprecated parameters follow - contextEngineId=None): +def delV3User( + snmpEngine, + userName, + securityEngineId=None, + # deprecated parameters follow + contextEngineId=None, +): if securityEngineId is None: # backward compatibility securityEngineId = contextEngineId - (securityEngineId, usmUserEntry, tblIdx1, pysnmpUsmSecretEntry, - tblIdx2) = __cookV3UserInfo(snmpEngine, userName, securityEngineId) + ( + securityEngineId, + usmUserEntry, + tblIdx1, + pysnmpUsmSecretEntry, + tblIdx2, + ) = __cookV3UserInfo(snmpEngine, userName, securityEngineId) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((usmUserEntry.name + (13,) + tblIdx1, 'destroy'),) + ((usmUserEntry.name + (13,) + tblIdx1, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((pysnmpUsmSecretEntry.name + (4,) + tblIdx2, 'destroy'),) + ((pysnmpUsmSecretEntry.name + (4,) + tblIdx2, "destroy"),) ) debug.logger & debug.flagSM and debug.logger( - 'delV3User: deleted table entries by index ' + "delV3User: deleted table entries by index " 'userName "%s" securityEngineId ' - '"%s"' % ( - userName, - securityEngineId.prettyPrint())) + '"%s"' % (userName, securityEngineId.prettyPrint()) + ) # Drop all derived rows varBinds = initialVarBinds = ( (usmUserEntry.name + (1,), None), # usmUserEngineID (usmUserEntry.name + (2,), None), # usmUserName - (usmUserEntry.name + (4,), None) # usmUserCloneFrom + (usmUserEntry.name + (4,), None), # usmUserCloneFrom ) while varBinds: - varBinds = snmpEngine.msgAndPduDsp.mibInstrumController.readNextVars( - varBinds - ) + varBinds = snmpEngine.msgAndPduDsp.mibInstrumController.readNextVars(varBinds) if varBinds[0][1].isSameTypeWith(rfc1905.endOfMibView): break - if varBinds[0][0][:len(initialVarBinds[0][0])] != initialVarBinds[0][0]: + if varBinds[0][0][: len(initialVarBinds[0][0])] != initialVarBinds[0][0]: break elif varBinds[2][1] == tblIdx1: # cloned from this entry delV3User(snmpEngine, varBinds[1][1], varBinds[0][1]) @@ -339,7 +404,9 @@ def delV3User(snmpEngine, def __cookTargetParamsInfo(snmpEngine, name): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - snmpTargetParamsEntry, = mibBuilder.importSymbols('SNMP-TARGET-MIB', 'snmpTargetParamsEntry') + (snmpTargetParamsEntry,) = mibBuilder.importSymbols( + "SNMP-TARGET-MIB", "snmpTargetParamsEntry" + ) tblIdx = snmpTargetParamsEntry.getInstIdFromIndices(name) return snmpTargetParamsEntry, tblIdx @@ -353,81 +420,101 @@ def addTargetParams(snmpEngine, name, securityName, securityLevel, mpModel=3): elif mpModel == 3: securityModel = 3 else: - raise error.PySnmpError('Unknown MP model %s' % mpModel) + raise error.PySnmpError("Unknown MP model %s" % mpModel) snmpTargetParamsEntry, tblIdx = __cookTargetParamsInfo(snmpEngine, name) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpTargetParamsEntry.name + (7,) + tblIdx, 'destroy'),) + ((snmpTargetParamsEntry.name + (7,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpTargetParamsEntry.name + (1,) + tblIdx, name), - (snmpTargetParamsEntry.name + (2,) + tblIdx, mpModel), - (snmpTargetParamsEntry.name + (3,) + tblIdx, securityModel), - (snmpTargetParamsEntry.name + (4,) + tblIdx, securityName), - (snmpTargetParamsEntry.name + (5,) + tblIdx, securityLevel), - (snmpTargetParamsEntry.name + (7,) + tblIdx, 'createAndGo')) + ( + (snmpTargetParamsEntry.name + (1,) + tblIdx, name), + (snmpTargetParamsEntry.name + (2,) + tblIdx, mpModel), + (snmpTargetParamsEntry.name + (3,) + tblIdx, securityModel), + (snmpTargetParamsEntry.name + (4,) + tblIdx, securityName), + (snmpTargetParamsEntry.name + (5,) + tblIdx, securityLevel), + (snmpTargetParamsEntry.name + (7,) + tblIdx, "createAndGo"), + ) ) def delTargetParams(snmpEngine, name): snmpTargetParamsEntry, tblIdx = __cookTargetParamsInfo(snmpEngine, name) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpTargetParamsEntry.name + (7,) + tblIdx, 'destroy'),) + ((snmpTargetParamsEntry.name + (7,) + tblIdx, "destroy"),) ) def __cookTargetAddrInfo(snmpEngine, addrName): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - snmpTargetAddrEntry, = mibBuilder.importSymbols('SNMP-TARGET-MIB', 'snmpTargetAddrEntry') - snmpSourceAddrEntry, = mibBuilder.importSymbols('PYSNMP-SOURCE-MIB', 'snmpSourceAddrEntry') + (snmpTargetAddrEntry,) = mibBuilder.importSymbols( + "SNMP-TARGET-MIB", "snmpTargetAddrEntry" + ) + (snmpSourceAddrEntry,) = mibBuilder.importSymbols( + "PYSNMP-SOURCE-MIB", "snmpSourceAddrEntry" + ) tblIdx = snmpTargetAddrEntry.getInstIdFromIndices(addrName) return snmpTargetAddrEntry, snmpSourceAddrEntry, tblIdx -def addTargetAddr(snmpEngine, addrName, transportDomain, transportAddress, - params, timeout=None, retryCount=None, tagList=null, - sourceAddress=None): +def addTargetAddr( + snmpEngine, + addrName, + transportDomain, + transportAddress, + params, + timeout=None, + retryCount=None, + tagList=null, + sourceAddress=None, +): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - (snmpTargetAddrEntry, snmpSourceAddrEntry, - tblIdx) = __cookTargetAddrInfo(snmpEngine, addrName) + (snmpTargetAddrEntry, snmpSourceAddrEntry, tblIdx) = __cookTargetAddrInfo( + snmpEngine, addrName + ) - if transportDomain[:len(snmpUDPDomain)] == snmpUDPDomain: - SnmpUDPAddress, = mibBuilder.importSymbols('SNMPv2-TM', 'SnmpUDPAddress') + if transportDomain[: len(snmpUDPDomain)] == snmpUDPDomain: + (SnmpUDPAddress,) = mibBuilder.importSymbols("SNMPv2-TM", "SnmpUDPAddress") transportAddress = SnmpUDPAddress(transportAddress) if sourceAddress is None: - sourceAddress = ('0.0.0.0', 0) + sourceAddress = ("0.0.0.0", 0) sourceAddress = SnmpUDPAddress(sourceAddress) - elif transportDomain[:len(snmpUDP6Domain)] == snmpUDP6Domain: - TransportAddressIPv6, = mibBuilder.importSymbols('TRANSPORT-ADDRESS-MIB', 'TransportAddressIPv6') + elif transportDomain[: len(snmpUDP6Domain)] == snmpUDP6Domain: + (TransportAddressIPv6,) = mibBuilder.importSymbols( + "TRANSPORT-ADDRESS-MIB", "TransportAddressIPv6" + ) transportAddress = TransportAddressIPv6(transportAddress) if sourceAddress is None: - sourceAddress = ('::', 0) + sourceAddress = ("::", 0) sourceAddress = TransportAddressIPv6(sourceAddress) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpTargetAddrEntry.name + (9,) + tblIdx, 'destroy'),) + ((snmpTargetAddrEntry.name + (9,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpTargetAddrEntry.name + (1,) + tblIdx, addrName), - (snmpTargetAddrEntry.name + (2,) + tblIdx, transportDomain), - (snmpTargetAddrEntry.name + (3,) + tblIdx, transportAddress), - (snmpTargetAddrEntry.name + (4,) + tblIdx, timeout), - (snmpTargetAddrEntry.name + (5,) + tblIdx, retryCount), - (snmpTargetAddrEntry.name + (6,) + tblIdx, tagList), - (snmpTargetAddrEntry.name + (7,) + tblIdx, params), - (snmpSourceAddrEntry.name + (1,) + tblIdx, sourceAddress), - (snmpTargetAddrEntry.name + (9,) + tblIdx, 'createAndGo')) + ( + (snmpTargetAddrEntry.name + (1,) + tblIdx, addrName), + (snmpTargetAddrEntry.name + (2,) + tblIdx, transportDomain), + (snmpTargetAddrEntry.name + (3,) + tblIdx, transportAddress), + (snmpTargetAddrEntry.name + (4,) + tblIdx, timeout), + (snmpTargetAddrEntry.name + (5,) + tblIdx, retryCount), + (snmpTargetAddrEntry.name + (6,) + tblIdx, tagList), + (snmpTargetAddrEntry.name + (7,) + tblIdx, params), + (snmpSourceAddrEntry.name + (1,) + tblIdx, sourceAddress), + (snmpTargetAddrEntry.name + (9,) + tblIdx, "createAndGo"), + ) ) def delTargetAddr(snmpEngine, addrName): - (snmpTargetAddrEntry, snmpSourceAddrEntry, - tblIdx) = __cookTargetAddrInfo(snmpEngine, addrName) + (snmpTargetAddrEntry, snmpSourceAddrEntry, tblIdx) = __cookTargetAddrInfo( + snmpEngine, addrName + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpTargetAddrEntry.name + (9,) + tblIdx, 'destroy'),) + ((snmpTargetAddrEntry.name + (9,) + tblIdx, "destroy"),) ) @@ -435,17 +522,16 @@ def addTransport(snmpEngine, transportDomain, transport): if snmpEngine.transportDispatcher: if not transport.isCompatibleWithDispatcher(snmpEngine.transportDispatcher): raise error.PySnmpError( - f'Transport {transport!r} is not compatible with dispatcher {snmpEngine.transportDispatcher!r}') + f"Transport {transport!r} is not compatible with dispatcher {snmpEngine.transportDispatcher!r}" + ) else: - snmpEngine.registerTransportDispatcher( - transport.protoTransportDispatcher() - ) + snmpEngine.registerTransportDispatcher(transport.protoTransportDispatcher()) # here we note that we have created transportDispatcher automatically snmpEngine.setUserContext(automaticTransportDispatcher=0) snmpEngine.transportDispatcher.registerTransport(transportDomain, transport) automaticTransportDispatcher = snmpEngine.getUserContext( - 'automaticTransportDispatcher' + "automaticTransportDispatcher" ) if automaticTransportDispatcher is not None: snmpEngine.setUserContext( @@ -469,7 +555,7 @@ def delTransport(snmpEngine, transportDomain): snmpEngine.transportDispatcher.unregisterTransport(transportDomain) # automatically shutdown automatically created transportDispatcher automaticTransportDispatcher = snmpEngine.getUserContext( - 'automaticTransportDispatcher' + "automaticTransportDispatcher" ) if automaticTransportDispatcher is not None: automaticTransportDispatcher -= 1 @@ -489,9 +575,12 @@ def delTransport(snmpEngine, transportDomain): # VACM shortcuts + def __cookVacmContextInfo(snmpEngine, contextName): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - vacmContextEntry, = mibBuilder.importSymbols('SNMP-VIEW-BASED-ACM-MIB', 'vacmContextEntry') + (vacmContextEntry,) = mibBuilder.importSymbols( + "SNMP-VIEW-BASED-ACM-MIB", "vacmContextEntry" + ) tblIdx = vacmContextEntry.getInstIdFromIndices(contextName) return vacmContextEntry, tblIdx @@ -500,11 +589,13 @@ def addContext(snmpEngine, contextName): vacmContextEntry, tblIdx = __cookVacmContextInfo(snmpEngine, contextName) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmContextEntry.name + (2,) + tblIdx, 'destroy'),) + ((vacmContextEntry.name + (2,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmContextEntry.name + (1,) + tblIdx, contextName), - (vacmContextEntry.name + (2,) + tblIdx, 'createAndGo')) + ( + (vacmContextEntry.name + (1,) + tblIdx, contextName), + (vacmContextEntry.name + (2,) + tblIdx, "createAndGo"), + ) ) @@ -512,31 +603,34 @@ def delContext(snmpEngine, contextName): vacmContextEntry, tblIdx = __cookVacmContextInfo(snmpEngine, contextName) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmContextEntry.name + (2,) + tblIdx, 'destroy'),) + ((vacmContextEntry.name + (2,) + tblIdx, "destroy"),) ) def __cookVacmGroupInfo(snmpEngine, securityModel, securityName): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - vacmSecurityToGroupEntry, = mibBuilder.importSymbols('SNMP-VIEW-BASED-ACM-MIB', - 'vacmSecurityToGroupEntry') - tblIdx = vacmSecurityToGroupEntry.getInstIdFromIndices(securityModel, - securityName) + (vacmSecurityToGroupEntry,) = mibBuilder.importSymbols( + "SNMP-VIEW-BASED-ACM-MIB", "vacmSecurityToGroupEntry" + ) + tblIdx = vacmSecurityToGroupEntry.getInstIdFromIndices(securityModel, securityName) return vacmSecurityToGroupEntry, tblIdx def addVacmGroup(snmpEngine, groupName, securityModel, securityName): - (vacmSecurityToGroupEntry, - tblIdx) = __cookVacmGroupInfo(snmpEngine, securityModel, securityName) + (vacmSecurityToGroupEntry, tblIdx) = __cookVacmGroupInfo( + snmpEngine, securityModel, securityName + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmSecurityToGroupEntry.name + (5,) + tblIdx, 'destroy'),) + ((vacmSecurityToGroupEntry.name + (5,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmSecurityToGroupEntry.name + (1,) + tblIdx, securityModel), - (vacmSecurityToGroupEntry.name + (2,) + tblIdx, securityName), - (vacmSecurityToGroupEntry.name + (3,) + tblIdx, groupName), - (vacmSecurityToGroupEntry.name + (5,) + tblIdx, 'createAndGo')) + ( + (vacmSecurityToGroupEntry.name + (1,) + tblIdx, securityModel), + (vacmSecurityToGroupEntry.name + (2,) + tblIdx, securityName), + (vacmSecurityToGroupEntry.name + (3,) + tblIdx, groupName), + (vacmSecurityToGroupEntry.name + (5,) + tblIdx, "createAndGo"), + ) ) @@ -545,67 +639,81 @@ def delVacmGroup(snmpEngine, securityModel, securityName): snmpEngine, securityModel, securityName ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmSecurityToGroupEntry.name + (5,) + tblIdx, 'destroy'),) + ((vacmSecurityToGroupEntry.name + (5,) + tblIdx, "destroy"),) ) -def __cookVacmAccessInfo(snmpEngine, groupName, contextName, securityModel, - securityLevel): +def __cookVacmAccessInfo( + snmpEngine, groupName, contextName, securityModel, securityLevel +): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - vacmAccessEntry, = mibBuilder.importSymbols('SNMP-VIEW-BASED-ACM-MIB', 'vacmAccessEntry') - tblIdx = vacmAccessEntry.getInstIdFromIndices(groupName, contextName, - securityModel, securityLevel) + (vacmAccessEntry,) = mibBuilder.importSymbols( + "SNMP-VIEW-BASED-ACM-MIB", "vacmAccessEntry" + ) + tblIdx = vacmAccessEntry.getInstIdFromIndices( + groupName, contextName, securityModel, securityLevel + ) return vacmAccessEntry, tblIdx -def addVacmAccess(snmpEngine, groupName, contextPrefix, securityModel, - securityLevel, contextMatch, readView, writeView, notifyView): +def addVacmAccess( + snmpEngine, + groupName, + contextPrefix, + securityModel, + securityLevel, + contextMatch, + readView, + writeView, + notifyView, +): vacmAccessEntry, tblIdx = __cookVacmAccessInfo( - snmpEngine, groupName, contextPrefix, securityModel, - securityLevel) + snmpEngine, groupName, contextPrefix, securityModel, securityLevel + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmAccessEntry.name + (9,) + tblIdx, 'destroy'),) + ((vacmAccessEntry.name + (9,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmAccessEntry.name + (1,) + tblIdx, contextPrefix), - (vacmAccessEntry.name + (2,) + tblIdx, securityModel), - (vacmAccessEntry.name + (3,) + tblIdx, securityLevel), - (vacmAccessEntry.name + (4,) + tblIdx, contextMatch), - (vacmAccessEntry.name + (5,) + tblIdx, readView), - (vacmAccessEntry.name + (6,) + tblIdx, writeView), - (vacmAccessEntry.name + (7,) + tblIdx, notifyView), - (vacmAccessEntry.name + (9,) + tblIdx, 'createAndGo')) + ( + (vacmAccessEntry.name + (1,) + tblIdx, contextPrefix), + (vacmAccessEntry.name + (2,) + tblIdx, securityModel), + (vacmAccessEntry.name + (3,) + tblIdx, securityLevel), + (vacmAccessEntry.name + (4,) + tblIdx, contextMatch), + (vacmAccessEntry.name + (5,) + tblIdx, readView), + (vacmAccessEntry.name + (6,) + tblIdx, writeView), + (vacmAccessEntry.name + (7,) + tblIdx, notifyView), + (vacmAccessEntry.name + (9,) + tblIdx, "createAndGo"), + ) ) -def delVacmAccess(snmpEngine, groupName, contextPrefix, securityModel, - securityLevel): +def delVacmAccess(snmpEngine, groupName, contextPrefix, securityModel, securityLevel): vacmAccessEntry, tblIdx = __cookVacmAccessInfo( - snmpEngine, groupName, contextPrefix, securityModel, securityLevel) + snmpEngine, groupName, contextPrefix, securityModel, securityLevel + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmAccessEntry.name + (9,) + tblIdx, 'destroy'),) + ((vacmAccessEntry.name + (9,) + tblIdx, "destroy"),) ) def __cookVacmViewInfo(snmpEngine, viewName, subTree): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - vacmViewTreeFamilyEntry, = mibBuilder.importSymbols( - 'SNMP-VIEW-BASED-ACM-MIB', 'vacmViewTreeFamilyEntry' + (vacmViewTreeFamilyEntry,) = mibBuilder.importSymbols( + "SNMP-VIEW-BASED-ACM-MIB", "vacmViewTreeFamilyEntry" ) tblIdx = vacmViewTreeFamilyEntry.getInstIdFromIndices(viewName, subTree) return vacmViewTreeFamilyEntry, tblIdx def addVacmView(snmpEngine, viewName, viewType, subTree, subTreeMask): - vacmViewTreeFamilyEntry, tblIdx = __cookVacmViewInfo( - snmpEngine, viewName, subTree) + vacmViewTreeFamilyEntry, tblIdx = __cookVacmViewInfo(snmpEngine, viewName, subTree) # Allow bitmask specification in form of an OID - if rfc1902.OctetString('.').asOctets() in rfc1902.OctetString(subTreeMask): + if rfc1902.OctetString(".").asOctets() in rfc1902.OctetString(subTreeMask): subTreeMask = rfc1902.ObjectIdentifier(subTreeMask) if isinstance(subTreeMask, rfc1902.ObjectIdentifier): @@ -614,207 +722,314 @@ def addVacmView(snmpEngine, viewName, viewType, subTree, subTreeMask): subTreeMask += (1,) * (len(subTree) - len(subTreeMask)) subTreeMask = rfc1902.OctetString.fromBinaryString( - ''.join(str(x) for x in subTreeMask)) + "".join(str(x) for x in subTreeMask) + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmViewTreeFamilyEntry.name + (6,) + tblIdx, 'destroy'),) + ((vacmViewTreeFamilyEntry.name + (6,) + tblIdx, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmViewTreeFamilyEntry.name + (1,) + tblIdx, viewName), - (vacmViewTreeFamilyEntry.name + (2,) + tblIdx, subTree), - (vacmViewTreeFamilyEntry.name + (3,) + tblIdx, subTreeMask), - (vacmViewTreeFamilyEntry.name + (4,) + tblIdx, viewType), - (vacmViewTreeFamilyEntry.name + (6,) + tblIdx, 'createAndGo')) + ( + (vacmViewTreeFamilyEntry.name + (1,) + tblIdx, viewName), + (vacmViewTreeFamilyEntry.name + (2,) + tblIdx, subTree), + (vacmViewTreeFamilyEntry.name + (3,) + tblIdx, subTreeMask), + (vacmViewTreeFamilyEntry.name + (4,) + tblIdx, viewType), + (vacmViewTreeFamilyEntry.name + (6,) + tblIdx, "createAndGo"), + ) ) def delVacmView(snmpEngine, viewName, subTree): - vacmViewTreeFamilyEntry, tblIdx = __cookVacmViewInfo(snmpEngine, viewName, - subTree) + vacmViewTreeFamilyEntry, tblIdx = __cookVacmViewInfo(snmpEngine, viewName, subTree) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((vacmViewTreeFamilyEntry.name + (6,) + tblIdx, 'destroy'),) + ((vacmViewTreeFamilyEntry.name + (6,) + tblIdx, "destroy"),) ) # VACM simplicity wrappers + def __cookVacmUserInfo(snmpEngine, securityModel, securityName, securityLevel): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - groupName = 'v-%s-%d' % (hash(securityName), securityModel) - SnmpSecurityLevel, = mibBuilder.importSymbols('SNMP-FRAMEWORK-MIB', 'SnmpSecurityLevel') + groupName = "v-%s-%d" % (hash(securityName), securityModel) + (SnmpSecurityLevel,) = mibBuilder.importSymbols( + "SNMP-FRAMEWORK-MIB", "SnmpSecurityLevel" + ) securityLevel = SnmpSecurityLevel(securityLevel) - return (groupName, securityLevel, - 'r' + groupName, 'w' + groupName, 'n' + groupName) - - -def addVacmUser(snmpEngine, securityModel, securityName, securityLevel, - readSubTree=(), writeSubTree=(), notifySubTree=(), - contextName=null): - (groupName, securityLevel, readView, writeView, - notifyView) = __cookVacmUserInfo(snmpEngine, securityModel, securityName, - securityLevel) + return (groupName, securityLevel, "r" + groupName, "w" + groupName, "n" + groupName) + + +def addVacmUser( + snmpEngine: SnmpEngine, + securityModel: int, + securityName: str, + securityLevel: str, + readSubTree=(), + writeSubTree=(), + notifySubTree=(), + contextName=null, +): + (groupName, securityLevel, readView, writeView, notifyView) = __cookVacmUserInfo( + snmpEngine, securityModel, securityName, securityLevel + ) addContext(snmpEngine, contextName) addVacmGroup(snmpEngine, groupName, securityModel, securityName) - addVacmAccess(snmpEngine, groupName, contextName, securityModel, - securityLevel, 'exact', readView, writeView, notifyView) + addVacmAccess( + snmpEngine, + groupName, + contextName, + securityModel, + securityLevel, + "exact", + readView, + writeView, + notifyView, + ) if readSubTree: - addVacmView(snmpEngine, readView, 'included', readSubTree, null) + addVacmView(snmpEngine, readView, "included", readSubTree, null) if writeSubTree: - addVacmView(snmpEngine, writeView, 'included', writeSubTree, null) + addVacmView(snmpEngine, writeView, "included", writeSubTree, null) if notifySubTree: - addVacmView(snmpEngine, notifyView, 'included', notifySubTree, null) - - -def delVacmUser(snmpEngine, securityModel, securityName, securityLevel, - readSubTree=(), writeSubTree=(), notifySubTree=(), - contextName=null): - (groupName, securityLevel, readView, writeView, - notifyView) = __cookVacmUserInfo(snmpEngine, securityModel, - securityName, securityLevel) + addVacmView(snmpEngine, notifyView, "included", notifySubTree, null) + + +def delVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + readSubTree=(), + writeSubTree=(), + notifySubTree=(), + contextName=null, +): + (groupName, securityLevel, readView, writeView, notifyView) = __cookVacmUserInfo( + snmpEngine, securityModel, securityName, securityLevel + ) delContext(snmpEngine, contextName) delVacmGroup(snmpEngine, securityModel, securityName) delVacmAccess(snmpEngine, groupName, contextName, securityModel, securityLevel) if readSubTree: - delVacmView( - snmpEngine, readView, readSubTree - ) + delVacmView(snmpEngine, readView, readSubTree) if writeSubTree: - delVacmView( - snmpEngine, writeView, writeSubTree - ) + delVacmView(snmpEngine, writeView, writeSubTree) if notifySubTree: - delVacmView( - snmpEngine, notifyView, notifySubTree - ) + delVacmView(snmpEngine, notifyView, notifySubTree) # Obsolete shortcuts for add/delVacmUser() wrappers -def addRoUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, contextName=null): - addVacmUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, contextName=contextName) + +def addRoUser( + snmpEngine, securityModel, securityName, securityLevel, subTree, contextName=null +): + addVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + subTree, + contextName=contextName, + ) -def delRoUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, contextName=null): - delVacmUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, contextName=contextName) +def delRoUser( + snmpEngine, securityModel, securityName, securityLevel, subTree, contextName=null +): + delVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + subTree, + contextName=contextName, + ) -def addRwUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, contextName=null): - addVacmUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, subTree, contextName=contextName) +def addRwUser( + snmpEngine, securityModel, securityName, securityLevel, subTree, contextName=null +): + addVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + subTree, + subTree, + contextName=contextName, + ) -def delRwUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, contextName=null): - delVacmUser(snmpEngine, securityModel, securityName, securityLevel, - subTree, subTree, contextName=contextName) +def delRwUser( + snmpEngine, securityModel, securityName, securityLevel, subTree, contextName=null +): + delVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + subTree, + subTree, + contextName=contextName, + ) -def addTrapUser(snmpEngine, securityModel, securityName, - securityLevel, subTree, contextName=null): - addVacmUser(snmpEngine, securityModel, securityName, securityLevel, - (), (), subTree, contextName=contextName) +def addTrapUser( + snmpEngine, securityModel, securityName, securityLevel, subTree, contextName=null +): + addVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + (), + (), + subTree, + contextName=contextName, + ) -def delTrapUser(snmpEngine, securityModel, securityName, - securityLevel, subTree, contextName=null): - delVacmUser(snmpEngine, securityModel, securityName, securityLevel, - (), (), subTree, contextName=contextName) +def delTrapUser( + snmpEngine, securityModel, securityName, securityLevel, subTree, contextName=null +): + delVacmUser( + snmpEngine, + securityModel, + securityName, + securityLevel, + (), + (), + subTree, + contextName=contextName, + ) # Notification target setup -def __cookNotificationTargetInfo(snmpEngine, notificationName, paramsName, - filterSubtree=None): + +def __cookNotificationTargetInfo( + snmpEngine, notificationName, paramsName, filterSubtree=None +): mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder - snmpNotifyEntry, = mibBuilder.importSymbols('SNMP-NOTIFICATION-MIB', 'snmpNotifyEntry') + (snmpNotifyEntry,) = mibBuilder.importSymbols( + "SNMP-NOTIFICATION-MIB", "snmpNotifyEntry" + ) tblIdx1 = snmpNotifyEntry.getInstIdFromIndices(notificationName) - snmpNotifyFilterProfileEntry, = mibBuilder.importSymbols('SNMP-NOTIFICATION-MIB', - 'snmpNotifyFilterProfileEntry') + (snmpNotifyFilterProfileEntry,) = mibBuilder.importSymbols( + "SNMP-NOTIFICATION-MIB", "snmpNotifyFilterProfileEntry" + ) tblIdx2 = snmpNotifyFilterProfileEntry.getInstIdFromIndices(paramsName) - profileName = '%s-filter' % hash(notificationName) + profileName = "%s-filter" % hash(notificationName) if filterSubtree: - snmpNotifyFilterEntry, = mibBuilder.importSymbols('SNMP-NOTIFICATION-MIB', - 'snmpNotifyFilterEntry') - tblIdx3 = snmpNotifyFilterEntry.getInstIdFromIndices(profileName, - filterSubtree) + (snmpNotifyFilterEntry,) = mibBuilder.importSymbols( + "SNMP-NOTIFICATION-MIB", "snmpNotifyFilterEntry" + ) + tblIdx3 = snmpNotifyFilterEntry.getInstIdFromIndices(profileName, filterSubtree) else: snmpNotifyFilterEntry = tblIdx3 = None - return (snmpNotifyEntry, tblIdx1, - snmpNotifyFilterProfileEntry, tblIdx2, profileName, - snmpNotifyFilterEntry, tblIdx3) + return ( + snmpNotifyEntry, + tblIdx1, + snmpNotifyFilterProfileEntry, + tblIdx2, + profileName, + snmpNotifyFilterEntry, + tblIdx3, + ) -def addNotificationTarget(snmpEngine, notificationName, paramsName, - transportTag, notifyType=None, filterSubtree=None, - filterMask=None, filterType=None): - (snmpNotifyEntry, tblIdx1, snmpNotifyFilterProfileEntry, tblIdx2, - profileName, snmpNotifyFilterEntry, - tblIdx3) = __cookNotificationTargetInfo(snmpEngine, notificationName, - paramsName, filterSubtree) +def addNotificationTarget( + snmpEngine, + notificationName, + paramsName, + transportTag, + notifyType=None, + filterSubtree=None, + filterMask=None, + filterType=None, +): + ( + snmpNotifyEntry, + tblIdx1, + snmpNotifyFilterProfileEntry, + tblIdx2, + profileName, + snmpNotifyFilterEntry, + tblIdx3, + ) = __cookNotificationTargetInfo( + snmpEngine, notificationName, paramsName, filterSubtree + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyEntry.name + (5,) + tblIdx1, 'destroy'),) + ((snmpNotifyEntry.name + (5,) + tblIdx1, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyEntry.name + (2,) + tblIdx1, transportTag), - (snmpNotifyEntry.name + (3,) + tblIdx1, notifyType), - (snmpNotifyEntry.name + (5,) + tblIdx1, 'createAndGo')) + ( + (snmpNotifyEntry.name + (2,) + tblIdx1, transportTag), + (snmpNotifyEntry.name + (3,) + tblIdx1, notifyType), + (snmpNotifyEntry.name + (5,) + tblIdx1, "createAndGo"), + ) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, 'destroy'),) + ((snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyFilterProfileEntry.name + (1,) + tblIdx2, profileName), - (snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, 'createAndGo')) + ( + (snmpNotifyFilterProfileEntry.name + (1,) + tblIdx2, profileName), + (snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, "createAndGo"), + ) ) if not snmpNotifyFilterEntry: return snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyFilterEntry.name + (5,) + tblIdx3, 'destroy'),) + ((snmpNotifyFilterEntry.name + (5,) + tblIdx3, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyFilterEntry.name + (1,) + tblIdx3, filterSubtree), - (snmpNotifyFilterEntry.name + (2,) + tblIdx3, filterMask), - (snmpNotifyFilterEntry.name + (3,) + tblIdx3, filterType), - (snmpNotifyFilterEntry.name + (5,) + tblIdx3, 'createAndGo')) + ( + (snmpNotifyFilterEntry.name + (1,) + tblIdx3, filterSubtree), + (snmpNotifyFilterEntry.name + (2,) + tblIdx3, filterMask), + (snmpNotifyFilterEntry.name + (3,) + tblIdx3, filterType), + (snmpNotifyFilterEntry.name + (5,) + tblIdx3, "createAndGo"), + ) ) -def delNotificationTarget(snmpEngine, notificationName, paramsName, - filterSubtree=None): - (snmpNotifyEntry, tblIdx1, snmpNotifyFilterProfileEntry, - tblIdx2, profileName, snmpNotifyFilterEntry, - tblIdx3) = __cookNotificationTargetInfo(snmpEngine, notificationName, - paramsName, filterSubtree) +def delNotificationTarget(snmpEngine, notificationName, paramsName, filterSubtree=None): + ( + snmpNotifyEntry, + tblIdx1, + snmpNotifyFilterProfileEntry, + tblIdx2, + profileName, + snmpNotifyFilterEntry, + tblIdx3, + ) = __cookNotificationTargetInfo( + snmpEngine, notificationName, paramsName, filterSubtree + ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyEntry.name + (5,) + tblIdx1, 'destroy'),) + ((snmpNotifyEntry.name + (5,) + tblIdx1, "destroy"),) ) snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, 'destroy'),) + ((snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, "destroy"),) ) if not snmpNotifyFilterEntry: return snmpEngine.msgAndPduDsp.mibInstrumController.writeVars( - ((snmpNotifyFilterEntry.name + (5,) + tblIdx3, 'destroy'),) + ((snmpNotifyFilterEntry.name + (5,) + tblIdx3, "destroy"),) ) @@ -829,23 +1044,44 @@ def setInitialVacmParameters(snmpEngine): addVacmGroup(snmpEngine, "initial", 3, "initial") # rfc3415: A.1.4 - addVacmAccess(snmpEngine, "initial", "", 3, "noAuthNoPriv", "exact", - "restricted", None, "restricted") - addVacmAccess(snmpEngine, "initial", "", 3, "authNoPriv", "exact", - "internet", "internet", "internet") - addVacmAccess(snmpEngine, "initial", "", 3, "authPriv", "exact", - "internet", "internet", "internet") + addVacmAccess( + snmpEngine, + "initial", + "", + 3, + "noAuthNoPriv", + "exact", + "restricted", + None, + "restricted", + ) + addVacmAccess( + snmpEngine, + "initial", + "", + 3, + "authNoPriv", + "exact", + "internet", + "internet", + "internet", + ) + addVacmAccess( + snmpEngine, + "initial", + "", + 3, + "authPriv", + "exact", + "internet", + "internet", + "internet", + ) # rfc3415: A.1.5 (semi-secure) - addVacmView(snmpEngine, "internet", - "included", (1, 3, 6, 1), "") - addVacmView(snmpEngine, "restricted", - "included", (1, 3, 6, 1, 2, 1, 1), "") - addVacmView(snmpEngine, "restricted", - "included", (1, 3, 6, 1, 2, 1, 11), "") - addVacmView(snmpEngine, "restricted", - "included", (1, 3, 6, 1, 6, 3, 10, 2, 1), "") - addVacmView(snmpEngine, "restricted", - "included", (1, 3, 6, 1, 6, 3, 11, 2, 1), "") - addVacmView(snmpEngine, "restricted", - "included", (1, 3, 6, 1, 6, 3, 15, 1, 1), "") + addVacmView(snmpEngine, "internet", "included", (1, 3, 6, 1), "") + addVacmView(snmpEngine, "restricted", "included", (1, 3, 6, 1, 2, 1, 1), "") + addVacmView(snmpEngine, "restricted", "included", (1, 3, 6, 1, 2, 1, 11), "") + addVacmView(snmpEngine, "restricted", "included", (1, 3, 6, 1, 6, 3, 10, 2, 1), "") + addVacmView(snmpEngine, "restricted", "included", (1, 3, 6, 1, 6, 3, 11, 2, 1), "") + addVacmView(snmpEngine, "restricted", "included", (1, 3, 6, 1, 6, 3, 15, 1, 1), "") diff --git a/tests/agent_context.py b/tests/agent_context.py new file mode 100644 index 000000000..e88570bc1 --- /dev/null +++ b/tests/agent_context.py @@ -0,0 +1,126 @@ +from pysnmp.carrier.asyncio.dgram import udp +from pysnmp.entity import config, engine +from pysnmp.entity.rfc3413 import cmdrsp, context +from pysnmp.hlapi.asyncio import SnmpEngine +from pysnmp.proto.api import v2c + + +import asyncio +import time + +# Set the port to 1611 instead of 161, because 161 is a +# privileged port and requires root access +AGENT_PORT = 1611 + + +async def start_agent() -> SnmpEngine: + # Create SNMP engine + snmpEngine = engine.SnmpEngine() + + # Set up transport endpoint + config.addTransport( + snmpEngine, + udp.domainName, + udp.UdpTransport().openServerMode(("localhost", AGENT_PORT)), + ) + + # Set up community data + config.addV1System(snmpEngine, "public", "public") + # Add SNMP v3 user + config.addV3User( + snmpEngine, "usr-none-none", config.usmNoAuthProtocol, config.usmNoPrivProtocol + ) + + config.addV3User( + snmpEngine, + "usr-sha-aes", + config.usmHMACSHAAuthProtocol, + "authkey1", + config.usmAesCfb128Protocol, + "privkey1", + ) + + config.addV3User( + snmpEngine, "usr-sha-none", config.usmHMACSHAAuthProtocol, "authkey1" + ) + + # Allow read MIB access for this user / securityModels at VACM + config.addVacmUser(snmpEngine, 1, "public", "noAuthNoPriv", (1, 3, 6), (1, 3, 6)) + config.addVacmUser(snmpEngine, 2, "public", "noAuthNoPriv", (1, 3, 6), (1, 3, 6)) + config.addVacmUser(snmpEngine, 3, "usr-none-none", "noAuthNoPriv", (1, 3, 6)) + config.addVacmUser(snmpEngine, 3, "usr-sha-none", "authNoPriv", (1, 3, 6)) + config.addVacmUser(snmpEngine, 3, "usr-sha-aes", "authPriv", (1, 3, 6)) + + # Configure SNMP context + snmpContext = context.SnmpContext(snmpEngine) + + # --- create custom Managed Object Instance --- + mibBuilder = snmpContext.getMibInstrum().getMibBuilder() + + MibScalar, MibScalarInstance = mibBuilder.importSymbols( + "SNMPv2-SMI", "MibScalar", "MibScalarInstance" + ) + + class MyStaticMibScalarInstance(MibScalarInstance): + # noinspection PyUnusedLocal,PyUnusedLocal + def getValue(self, name, idx): + time.sleep(3) # Add a 2-second sleep + return self.getSyntax().clone(f"Test agent") + + def setValue(self, name, idx, value): + # Here you can handle the SET operation. + # `value` is the new value for the scalar instance. + # You should store this value somewhere, and return it in the `getValue` method. + # For this example, let's just print it. + print(f"SET operation received. New value: {value}") + return self.getSyntax().clone(value) + + mibBuilder.exportSymbols( + "__MY_MIB", + MibScalar((1, 3, 6, 1, 4, 1, 60069, 9, 1), v2c.OctetString()), + MyStaticMibScalarInstance( + (1, 3, 6, 1, 4, 1, 60069, 9, 1), (0,), v2c.OctetString() + ), + ) + + # --- end of Managed Object Instance initialization ---- + + # Register SNMP Applications at the SNMP engine for particular SNMP context + cmdrsp.GetCommandResponder(snmpEngine, snmpContext) + cmdrsp.NextCommandResponder(snmpEngine, snmpContext) + cmdrsp.BulkCommandResponder(snmpEngine, snmpContext) + cmdrsp.SetCommandResponder(snmpEngine, snmpContext) + + # Start the event loop + snmpEngine.transportDispatcher.jobStarted(1) + + snmpEngine.transportDispatcher.runDispatcher() + + # Wait for the agent to start + await asyncio.sleep(1) + + # return the engine + return snmpEngine + + +class AgentContextManager: + """ + A context manager for managing the lifecycle of an SNMP test agent. + + Usage: + async with AgentContextManager() as agent: + # Perform operations with the agent + + When the context is entered, the agent is started using the `start_agent()` function. + When the context is exited, the agent's transport dispatcher is stopped and closed. + + Note: The `start_agent()` function and the `transportDispatcher` attribute are not defined in this code snippet. + """ + + async def __aenter__(self): + self.agent = await start_agent() + return self.agent + + async def __aexit__(self, exc_type, exc_val, exc_tb): + self.agent.transportDispatcher.jobFinished(1) + self.agent.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py b/tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py index f515514c3..00cc4067b 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py @@ -1,21 +1,27 @@ import pytest from pysnmp.hlapi.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_custom_asn1_mib_search_path(): - snmpEngine = SnmpEngine() - errorIndication, errorStatus, errorIndex, varBinds = await getCmd( - snmpEngine, - CommunityData('public'), - UdpTransportTarget(('demo.pysnmp.com', 161)), - ContextData(), - ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1).addAsn1MibSource('file:///usr/share/snmp', - 'https://mibs.pysnmp.com/asn1/@mib@')) - ) + async with AgentContextManager(): + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpEngine, + CommunityData("public"), + UdpTransportTarget(("localhost", AGENT_PORT)), + ContextData(), + ObjectType( + ObjectIdentity("IF-MIB", "ifInOctets", 1).addAsn1MibSource( + "file:///usr/share/snmp", "https://mibs.pysnmp.com/asn1/@mib@" + ) + ), + ) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == 'IF-MIB::ifInOctets.1' + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "IF-MIB::ifInOctets.1" - snmpEngine.transportDispatcher.closeDispatcher() \ No newline at end of file + snmpEngine.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_usm_none_none.py b/tests/hlapi/asyncio/manager/cmdgen/test_usm_none_none.py index e84017e54..18ad109a9 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_usm_none_none.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_usm_none_none.py @@ -1,41 +1,43 @@ import pytest from pysnmp.hlapi.asyncio import * from pysnmp.proto.errind import UnknownUserName +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_usm_no_auth_no_priv(): - snmpEngine = SnmpEngine() - authData = UsmUserData( - "usr-none-none" - ) - errorIndication, errorStatus, errorIndex, varBinds = await getCmd( - snmpEngine, - authData, - UdpTransportTarget(("demo.pysnmp.com", 161), retries=0), - ContextData(), - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) + async with AgentContextManager(): + snmpEngine = SnmpEngine() + authData = UsmUserData("usr-none-none") + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpEngine, + authData, + UdpTransportTarget(("localhost", AGENT_PORT), retries=0), + ContextData(), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + isinstance(varBinds[0][1], OctetString) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" - isinstance(varBinds[0][1], OctetString) @pytest.mark.asyncio async def test_usm_no_auth_no_priv_wrong_user(): - snmpEngine = SnmpEngine() - authData = UsmUserData( - "usr-none-none-not-exist" - ) - errorIndication, errorStatus, errorIndex, varBinds = await getCmd( - snmpEngine, - authData, - UdpTransportTarget(("demo.pysnmp.com", 161), retries=0), - ContextData(), - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) + async with AgentContextManager(): + snmpEngine = SnmpEngine() + authData = UsmUserData("usr-none-none-not-exist") + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpEngine, + authData, + UdpTransportTarget(("localhost", AGENT_PORT), retries=0), + ContextData(), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) - assert isinstance(errorIndication, UnknownUserName) - assert str(errorIndication) == 'Unknown USM user' - snmpEngine.transportDispatcher.closeDispatcher() + assert isinstance(errorIndication, UnknownUserName) + assert str(errorIndication) == "Unknown USM user" + snmpEngine.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py b/tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py index 6415ef5e6..10d270fcc 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py @@ -1,21 +1,26 @@ import pytest from pysnmp.hlapi.asyncio import * +from tests.agent_context import AgentContextManager + + @pytest.mark.asyncio async def test_v1_get(): - snmpEngine = SnmpEngine() - errorIndication, errorStatus, errorIndex, varBinds = await getCmd( - snmpEngine, - CommunityData('public', mpModel=0), - UdpTransportTarget(('demo.pysnmp.com', 161)), - ContextData(), - ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) - ) + async with AgentContextManager(): + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpEngine, + CommunityData("public", mpModel=0), + UdpTransportTarget(("localhost", 1611)), + ContextData(), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == 'SNMPv2-MIB::sysDescr.0' - assert isinstance(varBinds[0][1], OctetString) + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + assert isinstance(varBinds[0][1], OctetString) - snmpEngine.transportDispatcher.closeDispatcher() \ No newline at end of file + snmpEngine.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py b/tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py index 18fecf2ef..2721d2d07 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py @@ -14,25 +14,28 @@ | $ snmpgetnext -v1 -c public demo.pysnmp.com SNMPv2-MIB::sysDescr.0 -"""# +""" # import asyncio import pytest from pysnmp.hlapi.asyncio.slim import Slim from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_v1_next(): - slim = Slim(1) - errorIndication, errorStatus, errorIndex, varBinds = await slim.next( - 'public', - 'demo.pysnmp.com', - 161, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) - - assert errorIndication is None - assert errorStatus == 0 - assert errorIndex == 0 - assert len(varBinds) > 0 - - slim.close() \ No newline at end of file + async with AgentContextManager(): + slim = Slim(1) + errorIndication, errorStatus, errorIndex, varBinds = await slim.next( + "public", + "localhost", + AGENT_PORT, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + + slim.close() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py b/tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py index eafc457d9..eb6f266f4 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py @@ -1,21 +1,24 @@ import pytest from pysnmp.hlapi.asyncio.slim import Slim from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_v1_set(): - slim = Slim(1) - errorIndication, errorStatus, errorIndex, varBinds = await slim.set( - 'public', - 'demo.pysnmp.com', - 161, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai") - ) + async with AgentContextManager(): + slim = Slim(1) + errorIndication, errorStatus, errorIndex, varBinds = await slim.set( + "public", + "localhost", + AGENT_PORT, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai"), + ) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == 'SNMPv2-MIB::sysLocation.0' - assert varBinds[0][1].prettyPrint() == 'Shanghai' + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysLocation.0" + assert varBinds[0][1].prettyPrint() == "Shanghai" - slim.close() \ No newline at end of file + slim.close() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py index 707881df8..ae27d09f6 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py @@ -1,21 +1,24 @@ import pytest from pysnmp.hlapi.asyncio.slim import Slim from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_v2c_bulk(): - slim = Slim() - errorIndication, errorStatus, errorIndex, varBinds = await slim.bulk( - 'public', - 'demo.pysnmp.com', - 161, - 0, - 50, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) + async with AgentContextManager(): + slim = Slim() + errorIndication, errorStatus, errorIndex, varBinds = await slim.bulk( + "public", + "localhost", + AGENT_PORT, + 0, + 50, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) > 0 + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 50 - slim.close() \ No newline at end of file + slim.close() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py index df523f97c..139ceae9d 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py @@ -1,21 +1,25 @@ import pytest from pysnmp.hlapi.asyncio.slim import Slim from pysnmp.hlapi.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_v1_get(): - slim = Slim() - errorIndication, errorStatus, errorIndex, varBinds = await slim.get( - 'public', - 'demo.pysnmp.com', - 161, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) + async with AgentContextManager(): + slim = Slim() + errorIndication, errorStatus, errorIndex, varBinds = await slim.get( + "public", + "localhost", + AGENT_PORT, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == 'SNMPv2-MIB::sysDescr.0' - assert isinstance(varBinds[0][1], OctetString) + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + assert isinstance(varBinds[0][1], OctetString) - slim.close() \ No newline at end of file + slim.close() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py index f34356c4f..f6d39c04d 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py @@ -14,25 +14,28 @@ | $ snmpgetnext -v1 -c public demo.pysnmp.com SNMPv2-MIB::sysDescr.0 -"""# +""" # import asyncio import pytest from pysnmp.hlapi.asyncio.slim import Slim from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio async def test_v1_next(): - slim = Slim() - errorIndication, errorStatus, errorIndex, varBinds = await slim.next( - 'public', - 'demo.pysnmp.com', - 161, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) - - assert errorIndication is None - assert errorStatus == 0 - assert errorIndex == 0 - assert len(varBinds) > 0 - - slim.close() \ No newline at end of file + async with AgentContextManager(): + slim = Slim() + errorIndication, errorStatus, errorIndex, varBinds = await slim.next( + "public", + "localhost", + AGENT_PORT, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + + slim.close() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py index 2e71ddc4e..26796ea56 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py +++ b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py @@ -1,21 +1,24 @@ import pytest from pysnmp.hlapi.asyncio.slim import Slim from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + @pytest.mark.asyncio -async def test_v1_set(): - slim = Slim() - errorIndication, errorStatus, errorIndex, varBinds = await slim.set( - 'public', - 'demo.pysnmp.com', - 161, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai") - ) +async def test_v2_set(): + async with AgentContextManager(): + slim = Slim() + errorIndication, errorStatus, errorIndex, varBinds = await slim.set( + "public", + "localhost", + AGENT_PORT, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai"), + ) - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == 'SNMPv2-MIB::sysLocation.0' - assert varBinds[0][1].prettyPrint() == 'Shanghai' + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysLocation.0" + assert varBinds[0][1].prettyPrint() == "Shanghai" - slim.close() \ No newline at end of file + slim.close()