Added a few test cases.

examples/hlapi/asyncio/agent/ntforg/multiple-notifications-at-once.py

@@ -20,7 +20,6 @@
 
 | $ snmptrap -v2c -c public demo.pysnmp.com 12345 1.3.6.1.6.3.1.1.5.2
 | $ snmpinform -v2c -c public demo.pysnmp.com 12345 1.3.6.1.6.3.1.1.5.2
-| $ snmptrap -v2c -c public demo.pysnmp.com 12345 1.3.6.1.6.3.1.1.5.2
 
 """#
 import asyncio

tests/hlapi/asyncio/agent/ntforg/test_default-v1-trap.py

@@ -19,3 +19,4 @@ async def test_send_notification():
     assert errorStatus == 0
     assert errorIndex == 0
     assert varBinds == []
+    snmpEngine.transportDispatcher.closeDispatcher()

tests/hlapi/asyncio/agent/ntforg/test_v3-inform.py

@@ -6,8 +6,9 @@
 
 @pytest.mark.asyncio
 async def test_send_v3_inform_notification():
+    snmpEngine = SnmpEngine()
     errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
-        SnmpEngine(),
+        snmpEngine,
         UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
         UdpTransportTarget(('demo.pysnmp.com', 162)),
         ContextData(),
@@ -22,4 +23,5 @@ async def test_send_v3_inform_notification():
     assert isinstance(errorIndication, RequestTimedOut)
     assert errorStatus == 0
     assert errorIndex == 0
-    assert varBinds == []
+    assert varBinds == []
+    snmpEngine.transportDispatcher.closeDispatcher()

tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py

@@ -0,0 +1,21 @@
+import pytest
+from pysnmp.hlapi.asyncio import *
+
+@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@'))
+    )
+
+    assert errorIndication is None
+    assert errorStatus == 0
+    assert len(varBinds) == 1
+    assert varBinds[0][0].prettyPrint() == 'IF-MIB::ifInOctets.1'
+
+    snmpEngine.transportDispatcher.closeDispatcher()

tests/hlapi/asyncio/manager/cmdgen/test_send_trap.py

@@ -0,0 +1,24 @@
+import pytest
+from pysnmp.hlapi.asyncio import *
+
+@pytest.mark.asyncio
+async def test_send_trap():
+    snmpEngine = SnmpEngine()
+    mibBuilder = snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder
+    sysUpTime, = mibBuilder.importSymbols('__SNMPv2-MIB', 'sysUpTime')
+    sysUpTime.syntax = TimeTicks(12345)
+
+    errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
+        snmpEngine,
+        CommunityData('public', mpModel=0),
+        UdpTransportTarget(('demo.pysnmp.com', 162)),
+        ContextData(),
+        "trap",
+        NotificationType(ObjectIdentity('NET-SNMP-EXAMPLES-MIB', 'netSnmpExampleNotification')).addVarBinds(ObjectType(ObjectIdentity('NET-SNMP-EXAMPLES-MIB','netSnmpExampleHeartbeatRate'), 1))
+    )
+
+    assert errorIndication is None
+    assert errorStatus == 0
+    assert len(varBinds) == 0
+
+    snmpEngine.transportDispatcher.closeDispatcher()

tests/hlapi/asyncio/manager/cmdgen/test_usm_sha_aes128.py

@@ -0,0 +1,29 @@
+import pytest
+from pysnmp.hlapi.asyncio import *
+
+
+@pytest.mark.asyncio
+async def test_usm_sha_aes128():
+    snmpEngine = SnmpEngine()
+    authData = UsmUserData(
+        "usr-sha-aes",
+        "authkey1",
+        "privkey1",
+        authProtocol=usmHMACSHAAuthProtocol,
+        privProtocol=usmAesCfb128Protocol,
+    )
+    errorIndication, errorStatus, errorIndex, varBinds = await getCmd(
+        snmpEngine,
+        authData,
+        UdpTransportTarget(("demo.pysnmp.com", 161)),
+        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"
+    isinstance(varBinds[0][1], OctetString)
+
+    snmpEngine.transportDispatcher.closeDispatcher()

tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py

@@ -0,0 +1,21 @@
+import pytest
+from pysnmp.hlapi.asyncio import *
+
+@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))
+    )
+
+    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)
+
+    snmpEngine.transportDispatcher.closeDispatcher()

tests/hlapi/asyncio/manager/cmdgen/test_v1-next.py → tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py

@@ -33,4 +33,6 @@ async def test_v1_next():
     assert errorIndication is None
     assert errorStatus == 0
     assert errorIndex == 0
-    assert len(varBinds) > 0
+    assert len(varBinds) > 0
+
+    slim.close()

tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py

@@ -0,0 +1,21 @@
+import pytest
+from pysnmp.hlapi.asyncio.slim import Slim
+from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType
+
+@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")
+    )
+
+    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()

tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py

@@ -0,0 +1,21 @@
+import pytest
+from pysnmp.hlapi.asyncio.slim import Slim
+from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType
+
+@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)),
+    )
+
+    assert errorIndication is None
+    assert errorStatus == 0
+    assert len(varBinds) > 0
+
+    slim.close()

tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py

@@ -0,0 +1,21 @@
+import pytest
+from pysnmp.hlapi.asyncio.slim import Slim
+from pysnmp.hlapi.asyncio import *
+
+@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)),
+    )
+
+    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)
+
+    slim.close()

tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py

@@ -0,0 +1,38 @@
+"""
+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 asyncio
+import pytest
+from pysnmp.hlapi.asyncio.slim import Slim
+from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType
+
+@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()

tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py

@@ -0,0 +1,21 @@
+import pytest
+from pysnmp.hlapi.asyncio.slim import Slim
+from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType
+
+@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")
+    )
+
+    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()