diff --git a/pysnmp/hlapi/v1arch/__init__.py b/pysnmp/hlapi/v1arch/__init__.py index 6aeacec7..1c9bedde 100644 --- a/pysnmp/hlapi/v1arch/__init__.py +++ b/pysnmp/hlapi/v1arch/__init__.py @@ -4,7 +4,7 @@ # Copyright (c) 2005-2020, Ilya Etingof # License: https://www.pysnmp.com/pysnmp/license.html # -from pysnmp.hlapi.v1arch.auth import * +from pysnmp.hlapi.v1arch.asyncio.auth import * from pysnmp.hlapi.v1arch.asyncio.dispatch import * from pysnmp.proto.rfc1902 import * from pysnmp.proto.rfc1905 import EndOfMibView diff --git a/pysnmp/hlapi/v1arch/asyncio/__init__.py b/pysnmp/hlapi/v1arch/asyncio/__init__.py index 18c420ab..37ba0158 100644 --- a/pysnmp/hlapi/v1arch/asyncio/__init__.py +++ b/pysnmp/hlapi/v1arch/asyncio/__init__.py @@ -4,10 +4,11 @@ # Copyright (c) 2005-2020, Ilya Etingof # License: https://www.pysnmp.com/pysnmp/license.html # -from pysnmp.hlapi.v1arch.auth import * +from pysnmp.hlapi.v1arch.asyncio.auth import * from pysnmp.hlapi.v1arch.asyncio.transport import * from pysnmp.hlapi.v1arch.asyncio.cmdgen import * from pysnmp.hlapi.v1arch.asyncio.ntforg import * from pysnmp.hlapi.v1arch.asyncio.dispatch import * +from pysnmp.hlapi.v1arch.asyncio.slim import * from pysnmp.proto.rfc1902 import * from pysnmp.smi.rfc1902 import * diff --git a/pysnmp/hlapi/v1arch/auth.py b/pysnmp/hlapi/v1arch/asyncio/auth.py similarity index 100% rename from pysnmp/hlapi/v1arch/auth.py rename to pysnmp/hlapi/v1arch/asyncio/auth.py diff --git a/pysnmp/hlapi/v1arch/asyncio/cmdgen.py b/pysnmp/hlapi/v1arch/asyncio/cmdgen.py index f04d7e0f..f9c10fcf 100644 --- a/pysnmp/hlapi/v1arch/asyncio/cmdgen.py +++ b/pysnmp/hlapi/v1arch/asyncio/cmdgen.py @@ -4,21 +4,40 @@ # Copyright (c) 2005-2020, Ilya Etingof # License: https://www.pysnmp.com/pysnmp/license.html # -from pysnmp.hlapi.v1arch.auth import * +from typing import AsyncGenerator +from pysnmp.hlapi.transport import AbstractTransportTarget +from pysnmp.hlapi.v1arch.asyncio.dispatch import SnmpDispatcher +from pysnmp.hlapi.v1arch.asyncio.auth import * from pysnmp.hlapi import varbinds from pysnmp.hlapi.v1arch.asyncio.transport import * +from pysnmp.proto.rfc1902 import Integer32, Null +from pysnmp.proto.rfc1905 import EndOfMibView, endOfMibView from pysnmp.smi.rfc1902 import * -from pysnmp.proto import api +from pysnmp.proto import api, errind import asyncio -__all__ = ["getCmd", "nextCmd", "setCmd", "bulkCmd", "isEndOfMib"] +__all__ = [ + "getCmd", + "nextCmd", + "setCmd", + "bulkCmd", + "walkCmd", + "bulkWalkCmd", + "isEndOfMib", +] VB_PROCESSOR = varbinds.CommandGeneratorVarBinds() isEndOfMib = varbinds.isEndOfMib -async def getCmd(snmpDispatcher, authData, transportTarget, *varBinds, **options): +async def getCmd( + snmpDispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + *varBinds: ObjectType, + **options +) -> "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GET query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -141,7 +160,13 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): return await future -async def setCmd(snmpDispatcher, authData, transportTarget, *varBinds, **options): +async def setCmd( + snmpDispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + *varBinds: ObjectType, + **options +) -> "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP SET query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -264,7 +289,13 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): return await future -async def nextCmd(snmpDispatcher, authData, transportTarget, *varBinds, **options): +async def nextCmd( + snmpDispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + *varBinds: ObjectType, + **options +) -> "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GETNEXT query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -360,12 +391,9 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): if lookupMib: try: - varBindTable = [ - VB_PROCESSOR.unmakeVarBinds( - snmpDispatcher.cache, varBindTableRow, lookupMib - ) - for varBindTableRow in varBindTable - ] + varBindTable = VB_PROCESSOR.unmakeVarBinds( + snmpDispatcher.cache, varBindTable[0], lookupMib + ) except Exception as e: future.set_exception(e) @@ -395,14 +423,14 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): async def bulkCmd( - snmpDispatcher, - authData, - transportTarget, - nonRepeaters, - maxRepetitions, - *varBinds, + snmpDispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + nonRepeaters: int, + maxRepetitions: int, + *varBinds: ObjectType, **options -): +) -> "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GETBULK query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -520,12 +548,9 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): if lookupMib: try: - varBindTable = [ - VB_PROCESSOR.unmakeVarBinds( - snmpDispatcher.cache, varBindTableRow, lookupMib - ) - for varBindTableRow in varBindTable - ] + varBindTable = VB_PROCESSOR.unmakeVarBinds( + snmpDispatcher.cache, varBindTable[0], lookupMib + ) except Exception as e: future.set_exception(e) @@ -554,3 +579,409 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): snmpDispatcher.sendPdu(authData, transportTarget, reqPdu, cbFun=_cbFun) return await future + + +async def walkCmd( + dispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + varBind: ObjectType, + **options +) -> AsyncGenerator[ + "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]", + None, +]: + r"""Creates a generator to perform one or more SNMP GETNEXT queries. + + On each iteration, new SNMP GETNEXT request is send + (:RFC:`1905#section-4.2.2`). The iterator blocks waiting for response + to arrive or error to occur. + + Parameters + ---------- + snmpEngine : :py:class:`~pysnmp.hlapi.v1arch.asyncio.SnmpDispatcher` + Class instance representing SNMP engine. + + authData : :py:class:`~pysnmp.hlapi.v1arch.asyncio.CommunityData` + Class instance representing SNMP credentials. + + transportTarget : :py:class:`~pysnmp.hlapi.v1arch.asyncio.UdpTransportTarget` or :py:class:`~pysnmp.hlapi.v1arch.asyncio.Udp6TransportTarget` + Class instance representing transport type along with SNMP peer address. + + varBind : :py:class:`~pysnmp.smi.rfc1902.ObjectType` + One class instance representing MIB variables to place + into SNMP request. + + Other Parameters + ---------------- + \*\*options : + Request options: + + * `lookupMib` - load MIB and resolve response MIB variables at + the cost of slightly reduced performance. Default is `True`. + * `lexicographicMode` - walk SNMP agent's MIB till the end (if `True`), + otherwise (if `False`) stop iteration when all response MIB + variables leave the scope of initial MIB variables in + `varBinds`. Default is `True`. + * `ignoreNonIncreasingOid` - continue iteration even if response + MIB variables (OIDs) are not greater then request MIB variables. + Be aware that setting it to `True` may cause infinite loop between + SNMP management and agent applications. Default is `False`. + * `maxRows` - stop iteration once this generator instance processed + `maxRows` of SNMP conceptual table. Default is `0` (no limit). + * `maxCalls` - stop iteration once this generator instance processed + `maxCalls` responses. Default is 0 (no limit). + + Yields + ------ + errorIndication : str + True value indicates SNMP engine error. + errorStatus : str + True value indicates SNMP PDU error. + errorIndex : int + Non-zero value refers to `varBinds[errorIndex-1]` + varBinds : tuple + A sequence of :py:class:`~pysnmp.smi.rfc1902.ObjectType` class + instances representing MIB variables returned in SNMP response. + + Raises + ------ + PySnmpError + Or its derivative indicating that an error occurred while + performing SNMP operation. + + Notes + ----- + The `nextCmd` generator will be exhausted on any of the following + conditions: + + * SNMP engine error occurs thus `errorIndication` is `True` + * SNMP PDU `errorStatus` is reported as `True` + * SNMP :py:class:`~pysnmp.proto.rfc1905.EndOfMibView` values + (also known as *SNMP exception values*) are reported for all + MIB variables in `varBinds` + * *lexicographicMode* option is `True` and SNMP agent reports + end-of-mib or *lexicographicMode* is `False` and all + response MIB variables leave the scope of `varBinds` + + At any moment a new sequence of `varBinds` could be send back into + running generator (supported since Python 2.6). + + Examples + -------- + >>> from pysnmp.hlapi.v1arch.asyncio import * + >>> g = await walkCmd(SnmpEngine(), + ... CommunityData('public'), + ... await UdpTransportTarget.create(('demo.pysnmp.com', 161)), + ... ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr'))) + >>> next(g) + (None, 0, 0, [ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.1.0')), DisplayString('SunOS zeus.pysnmp.com 4.1.3_U1 1 sun4m'))]) + >>> g.send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] ) + (None, 0, 0, [(ObjectName('1.3.6.1.2.1.2.2.1.10.1'), Counter32(284817787))]) + """ + lexicographicMode = options.get("lexicographicMode", True) + ignoreNonIncreasingOid = options.get("ignoreNonIncreasingOid", False) + maxRows = options.get("maxRows", 0) + maxCalls = options.get("maxCalls", 0) + + initialVars = [ + x[0] for x in VB_PROCESSOR.makeVarBinds(dispatcher.cache, (varBind,)) + ] + + totalRows = totalCalls = 0 + + while True: + if varBind: + errorIndication, errorStatus, errorIndex, varBindTable = await nextCmd( + dispatcher, + authData, + transportTarget, + (varBind[0], Null("")), + **dict(lookupMib=options.get("lookupMib", True)) + ) + if ( + ignoreNonIncreasingOid + and errorIndication + and isinstance(errorIndication, errind.OidNotIncreasing) + ): + errorIndication = None + + if errorIndication: + yield (errorIndication, errorStatus, errorIndex, (varBind,)) + return + elif errorStatus: + if errorStatus == 2: + # Hide SNMPv1 noSuchName error which leaks in here + # from SNMPv1 Agent through internal pysnmp proxy. + errorStatus = 0 + errorIndex = 0 + return + else: + stopFlag = True + + varBind = varBindTable[0] + + name, val = varBind + foundEnding = isinstance(val, Null) or isinstance(val, EndOfMibView) + foundBeyond = not lexicographicMode and not initialVars[0].isPrefixOf( + name + ) + if foundEnding or foundBeyond: + return + + if stopFlag and varBind[1] is not endOfMibView: + stopFlag = False + + if stopFlag: + return + + totalRows += 1 + totalCalls += 1 + else: + errorIndication = errorStatus = errorIndex = None + varBind = None + + initialVarBinds: "tuple[ObjectType, ...]" = ( + yield errorIndication, + errorStatus, + errorIndex, + (varBind,), + ) + + if initialVarBinds: + varBind = initialVarBinds[0] + initialVars = [ + x[0] + for x in VB_PROCESSOR.makeVarBinds(dispatcher.cache, initialVarBinds) + ] + + if maxRows and totalRows >= maxRows: + return + + if maxCalls and totalCalls >= maxCalls: + return + + +async def bulkWalkCmd( + dispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + nonRepeaters: int, + maxRepetitions: int, + varBind: ObjectType, + **options +) -> AsyncGenerator[ + "tuple[errind.ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]", + None, +]: + r"""Creates a generator to perform one or more SNMP GETBULK queries. + + On each iteration, new SNMP GETBULK request is send + (:RFC:`1905#section-4.2.3`). The iterator blocks waiting for response + to arrive or error to occur. + + Parameters + ---------- + snmpEngine : :py:class:`~pysnmp.hlapi.v1arch.asyncio.SnmpDispatcher` + Class instance representing SNMP engine. + + authData : :py:class:`~pysnmp.hlapi.v1arch.asyncio.CommunityData` + Class instance representing SNMP credentials. + + transportTarget : :py:class:`~pysnmp.hlapi.v1arch.asyncio.UdpTransportTarget` or :py:class:`~pysnmp.hlapi.v1arch.asyncio.Udp6TransportTarget` + Class instance representing transport type along with SNMP peer address. + + nonRepeaters : int + One MIB variable is requested in response for the first + `nonRepeaters` MIB variables in request. + + maxRepetitions : int + `maxRepetitions` MIB variables are requested in response for each + of the remaining MIB variables in the request (e.g. excluding + `nonRepeaters`). Remote SNMP engine may choose lesser value than + requested. + + varBind : :py:class:`~pysnmp.smi.rfc1902.ObjectType` + One class instance representing MIB variables to place + into SNMP request. + + Other Parameters + ---------------- + \*\*options : + Request options: + + * `lookupMib` - load MIB and resolve response MIB variables at + the cost of slightly reduced performance. Default is `True`. + * `lexicographicMode` - walk SNMP agent's MIB till the end (if `True`), + otherwise (if `False`) stop iteration when all response MIB + variables leave the scope of initial MIB variables in + `varBinds`. Default is `True`. + * `ignoreNonIncreasingOid` - continue iteration even if response + MIB variables (OIDs) are not greater then request MIB variables. + Be aware that setting it to `True` may cause infinite loop between + SNMP management and agent applications. Default is `False`. + * `maxRows` - stop iteration once this generator instance processed + `maxRows` of SNMP conceptual table. Default is `0` (no limit). + * `maxCalls` - stop iteration once this generator instance processed + `maxCalls` responses. Default is 0 (no limit). + + Yields + ------ + errorIndication : str + True value indicates SNMP engine error. + errorStatus : str + True value indicates SNMP PDU error. + errorIndex : int + Non-zero value refers to \*varBinds[errorIndex-1] + varBinds : tuple + A sequence of :py:class:`~pysnmp.smi.rfc1902.ObjectType` class + instances representing MIB variables returned in SNMP response. + + Raises + ------ + PySnmpError + Or its derivative indicating that an error occurred while + performing SNMP operation. + + Notes + ----- + The `bulkCmd` generator will be exhausted on any of the following + conditions: + + * SNMP engine error occurs thus `errorIndication` is `True` + * SNMP PDU `errorStatus` is reported as `True` + * SNMP :py:class:`~pysnmp.proto.rfc1905.EndOfMibView` values + (also known as *SNMP exception values*) are reported for all + MIB variables in `varBinds` + * *lexicographicMode* option is `True` and SNMP agent reports + end-of-mib or *lexicographicMode* is `False` and all + response MIB variables leave the scope of `varBinds` + + At any moment a new sequence of `varBinds` could be send back into + running generator (supported since Python 2.6). + + Setting `maxRepetitions` value to 15..50 might significantly improve + system performance, as many MIB variables get packed into a single + response message at once. + + Examples + -------- + >>> from pysnmp.hlapi.v1arch.asyncio import * + >>> g = await bulkWalkCmd(SnmpEngine(), + ... CommunityData('public'), + ... await UdpTransportTarget.create(('demo.pysnmp.com', 161)), + ... 0, 25, + ... ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr'))) + >>> next(g) + (None, 0, 0, [ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.1.0')), DisplayString('SunOS zeus.pysnmp.com 4.1.3_U1 1 sun4m'))]) + >>> g.send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] ) + (None, 0, 0, [(ObjectName('1.3.6.1.2.1.2.2.1.10.1'), Counter32(284817787))]) + """ + lexicographicMode = options.get("lexicographicMode", True) + ignoreNonIncreasingOid = options.get("ignoreNonIncreasingOid", False) + maxRows = options.get("maxRows", 0) + maxCalls = options.get("maxCalls", 0) + + initialVars = [ + x[0] for x in VB_PROCESSOR.makeVarBinds(dispatcher.cache, (varBind,)) + ] + + totalRows = totalCalls = 0 + + varBinds: "tuple[ObjectType, ...]" = (varBind,) + + while True: + if maxRows and totalRows < maxRows: + maxRepetitions = min(maxRepetitions, maxRows - totalRows) + + if varBinds: + errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd( + dispatcher, + authData, + transportTarget, + nonRepeaters, + maxRepetitions, + *[ObjectType(varBinds[-1][0], Null(""))], + **dict(lookupMib=options.get("lookupMib", True)) + ) + + if ( + ignoreNonIncreasingOid + and errorIndication + and isinstance(errorIndication, errind.OidNotIncreasing) + ): + errorIndication = None + + if errorIndication: + yield ( + errorIndication, + errorStatus, + errorIndex, + varBindTable and varBinds, + ) + if errorIndication != errind.requestTimedOut: + return + elif errorStatus: + if errorStatus == 2: + # Hide SNMPv1 noSuchName error which leaks in here + # from SNMPv1 Agent through internal pysnmp proxy. + errorStatus = 0 + errorIndex = 0 + yield ( + errorIndication, + errorStatus, + errorIndex, + varBinds, + ) + return + else: + stopFlag = True + varBinds = varBindTable + + for col, varBind in enumerate(varBinds): + name, val = varBind + foundEnding = isinstance(val, Null) or isinstance(val, EndOfMibView) + foundBeyond = not lexicographicMode and not initialVars[ + 0 + ].isPrefixOf(name) + if foundEnding or foundBeyond: + stopFlag = True + result = varBinds[:col] + if len(result) > 0: + yield ( + errorIndication, + errorStatus, + errorIndex, + result, + ) + return + + if stopFlag and varBinds[col][1] is not endOfMibView: + stopFlag = False + + if stopFlag: + return + + totalRows += 1 + totalCalls += 1 + else: + errorIndication = errorStatus = errorIndex = None + varBinds = () + + initialVarBinds: "tuple[ObjectType, ...]" = ( + yield errorIndication, + errorStatus, + errorIndex, + varBinds, + ) + if initialVarBinds: + varBinds = initialVarBinds + initialVars = [ + x[0] for x in VB_PROCESSOR.makeVarBinds(dispatcher.cache, varBinds) + ] + + if maxRows and totalRows >= maxRows: + return + + if maxCalls and totalCalls >= maxCalls: + return diff --git a/pysnmp/hlapi/v1arch/asyncio/ntforg.py b/pysnmp/hlapi/v1arch/asyncio/ntforg.py index 0d97c66c..0e15268b 100644 --- a/pysnmp/hlapi/v1arch/asyncio/ntforg.py +++ b/pysnmp/hlapi/v1arch/asyncio/ntforg.py @@ -4,13 +4,16 @@ # Copyright (c) 2005-2020, Ilya Etingof # License: https://www.pysnmp.com/pysnmp/license.html # -from pysnmp.hlapi.v1arch.auth import * +from pysnmp.hlapi.transport import AbstractTransportTarget +from pysnmp.hlapi.v1arch.asyncio.auth import * +from pysnmp.hlapi.v1arch.asyncio.dispatch import SnmpDispatcher from pysnmp.hlapi.varbinds import * from pysnmp.hlapi.v1arch.asyncio.transport import * from pysnmp.proto.api import v2c from pysnmp.proto.proxy import rfc2576 +from pysnmp.proto.rfc1902 import Integer32 from pysnmp.smi.rfc1902 import * -from pysnmp.proto import api +from pysnmp.proto import api, errind, error import asyncio @@ -20,8 +23,13 @@ async def sendNotification( - snmpDispatcher, authData, transportTarget, notifyType, *varBinds, **options -): + snmpDispatcher: SnmpDispatcher, + authData: CommunityData, + transportTarget: AbstractTransportTarget, + notifyType: str, + *varBinds: ObjectType, + **options +) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, tuple[ObjectType, ...]]": r"""Creates a generator to send SNMP notification. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -166,6 +174,7 @@ def _ensureVarBinds(varBinds): return varBinds def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): + lookupMib, future = _cbCtx if future.cancelled(): return @@ -231,7 +240,9 @@ def _cbFun(snmpDispatcher, stateHandle, errorIndication, rspPdu, _cbCtx): future = asyncio.Future() - snmpDispatcher.sendPdu(authData, transportTarget, reqPdu, cbFun=_cbFun) + snmpDispatcher.sendPdu( + authData, transportTarget, reqPdu, cbFun=_cbFun, cbCtx=(lookupMib, future) + ) if notifyType == "trap": diff --git a/pysnmp/hlapi/v3arch/asyncio/slim.py b/pysnmp/hlapi/v1arch/asyncio/slim.py similarity index 92% rename from pysnmp/hlapi/v3arch/asyncio/slim.py rename to pysnmp/hlapi/v1arch/asyncio/slim.py index a3c8b3e5..18989c7b 100644 --- a/pysnmp/hlapi/v3arch/asyncio/slim.py +++ b/pysnmp/hlapi/v1arch/asyncio/slim.py @@ -4,12 +4,11 @@ # Copyright (c) 2023-2024, LeXtudio Inc. # License: https://www.pysnmp.com/pysnmp/license.html # -from pysnmp.entity.engine import SnmpEngine from pysnmp.error import PySnmpError -from pysnmp.hlapi.v3arch.asyncio.auth import CommunityData -from pysnmp.hlapi.v3arch.asyncio.cmdgen import bulkCmd, getCmd, nextCmd, setCmd -from pysnmp.hlapi.v3arch.asyncio.context import ContextData -from pysnmp.hlapi.v3arch.asyncio.transport import ( +from pysnmp.hlapi.v1arch.asyncio.auth import CommunityData +from pysnmp.hlapi.v1arch.asyncio.cmdgen import bulkCmd, getCmd, nextCmd, setCmd +from pysnmp.hlapi.v1arch.asyncio.dispatch import SnmpDispatcher +from pysnmp.hlapi.v1arch.asyncio.transport import ( Udp6TransportTarget, UdpTransportTarget, ) @@ -43,18 +42,18 @@ class Slim: >>> """ - snmpEngine: SnmpEngine + snmpDispatcher: SnmpDispatcher version: int def __init__(self, version: int = 2): - self.snmpEngine = SnmpEngine() + self.snmpDispatcher = SnmpDispatcher() if version not in (1, 2): raise PySnmpError(f"Not supported version {version}") self.version = version def close(self): """Closes the wrapper to release its resources.""" - self.snmpEngine.closeDispatcher() + self.snmpDispatcher.transportDispatcher.closeDispatcher() def __enter__(self): return self @@ -70,7 +69,7 @@ async def get( *varBinds, timeout: int = 1, retries: int = 5, - ) -> "tuple[ErrorIndication, Integer32 | int, Integer32 | int, list[ObjectType]]": + ) -> "tuple[ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GET query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -123,7 +122,7 @@ async def get( Examples -------- >>> import asyncio - >>> from pysnmp.hlapi.v3arch.asyncio.slim import Slim + >>> from pysnmp.hlapi.v1arch.asyncio.slim import Slim >>> >>> async def run(): ... with Slim() as slim: @@ -140,12 +139,11 @@ async def get( >>> """ return await getCmd( - self.snmpEngine, + self.snmpDispatcher, CommunityData(communityName, mpModel=self.version - 1), await Udp6TransportTarget.create((address, port), timeout, retries) if ":" in address else await UdpTransportTarget.create((address, port), timeout, retries), - ContextData(), *varBinds, ) @@ -157,7 +155,7 @@ async def next( *varBinds, timeout: int = 1, retries: int = 5, - ) -> "tuple[ErrorIndication, Integer32 | str | int, Integer32 | int, list[ObjectType]]": + ) -> "tuple[ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GETNEXT query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -214,7 +212,7 @@ async def next( Examples -------- >>> import asyncio - >>> from pysnmp.hlapi.v3arch.asyncio.slim import Slim + >>> from pysnmp.hlapi.v1arch.asyncio.slim import Slim >>> >>> async def run(): ... with Slim() as slim: @@ -231,12 +229,11 @@ async def next( >>> """ return await nextCmd( - self.snmpEngine, + self.snmpDispatcher, CommunityData(communityName, mpModel=self.version - 1), await Udp6TransportTarget.create((address, port), timeout, retries) if ":" in address else await UdpTransportTarget.create((address, port), timeout, retries), - ContextData(), *varBinds, ) @@ -250,7 +247,7 @@ async def bulk( *varBinds, timeout: int = 1, retries: int = 5, - ) -> "tuple[ErrorIndication, Integer32 | str | int, Integer32 | int, list[ObjectType]]": + ) -> "tuple[ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GETBULK query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -332,7 +329,7 @@ async def bulk( Examples -------- >>> import asyncio - >>> from pysnmp.hlapi.v3arch.asyncio.slim import Slim + >>> from pysnmp.hlapi.v1arch.asyncio.slim import Slim >>> >>> async def run(): ... with Slim() as slim: @@ -354,12 +351,11 @@ async def bulk( if version == 0: raise PySnmpError("Cannot send V2 PDU on V1 session") return await bulkCmd( - self.snmpEngine, + self.snmpDispatcher, CommunityData(communityName, mpModel=version), await Udp6TransportTarget.create((address, port), timeout, retries) if ":" in address else await UdpTransportTarget.create((address, port), timeout, retries), - ContextData(), nonRepeaters, maxRepetitions, *varBinds, @@ -373,7 +369,7 @@ async def set( *varBinds, timeout: int = 1, retries: int = 5, - ) -> "tuple[ErrorIndication, Integer32 | int, Integer32 | int, list[ObjectType]]": + ) -> "tuple[ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP SET query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -423,7 +419,7 @@ async def set( Examples -------- >>> import asyncio - >>> from pysnmp.hlapi.v3arch.asyncio.slim import Slim + >>> from pysnmp.hlapi.v1arch.asyncio.slim import Slim >>> >>> async def run(): ... with Slim() as slim: @@ -440,11 +436,10 @@ async def set( >>> """ return await setCmd( - self.snmpEngine, + self.snmpDispatcher, CommunityData(communityName, mpModel=self.version - 1), await Udp6TransportTarget.create((address, port), timeout, retries) if ":" in address else await UdpTransportTarget.create((address, port), timeout, retries), - ContextData(), *varBinds, ) diff --git a/pysnmp/hlapi/v3arch/asyncio/__init__.py b/pysnmp/hlapi/v3arch/asyncio/__init__.py index df4bdda0..9b2cf332 100644 --- a/pysnmp/hlapi/v3arch/asyncio/__init__.py +++ b/pysnmp/hlapi/v3arch/asyncio/__init__.py @@ -13,7 +13,6 @@ from pysnmp.hlapi.v3arch.asyncio.cmdgen import * from pysnmp.hlapi.v3arch.asyncio.context import * from pysnmp.hlapi.v3arch.asyncio.ntforg import * -from pysnmp.hlapi.v3arch.asyncio.slim import * from pysnmp.hlapi.v3arch.asyncio.transport import * from pysnmp.proto.rfc1902 import * from pysnmp.proto.rfc1905 import EndOfMibView, NoSuchInstance, NoSuchObject diff --git a/pysnmp/hlapi/v3arch/asyncio/cmdgen.py b/pysnmp/hlapi/v3arch/asyncio/cmdgen.py index e72a592c..302f22e9 100644 --- a/pysnmp/hlapi/v3arch/asyncio/cmdgen.py +++ b/pysnmp/hlapi/v3arch/asyncio/cmdgen.py @@ -69,9 +69,9 @@ async def getCmd( authData: "CommunityData | UsmUserData", transportTarget: AbstractTransportTarget, contextData: ContextData, - *varBinds, + *varBinds: ObjectType, **options -) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, list[ObjectType]]": +) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GET query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -190,9 +190,9 @@ async def setCmd( authData: "CommunityData | UsmUserData", transportTarget: AbstractTransportTarget, contextData: ContextData, - *varBinds, + *varBinds: ObjectType, **options -) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, list[ObjectType]]": +) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP SET query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -311,9 +311,9 @@ async def nextCmd( authData: "CommunityData | UsmUserData", transportTarget: AbstractTransportTarget, contextData: ContextData, - *varBinds, + *varBinds: ObjectType, **options -) -> "tuple[errind.ErrorIndication, Integer32 | str | int, Integer32 | int, list[ObjectType]]": +) -> "tuple[errind.ErrorIndication, Integer32 | str | int, Integer32 | int, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GETNEXT query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -413,12 +413,9 @@ def __cbFun( ): errorIndication = None # TODO: fix this try: - varBindsUnmade = [ - VB_PROCESSOR.unmakeVarBinds( - snmpEngine.cache, varBindTableRow, lookupMib - ) - for varBindTableRow in varBindTable - ] + varBindsUnmade = VB_PROCESSOR.unmakeVarBinds( + snmpEngine.cache, varBindTable[0], lookupMib + ) except Exception as e: future.set_exception(e) else: @@ -453,7 +450,7 @@ async def bulkCmd( maxRepetitions: int, *varBinds: ObjectType, **options -) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, list[ObjectType]]": +) -> "tuple[errind.ErrorIndication, Integer32 | int, Integer32 | int, tuple[ObjectType, ...]]": r"""Creates a generator to perform SNMP GETBULK query. When iterator gets advanced by :py:mod:`asyncio` main loop, @@ -582,12 +579,9 @@ def __cbFun( ): errorIndication = None # TODO: fix here try: - varBindsUnmade = [ - VB_PROCESSOR.unmakeVarBinds( - snmpEngine.cache, varBindTableRow, lookupMib - ) - for varBindTableRow in varBindTable - ] + varBindsUnmade = VB_PROCESSOR.unmakeVarBinds( + snmpEngine.cache, varBindTable[0], lookupMib + ) except Exception as e: future.set_exception(e) else: @@ -623,7 +617,7 @@ async def walkCmd( varBind: ObjectType, **options ) -> AsyncGenerator[ - "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, list[ObjectType]]", + "tuple[errind.ErrorIndication | None, Integer32 | str | int | None, Integer32 | int | None, tuple[ObjectType, ...]]", None, ]: r"""Creates a generator to perform one or more SNMP GETNEXT queries. @@ -757,7 +751,7 @@ async def walkCmd( else: stopFlag = True - varBind = varBindTable[0][0] + varBind = varBindTable[0] name, val = varBind foundEnding = isinstance(val, Null) or isinstance(val, EndOfMibView) @@ -805,7 +799,7 @@ async def bulkWalkCmd( varBind: ObjectType, **options ) -> AsyncGenerator[ - "tuple[errind.ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, list[ObjectType]]", + "tuple[errind.ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]", None, ]: r"""Creates a generator to perform one or more SNMP GETBULK queries. @@ -920,11 +914,13 @@ async def bulkWalkCmd( maxRows = options.get("maxRows", 0) maxCalls = options.get("maxCalls", 0) - initialVars = [x[0] for x in VB_PROCESSOR.makeVarBinds(snmpEngine.cache, [varBind])] + initialVars = [ + x[0] for x in VB_PROCESSOR.makeVarBinds(snmpEngine.cache, (varBind,)) + ] totalRows = totalCalls = 0 - varBinds = [varBind] + varBinds: "tuple[ObjectType, ...]" = (varBind,) while True: if maxRows and totalRows < maxRows: @@ -954,7 +950,7 @@ async def bulkWalkCmd( errorIndication, errorStatus, errorIndex, - varBindTable[0] and varBinds, + varBindTable and varBinds, ) if errorIndication != errind.requestTimedOut: return @@ -973,7 +969,7 @@ async def bulkWalkCmd( return else: stopFlag = True - varBinds = varBindTable[0] + varBinds = varBindTable for col, varBind in enumerate(varBinds): name, val = varBind diff --git a/pysnmp/hlapi/varbinds.py b/pysnmp/hlapi/varbinds.py index c7eb16c1..6e3cc375 100644 --- a/pysnmp/hlapi/varbinds.py +++ b/pysnmp/hlapi/varbinds.py @@ -4,13 +4,13 @@ # Copyright (c) 2005-2020, Ilya Etingof # License: https://www.pysnmp.com/pysnmp/license.html # -from typing import Any, Dict +from typing import Any, Dict, Tuple from pysnmp.proto.api import v2c from pysnmp.smi import builder, view from pysnmp.smi.rfc1902 import NotificationType, ObjectIdentity, ObjectType -__all__ = ["CommandGeneratorVarBinds", "NotificationOriginatorVarBinds", "isEndOfMib"] +__all__ = ["CommandGeneratorVarBinds", "NotificationOriginatorVarBinds"] def isEndOfMib(var_binds): # noqa: N816 @@ -40,7 +40,9 @@ def getMibViewController(userCache): class CommandGeneratorVarBinds(MibViewControllerManager): - def makeVarBinds(self, userCache: Dict[str, Any], varBinds): + def makeVarBinds( + self, userCache: Dict[str, Any], varBinds: Tuple[ObjectType, ...] + ) -> Tuple[ObjectType, ...]: mibViewController = self.getMibViewController(userCache) resolvedVarBinds = [] @@ -65,15 +67,20 @@ def makeVarBinds(self, userCache: Dict[str, Any], varBinds): varBind.resolveWithMib(mibViewController, ignoreErrors=False) ) - return resolvedVarBinds + return tuple(resolvedVarBinds) - def unmakeVarBinds(self, userCache: Dict[str, Any], varBinds, lookupMib=True): + def unmakeVarBinds( + self, + userCache: Dict[str, Any], + varBinds: Tuple[ObjectType, ...], + lookupMib=True, + ) -> Tuple[ObjectType, ...]: if lookupMib: mibViewController = self.getMibViewController(userCache) - varBinds = [ + varBinds = tuple( ObjectType(ObjectIdentity(x[0]), x[1]).resolveWithMib(mibViewController) for x in varBinds - ] + ) return varBinds diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_slim.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_slim.py new file mode 100644 index 00000000..83192bd9 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_slim.py @@ -0,0 +1,174 @@ +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager +from pysnmp.proto.rfc1905 import errorStatus as pysnmp_errorStatus + + +@pytest.mark.asyncio +async def test_v1_get(): + async with AgentContextManager(): + with Slim(1) as slim: + errorIndication, errorStatus, errorIndex, varBinds = await slim.get( + "public", + "localhost", + AGENT_PORT, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + retries=0, + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 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) + + name = pysnmp_errorStatus.namedValues.getName(errorStatus) + assert name == "noError" + + +@pytest.mark.asyncio +async def test_v1_get_old(): + async with AgentContextManager(): + slim = Slim(1) + 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 errorIndex == 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() + + +@pytest.mark.asyncio +async def test_v1_next(): + async with AgentContextManager(): + with Slim(1) as slim: + errorIndication, errorStatus, errorIndex, varBinds = await slim.next( + "public", + "localhost", # "demo.pysnmp.com", + AGENT_PORT, # 161, + ObjectType( + ObjectIdentity("SNMPv2-MIB", "sysDescr", 0).loadMibs("PYSNMP-MIB") + ), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + assert ( + varBinds[0][1].prettyPrint() == "PYSNMP-MIB::pysnmp" + ) # IMPORTANT: MIB is needed to resolve this name + # assert type(varBinds[0][0][1]).__name__ == "ObjectIdentifier" # TODO: fix this + + +@pytest.mark.asyncio +async def test_v1_set(): + async with AgentContextManager(): + with Slim(1) as 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 isinstance(varBinds[0][1], OctetString) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("num_bulk", [1, 2, 50]) +async def test_v2c_bulk(num_bulk): + async with AgentContextManager(): + with Slim() as slim: + errorIndication, errorStatus, errorIndex, varBinds = await slim.bulk( + "public", + "localhost", + AGENT_PORT, + 0, + num_bulk, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + retries=0, + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == num_bulk + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + if num_bulk > 1: + assert varBinds[1][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0" + if num_bulk > 2: + assert varBinds[2][0].prettyPrint() == "SNMPv2-MIB::sysContact.0" + + +@pytest.mark.asyncio +async def test_v2_get(): + async with AgentContextManager(): + with Slim() as 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 varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + assert isinstance(varBinds[0][1], OctetString) + + +@pytest.mark.asyncio +async def test_v2_next(): + async with AgentContextManager(): + with Slim() as slim: + errorIndication, errorStatus, errorIndex, varBinds = await slim.next( + "public", + "localhost", # "demo.pysnmp.com", + AGENT_PORT, # 161, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + +@pytest.mark.asyncio +async def test_v2_set(): + async with AgentContextManager(): + with Slim() as 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 isinstance(varBinds[0][1], OctetString) diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py new file mode 100644 index 00000000..caf2dc9c --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py @@ -0,0 +1,134 @@ +import asyncio +from datetime import datetime +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from pysnmp.proto.rfc1905 import errorStatus as pysnmp_errorStatus + +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v1_get(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + + iterator = await getCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ("1.3.6.1.2.1.1.1.0", None), + ) + + errorIndication, errorStatus, errorIndex, varBinds = iterator + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "1.3.6.1.2.1.1.1.0" + assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + + name = pysnmp_errorStatus.namedValues.getName(errorStatus) + assert name == "noError" + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v1_get_ipv6(): + async with AgentContextManager(enable_ipv6=True): + snmpDispatcher = SnmpDispatcher() + + iterator = await getCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await Udp6TransportTarget.create(("localhost", AGENT_PORT)), + ("1.3.6.1.2.1.1.1.0", None), + ) + + errorIndication, errorStatus, errorIndex, varBinds = iterator + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "1.3.6.1.2.1.1.1.0" + assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + + name = pysnmp_errorStatus.namedValues.getName(errorStatus) + assert name == "noError" + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +# TODO: +# def test_v1_get_timeout_invalid_target(): +# loop = asyncio.get_event_loop() +# snmpDispatcher = SnmpDispatcher() + +# async def run_get(): +# errorIndication, errorStatus, errorIndex, varBinds = await getCmd( +# snmpDispatcher, +# CommunityData("community_string"), +# await UdpTransportTarget.create(("1.2.3.4", 161), timeout=1, retries=0), +# ObjectType(ObjectIdentity("1.3.6.1.4.1.60069.9.1.0")), +# ) +# for varBind in varBinds: +# print([str(varBind[0]), varBind[1]]) + +# start = datetime.now() +# try: +# loop.run_until_complete(asyncio.wait_for(run_get(), timeout=3)) +# end = datetime.now() +# elapsed_time = (end - start).total_seconds() +# assert elapsed_time >= 1 and elapsed_time <= 3 +# except asyncio.TimeoutError: +# assert False, "Test case timed out" +# finally: +# snmpDispatcher.transportDispatcher.closeDispatcher() + + +# @pytest.mark.asyncio +# async def test_v1_get_timeout_slow_object(): +# async with AgentContextManager(enable_custom_objects=True): +# snmpDispatcher = SnmpDispatcher() + +# async def run_get(): +# errorIndication, errorStatus, errorIndex, varBinds = await getCmd( +# snmpDispatcher, +# CommunityData("public", mpModel=0), +# await UdpTransportTarget.create( +# ("localhost", AGENT_PORT), timeout=1, retries=0 +# ), +# ObjectType(ObjectIdentity("1.3.6.1.4.1.60069.9.1.0")), +# ) +# for varBind in varBinds: +# print([str(varBind[0]), varBind[1]]) + +# start = datetime.now() +# try: +# await asyncio.wait_for(run_get(), timeout=3) +# end = datetime.now() +# elapsed_time = (end - start).total_seconds() +# assert elapsed_time >= 1 and elapsed_time <= 3 +# except asyncio.TimeoutError: +# assert False, "Test case timed out" +# finally: +# snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v1_get_no_access_object(): + async with AgentContextManager(enable_custom_objects=True): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create( + ("localhost", AGENT_PORT), timeout=1, retries=0 + ), + ObjectType(ObjectIdentity("1.3.6.1.4.1.60069.9.3")), + ) + assert errorIndication is None + assert errorStatus.prettyPrint() == "noSuchName" # v1 does not have noAccess + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_next.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_next.py new file mode 100644 index 00000000..be3dee17 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_next.py @@ -0,0 +1,47 @@ +""" +SNMPv1 +++++++ + +Send SNMP GET request using the following options: + + * with SNMPv1, community 'public' + * over IPv4/UDP + * to an Agent at demo.pysnmp.com:161 + * for an instance of SNMPv2-MIB::sysDescr.0 MIB object + * Based on asyncio I/O framework + +Functionally similar to: + +| $ snmpgetnext -v1 -c public demo.pysnmp.com SNMPv2-MIB::sysDescr.0 + +""" # +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from pysnmp.smi import builder, compiler, view +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v1_next(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await nextCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType( + ObjectIdentity("SNMPv2-MIB", "sysDescr", 0).loadMibs("PYSNMP-MIB") + ), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + assert ( + varBinds[0][1].prettyPrint() == "PYSNMP-MIB::pysnmp" + ) # IMPORTANT: MIB is needed to resolve this name + # assert type(varBinds[0][0][1]).__name__ == "ObjectIdentifier" # TODO: fix this + + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_set.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_set.py new file mode 100644 index 00000000..ecc98006 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_set.py @@ -0,0 +1,97 @@ +import pytest +from pysnmp.entity.engine import SnmpEngine +from pysnmp.hlapi.v1arch.asyncio.cmdgen import setCmd, walkCmd +from pysnmp.hlapi.v1arch.asyncio.dispatch import SnmpDispatcher +from pysnmp.hlapi.v1arch.asyncio.transport import UdpTransportTarget +from pysnmp.hlapi.v1arch.asyncio.auth import CommunityData +from pysnmp.proto.rfc1902 import Integer, OctetString +from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v1_set(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + + iterator = await setCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai"), + ) + + errorIndication, errorStatus, errorIndex, varBinds = iterator + + 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 isinstance(varBinds[0][1], OctetString) + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v1_set_table_creation(): + async with AgentContextManager(enable_table_creation=True): + snmpDispatcher = SnmpDispatcher() + + # Perform a SNMP walk to get all object counts + objects = walkCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("1.3.6")), + ) + + objects_list = [item async for item in objects] + assert len(objects_list) > 0 + + object_counts = len(objects_list) + + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType( + ObjectIdentity("1.3.6.6.1.5.2.97.98.99"), OctetString("My value") + ), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.2.97.98.99" + assert varBinds[0][1].prettyPrint() == "My value" + # TODO: assert isinstance(varBinds[0][1], OctetString) + + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("1.3.6.6.1.5.4.97.98.99"), Integer(4)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.4.97.98.99" + assert varBinds[0][1].prettyPrint() == "1" + # assert isinstance(varBinds[0][1], Integer) + + # Perform a SNMP walk to get all object counts + objects = walkCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("1.3.6")), + ) + + objects_list = [item async for item in objects] + assert len(objects_list) > 0 + + assert len(objects_list) == object_counts + 4 + + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_walk.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_walk.py new file mode 100644 index 00000000..23cdf483 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_walk.py @@ -0,0 +1,68 @@ +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v1_walk(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + objects = walkCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [item async for item in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0" + + assert len(objects_list) == 267 + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v1_walk_subtree(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + objects = walkCmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("SNMPv2-MIB", "system")), + lexicographicMode=False, + ) + + objects_list = [item async for item in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + assert len(objects_list) == 8 + + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2_walk.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2_walk.py new file mode 100644 index 00000000..0e9422e5 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2_walk.py @@ -0,0 +1,68 @@ +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v2_walk(): # some agents have different v2 GET NEXT behavior + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + objects = walkCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [item async for item in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0" + + assert len(objects_list) == 267 + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2_walk_subtree(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + objects = walkCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("SNMPv2-MIB", "system")), + lexicographicMode=False, + ) + + objects_list = [item async for item in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + assert len(objects_list) == 8 + + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulk.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulk.py new file mode 100644 index 00000000..6512cb01 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulk.py @@ -0,0 +1,58 @@ +import pytest + +from pysnmp.hlapi.v1arch.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +@pytest.mark.parametrize("num_bulk", [1, 2, 50]) +async def test_v2c_bulk(num_bulk): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await bulkCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + 0, + num_bulk, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + retries=0, + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == num_bulk + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + if num_bulk > 1: + assert varBinds[1][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0" + if num_bulk > 2: + assert varBinds[2][0].prettyPrint() == "SNMPv2-MIB::sysContact.0" + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_bulk_multiple_input(): + mib_objects = ( + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysContact")), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysORIndex")), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysORDescr")), + ) + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await bulkCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 1, + 2, + *mib_objects, + retries=0 + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 5 + + +# snmpbulkget -v2c -c public -C n1 -C r2 localhost 1.3.6.1.2.1.1.4 1.3.6.1.2.1.1.9.1.1 1.3.6.1.2.1.1.9.1.3 diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py new file mode 100644 index 00000000..7250aa00 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py @@ -0,0 +1,311 @@ +import pytest + +from pysnmp.hlapi.v1arch.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +@pytest.mark.parametrize("max_repetitions", [1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30]) +async def test_v2c_get_table_bulk(max_repetitions): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" + + assert len(objects_list) == 60 / max_repetitions + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_4_subtree(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + index = 0 + async for errorIndication, errorStatus, errorIndex, varBinds in bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + 0, + 4, + ObjectType(ObjectIdentity("SNMPv2-MIB", "snmp")), + lexicographicMode=False, + ): + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 4 + if index == 0: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpInPkts.0" + + if index == 1: + assert ( + varBinds[0][0].prettyPrint() + == "SNMPv2-MIB::snmpInBadCommunityUses.0" + ) + + if index == 26: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpSilentDrops.0" + + if index == 27: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpProxyDrops.0" + + index += 1 + + assert index == 7 + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_1_subtree(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + index = 0 + async for errorIndication, errorStatus, errorIndex, varBinds in bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + 0, + 1, + ObjectType(ObjectIdentity("SNMPv2-MIB", "snmp")), + lexicographicMode=False, + ): + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + if index == 0: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpInPkts.0" + + if index == 1: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpOutPkts.0" + + if index == 26: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpSilentDrops.0" + + if index == 27: + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpProxyDrops.0" + + index += 1 + + assert index == 28 + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_7(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + max_repetitions = 7 + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" + + assert len(objects_list) == 9 + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_8(): + snmpDispatcher = SnmpDispatcher() + max_repetitions = 8 + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" + + assert len(objects_list) == 8 + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_31(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + max_repetitions = 31 + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[1] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 60 - max_repetitions + # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" + + assert len(objects_list) == 2 + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_60(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + max_repetitions = 60 + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + assert len(objects_list) == 1 + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_5_subtree(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + max_repetitions = 5 + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "system")), + lexicographicMode=False, + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[3] + assert len(varBinds) == 1 + + assert len(objects_list) == 4 + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2c_get_table_bulk_0_6_subtree(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + max_repetitions = 6 + objects = bulkWalkCmd( + snmpDispatcher, + CommunityData("public"), + # await UdpTransportTarget.create(("localhost", AGENT_PORT)), + await UdpTransportTarget.create(("demo.pysnmp.com", 161)), + 0, + max_repetitions, + ObjectType(ObjectIdentity("SNMPv2-MIB", "system")), + lexicographicMode=False, + ) + + objects_list = [obj async for obj in objects] + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[0] + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == max_repetitions + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" + + errorIndication, errorStatus, errorIndex, varBinds = objects_list[2] + assert len(varBinds) == 4 + + assert len(objects_list) == 3 + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_get.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_get.py new file mode 100644 index 00000000..451226af --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_get.py @@ -0,0 +1,62 @@ +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v2_get(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("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 varBinds[0][1].prettyPrint().startswith("PySNMP engine version") + assert isinstance(varBinds[0][1], OctetString) + + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2_get_no_access_object(): + async with AgentContextManager(enable_custom_objects=True): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create( + ("localhost", AGENT_PORT), timeout=1, retries=0 + ), + ObjectType(ObjectIdentity("1.3.6.1.4.1.60069.9.3")), + ) + + assert errorIndication is None + assert errorStatus.prettyPrint() == "noAccess" # v2c and v3 use noAccess + snmpDispatcher.transportDispatcher.closeDispatcher() + + +@pytest.mark.asyncio +async def test_v2_get_legacy_object(): + async with AgentContextManager(enable_custom_objects=True): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create( + ("localhost", AGENT_PORT), timeout=1, retries=0 + ), + ObjectType(ObjectIdentity("1.3.6.1.4.1.60069.9.4")), + ) + + assert errorIndication is None + assert ( + errorStatus.prettyPrint() == "noAccess" + ) # PySMI <1.3.0 generates such objects + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_next.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_next.py new file mode 100644 index 00000000..8e01a1a8 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_next.py @@ -0,0 +1,41 @@ +""" +SNMPv1 +++++++ + +Send SNMP GET request using the following options: + + * with SNMPv1, community 'public' + * over IPv4/UDP + * to an Agent at demo.pysnmp.com:161 + * for an instance of SNMPv2-MIB::sysDescr.0 MIB object + * Based on asyncio I/O framework + +Functionally similar to: + +| $ snmpgetnext -v1 -c public demo.pysnmp.com SNMPv2-MIB::sysDescr.0 + +""" # +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v2_next(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await nextCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_set.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_set.py new file mode 100644 index 00000000..b65cfa75 --- /dev/null +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_set.py @@ -0,0 +1,26 @@ +import pytest +from pysnmp.hlapi.v1arch.asyncio import * +from pysnmp.proto.rfc1902 import OctetString +from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from tests.agent_context import AGENT_PORT, AgentContextManager + + +@pytest.mark.asyncio +async def test_v2_set(): + async with AgentContextManager(): + snmpDispatcher = SnmpDispatcher() + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpDispatcher, + CommunityData("public"), + await UdpTransportTarget.create(("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 isinstance(varBinds[0][1], OctetString) + + snmpDispatcher.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_get.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_get.py index c99287bb..85f3173c 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_get.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_get.py @@ -1,7 +1,7 @@ import asyncio from datetime import datetime import pytest -from pysnmp import debug + from pysnmp.hlapi.v3arch.asyncio import * from pysnmp.proto.rfc1905 import errorStatus as pysnmp_errorStatus @@ -10,52 +10,6 @@ @pytest.mark.asyncio async def test_v1_get(): - async with AgentContextManager(): - with Slim(1) as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.get( - "public", - "localhost", - AGENT_PORT, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - retries=0, - ) - - assert errorIndication is None - assert errorStatus == 0 - assert errorIndex == 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) - - name = pysnmp_errorStatus.namedValues.getName(errorStatus) - assert name == "noError" - - -@pytest.mark.asyncio -async def test_v1_get_old(): - async with AgentContextManager(): - slim = Slim(1) - 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 errorIndex == 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() - - -@pytest.mark.asyncio -async def test_v1_get_raw(): async with AgentContextManager(): snmpEngine = SnmpEngine() errorIndication, errorStatus, errorIndex, varBinds = await getCmd( diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_next.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_next.py index 22ce3dad..c0f9b108 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_next.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_next.py @@ -24,22 +24,25 @@ @pytest.mark.asyncio async def test_v1_next(): async with AgentContextManager(): - with Slim(1) as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.next( - "public", - "localhost", # "demo.pysnmp.com", - AGENT_PORT, # 161, - ObjectType( - ObjectIdentity("SNMPv2-MIB", "sysDescr", 0).loadMibs("PYSNMP-MIB") - ), - ) - - assert errorIndication is None - assert errorStatus == 0 - assert errorIndex == 0 - assert len(varBinds) == 1 - assert varBinds[0][0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" - assert ( - varBinds[0][0][1].prettyPrint() == "PYSNMP-MIB::pysnmp" - ) # IMPORTANT: MIB is needed to resolve this name - # assert type(varBinds[0][0][1]).__name__ == "ObjectIdentifier" # TODO: fix this + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await nextCmd( + snmpEngine, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + ObjectType( + ObjectIdentity("SNMPv2-MIB", "sysDescr", 0).loadMibs("PYSNMP-MIB") + ), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + assert ( + varBinds[0][1].prettyPrint() == "PYSNMP-MIB::pysnmp" + ) # IMPORTANT: MIB is needed to resolve this name + # assert type(varBinds[0][0][1]).__name__ == "ObjectIdentifier" # TODO: fix this + + snmpEngine.closeDispatcher() diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_set.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_set.py index 8cf7cb9f..3675078c 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_set.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v1_set.py @@ -1,94 +1,91 @@ import pytest -from pysnmp.entity.engine import SnmpEngine -from pysnmp.hlapi.v3arch.asyncio.cmdgen import walkCmd -from pysnmp.hlapi.v3arch.asyncio.slim import Slim -from pysnmp.hlapi.v3arch.asyncio.transport import UdpTransportTarget -from pysnmp.hlapi.v3arch.asyncio.auth import CommunityData -from pysnmp.hlapi.v3arch.asyncio.context import ContextData -from pysnmp.proto.rfc1902 import Integer, OctetString -from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType +from pysnmp.hlapi.v3arch.asyncio import * from tests.agent_context import AGENT_PORT, AgentContextManager @pytest.mark.asyncio async def test_v1_set(): async with AgentContextManager(): - with Slim(1) as 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 isinstance(varBinds[0][1], OctetString) + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpEngine, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + 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 isinstance(varBinds[0][1], OctetString) + + snmpEngine.closeDispatcher() @pytest.mark.asyncio async def test_v1_set_table_creation(): async with AgentContextManager(enable_table_creation=True): - with Slim(1) as slim: - snmpEngine = SnmpEngine() - - # Perform a SNMP walk to get all object counts - objects = walkCmd( - snmpEngine, - CommunityData("public", mpModel=0), - await UdpTransportTarget.create(("localhost", AGENT_PORT)), - ContextData(), - ObjectType(ObjectIdentity("1.3.6")), - ) - - objects_list = [item async for item in objects] - assert len(objects_list) > 0 - - object_counts = len(objects_list) - - errorIndication, errorStatus, errorIndex, varBinds = await slim.set( - "public", - "localhost", - AGENT_PORT, - ObjectType( - ObjectIdentity("1.3.6.6.1.5.2.97.98.99"), OctetString("My value") - ), - ) - - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.2.97.98.99" - assert varBinds[0][1].prettyPrint() == "My value" - assert isinstance(varBinds[0][1], OctetString) - - errorIndication, errorStatus, errorIndex, varBinds = await slim.set( - "public", - "localhost", - AGENT_PORT, - ObjectType(ObjectIdentity("1.3.6.6.1.5.4.97.98.99"), Integer(4)), - ) - - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 1 - assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.4.97.98.99" - assert varBinds[0][1].prettyPrint() == "1" - # assert isinstance(varBinds[0][1], Integer) - - # Perform a SNMP walk to get all object counts - objects = walkCmd( - snmpEngine, - CommunityData("public", mpModel=0), - await UdpTransportTarget.create(("localhost", AGENT_PORT)), - ContextData(), - ObjectType(ObjectIdentity("1.3.6")), - ) - - objects_list = [item async for item in objects] - assert len(objects_list) > 0 - - assert len(objects_list) == object_counts + 4 + snmpEngine = SnmpEngine() + + # Perform a SNMP walk to get all object counts + objects = walkCmd( + snmpEngine, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + ObjectType(ObjectIdentity("1.3.6")), + ) + + objects_list = [item async for item in objects] + assert len(objects_list) > 0 + + object_counts = len(objects_list) + + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpEngine, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + ObjectType( + ObjectIdentity("1.3.6.6.1.5.2.97.98.99"), OctetString("My value") + ), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.2.97.98.99" + assert varBinds[0][1].prettyPrint() == "My value" + assert isinstance(varBinds[0][1], OctetString) + + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpEngine, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + ObjectType(ObjectIdentity("1.3.6.6.1.5.4.97.98.99"), Integer(4)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.4.97.98.99" + assert varBinds[0][1].prettyPrint() == "1" + # assert isinstance(varBinds[0][1], Integer) + + # Perform a SNMP walk to get all object counts + objects = walkCmd( + snmpEngine, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + ObjectType(ObjectIdentity("1.3.6")), + ) + + objects_list = [item async for item in objects] + assert len(objects_list) > 0 + + assert len(objects_list) == object_counts + 4 diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulk.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulk.py index 1122ce80..3da2d163 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulk.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulk.py @@ -8,25 +8,28 @@ @pytest.mark.parametrize("num_bulk", [1, 2, 50]) async def test_v2c_bulk(num_bulk): async with AgentContextManager(): - with Slim() as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.bulk( - "public", - "localhost", - AGENT_PORT, - 0, - num_bulk, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - retries=0, - ) - - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == num_bulk - assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" - if num_bulk > 1: - assert varBinds[1][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0" - if num_bulk > 2: - assert varBinds[2][0].prettyPrint() == "SNMPv2-MIB::sysContact.0" + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await bulkCmd( + snmpEngine, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + 0, + num_bulk, + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + retries=0, + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == num_bulk + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + if num_bulk > 1: + assert varBinds[1][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0" + if num_bulk > 2: + assert varBinds[2][0].prettyPrint() == "SNMPv2-MIB::sysContact.0" + + snmpEngine.closeDispatcher() @pytest.mark.asyncio @@ -37,14 +40,21 @@ async def test_v2c_bulk_multiple_input(): ObjectType(ObjectIdentity("SNMPv2-MIB", "sysORDescr")), ] async with AgentContextManager(): - with Slim() as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.bulk( - "public", "demo.pysnmp.com", 161, 1, 2, *mib_objects, retries=0 - ) - - assert errorIndication is None - assert errorStatus == 0 - assert len(varBinds) == 5 + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await bulkCmd( + snmpEngine, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + 1, + 2, + *mib_objects, + retries=0 + ) + + assert errorIndication is None + assert errorStatus == 0 + assert len(varBinds) == 5 # snmpbulkget -v2c -c public -C n1 -C r2 localhost 1.3.6.1.2.1.1.4 1.3.6.1.2.1.1.9.1.1 1.3.6.1.2.1.1.9.1.3 diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_get.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_get.py index b5ca6e7b..31127f1d 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_get.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_get.py @@ -1,30 +1,28 @@ -from datetime import datetime import pytest -from pysnmp.hlapi.v3arch.asyncio.slim import Slim from pysnmp.hlapi.v3arch.asyncio import * -from pysnmp.proto import errind from tests.agent_context import AGENT_PORT, AgentContextManager -import asyncio - @pytest.mark.asyncio async def test_v2_get(): async with AgentContextManager(): - with Slim() as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.get( - "public", - "localhost", - AGENT_PORT, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await getCmd( + snmpEngine, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + 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") - 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.closeDispatcher() @pytest.mark.asyncio diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_next.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_next.py index 69fa6cbf..788b3afc 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_next.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_next.py @@ -16,7 +16,7 @@ """ # import pytest -from pysnmp.hlapi.v3arch.asyncio.slim import Slim +from pysnmp.hlapi.v3arch.asyncio import * from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType from tests.agent_context import AGENT_PORT, AgentContextManager @@ -24,16 +24,19 @@ @pytest.mark.asyncio async def test_v2_next(): async with AgentContextManager(): - with Slim() as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.next( - "public", - "localhost", # "demo.pysnmp.com", - AGENT_PORT, # 161, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), - ) - - assert errorIndication is None - assert errorStatus == 0 - assert errorIndex == 0 - assert len(varBinds) == 1 - assert varBinds[0][0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await nextCmd( + snmpEngine, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), + ) + + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" + + snmpEngine.closeDispatcher() diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_set.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_set.py index 6abecd52..ae7a27f8 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_set.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_set.py @@ -1,5 +1,5 @@ import pytest -from pysnmp.hlapi.v3arch.asyncio.slim import Slim +from pysnmp.hlapi.v3arch.asyncio import * from pysnmp.proto.rfc1902 import OctetString from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType from tests.agent_context import AGENT_PORT, AgentContextManager @@ -8,17 +8,20 @@ @pytest.mark.asyncio async def test_v2_set(): async with AgentContextManager(): - with Slim() as slim: - errorIndication, errorStatus, errorIndex, varBinds = await slim.set( - "public", - "localhost", - AGENT_PORT, - ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai"), - ) + snmpEngine = SnmpEngine() + errorIndication, errorStatus, errorIndex, varBinds = await setCmd( + snmpEngine, + CommunityData("public"), + await UdpTransportTarget.create(("localhost", AGENT_PORT)), + ContextData(), + 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 isinstance(varBinds[0][1], OctetString) + 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 isinstance(varBinds[0][1], OctetString) + + snmpEngine.closeDispatcher()