From 0cfe715d2c3063373df7f4403f4a575e530c4ba4 Mon Sep 17 00:00:00 2001 From: Lex Li Date: Sun, 17 Sep 2023 02:37:53 -0400 Subject: [PATCH] Added a few test cases. --- .../ntforg/multiple-notifications-at-once.py | 1 - .../agent/ntforg/test_default-v1-trap.py | 1 + .../asyncio/agent/ntforg/test_v3-inform.py | 6 ++- .../test_custom_asn1_mib_search_path.py | 21 ++++++++++ .../asyncio/manager/cmdgen/test_send_trap.py | 24 ++++++++++++ .../manager/cmdgen/test_usm_sha_aes128.py | 29 ++++++++++++++ .../asyncio/manager/cmdgen/test_v1_get.py | 21 ++++++++++ .../{test_v1-next.py => test_v1_next.py} | 4 +- .../asyncio/manager/cmdgen/test_v1_set.py | 21 ++++++++++ .../asyncio/manager/cmdgen/test_v2c_bulk.py | 21 ++++++++++ .../asyncio/manager/cmdgen/test_v2c_get.py | 21 ++++++++++ .../asyncio/manager/cmdgen/test_v2c_next.py | 38 +++++++++++++++++++ .../asyncio/manager/cmdgen/test_v2c_set.py | 21 ++++++++++ 13 files changed, 225 insertions(+), 4 deletions(-) create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_send_trap.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_usm_sha_aes128.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py rename tests/hlapi/asyncio/manager/cmdgen/{test_v1-next.py => test_v1_next.py} (94%) create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py create mode 100644 tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py diff --git a/examples/hlapi/asyncio/agent/ntforg/multiple-notifications-at-once.py b/examples/hlapi/asyncio/agent/ntforg/multiple-notifications-at-once.py index 4b6cc1022..df2898982 100644 --- a/examples/hlapi/asyncio/agent/ntforg/multiple-notifications-at-once.py +++ b/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 diff --git a/tests/hlapi/asyncio/agent/ntforg/test_default-v1-trap.py b/tests/hlapi/asyncio/agent/ntforg/test_default-v1-trap.py index 2b0629d60..1d3126fe8 100644 --- a/tests/hlapi/asyncio/agent/ntforg/test_default-v1-trap.py +++ b/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() diff --git a/tests/hlapi/asyncio/agent/ntforg/test_v3-inform.py b/tests/hlapi/asyncio/agent/ntforg/test_v3-inform.py index 9f0eb3e2d..952fc7c19 100644 --- a/tests/hlapi/asyncio/agent/ntforg/test_v3-inform.py +++ b/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 == [] \ No newline at end of file + assert varBinds == [] + snmpEngine.transportDispatcher.closeDispatcher() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py b/tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py new file mode 100644 index 000000000..f515514c3 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_send_trap.py b/tests/hlapi/asyncio/manager/cmdgen/test_send_trap.py new file mode 100644 index 000000000..6ca409cdd --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_usm_sha_aes128.py b/tests/hlapi/asyncio/manager/cmdgen/test_usm_sha_aes128.py new file mode 100644 index 000000000..d10447093 --- /dev/null +++ b/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() diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py b/tests/hlapi/asyncio/manager/cmdgen/test_v1_get.py new file mode 100644 index 000000000..6415ef5e6 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v1-next.py b/tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py similarity index 94% rename from tests/hlapi/asyncio/manager/cmdgen/test_v1-next.py rename to tests/hlapi/asyncio/manager/cmdgen/test_v1_next.py index d611e7227..18fecf2ef 100644 --- a/tests/hlapi/asyncio/manager/cmdgen/test_v1-next.py +++ b/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 \ No newline at end of file + assert len(varBinds) > 0 + + slim.close() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py b/tests/hlapi/asyncio/manager/cmdgen/test_v1_set.py new file mode 100644 index 000000000..eafc457d9 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py new file mode 100644 index 000000000..707881df8 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_get.py new file mode 100644 index 000000000..df523f97c --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_next.py new file mode 100644 index 000000000..f34356c4f --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_set.py new file mode 100644 index 000000000..2e71ddc4e --- /dev/null +++ b/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() \ No newline at end of file