Improved manager context.

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

@@ -4,19 +4,17 @@
 
 from pysnmp.hlapi.v3arch.asyncio import *
 from pysnmp.proto.api import v2c
-import tests.manager_context
+from tests.manager_context import MANAGER_PORT, ManagerContextManager
 
 
 @pytest.mark.asyncio
 async def test_send_trap_enterprise_specific():
-    async with tests.manager_context.ManagerContextManager():
+    async with ManagerContextManager() as (_, message_count):
         snmpEngine = SnmpEngine()
         errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
             snmpEngine,
             CommunityData("public", mpModel=0),
-            await UdpTransportTarget.create(
-                ("localhost", tests.manager_context.MANAGER_PORT)
-            ),
+            await UdpTransportTarget.create(("localhost", MANAGER_PORT)),
             ContextData(),
             "trap",
             NotificationType(
@@ -34,19 +32,17 @@ async def test_send_trap_enterprise_specific():
 
         snmpEngine.closeDispatcher()
         await asyncio.sleep(1)
-        assert tests.manager_context.message_count == 1
+        assert message_count == [1]
 
 
 @pytest.mark.asyncio
 async def test_send_trap_generic():
-    async with tests.manager_context.ManagerContextManager():
+    async with ManagerContextManager() as (_, message_count):
         snmpEngine = SnmpEngine()
         errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
             snmpEngine,
             CommunityData("public", mpModel=0),
-            await UdpTransportTarget.create(
-                ("localhost", tests.manager_context.MANAGER_PORT)
-            ),
+            await UdpTransportTarget.create(("localhost", MANAGER_PORT)),
             ContextData(),
             "trap",
             NotificationType(ObjectIdentity("1.3.6.1.6.3.1.1.5.2"))
@@ -62,4 +58,35 @@ async def test_send_trap_generic():
 
         snmpEngine.closeDispatcher()
         await asyncio.sleep(1)
-        assert tests.manager_context.message_count == 1
+        assert message_count == [1]
+
+
+@pytest.mark.asyncio
+async def test_send_trap_custom_mib():
+    async with ManagerContextManager() as (_, message_count):
+        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),
+            await UdpTransportTarget.create(("localhost", MANAGER_PORT)),
+            ContextData(),
+            "trap",
+            NotificationType(
+                ObjectIdentity("NET-SNMP-EXAMPLES-MIB", "netSnmpExampleNotification")
+            ).addVarBinds(
+                ObjectType(
+                    ObjectIdentity(
+                        "NET-SNMP-EXAMPLES-MIB", "netSnmpExampleHeartbeatRate"
+                    ),
+                    1,
+                )
+            ),
+        )
+
+        snmpEngine.closeDispatcher()
+        await asyncio.sleep(1)
+        assert message_count == [1]

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

@@ -3,25 +3,42 @@
 
 
 from pysnmp.hlapi.v3arch.asyncio import *
-import tests.manager_context
+from tests.manager_context import MANAGER_PORT, ManagerContextManager
 
 
 @pytest.mark.asyncio
 async def test_send_v3_trap_notification():
-    async with tests.manager_context.ManagerContextManager():
+    async with ManagerContextManager() as (_, message_count):
         # snmptrap -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 -e 8000000001020304 localhost:MANAGER_PORT 0 1.3.6.1.6.3.1.1.5.1 1.3.6.1.2.1.1.1.0 s "my system"
         snmpEngine = SnmpEngine(OctetString(hexValue="8000000001020304"))
         errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
             snmpEngine,
             UsmUserData("usr-md5-des", "authkey1", "privkey1"),
-            await UdpTransportTarget.create(
-                ("localhost", tests.manager_context.MANAGER_PORT)
-            ),
+            await UdpTransportTarget.create(("localhost", MANAGER_PORT)),
             ContextData(),
             "trap",
             NotificationType(ObjectIdentity("IF-MIB", "linkDown")),
         )
 
         snmpEngine.closeDispatcher()
         await asyncio.sleep(1)
-        assert tests.manager_context.message_count == 1
+        assert message_count == [1]
+
+
+@pytest.mark.asyncio
+async def test_send_v3_trap_notification_invalid_user():
+    async with ManagerContextManager() as (_, message_count):
+        # snmptrap -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 -e 8000000001020304 localhost:MANAGER_PORT 0 1.3.6.1.6.3.1.1.5.1 1.3.6.1.2.1.1.1.0 s "my system"
+        snmpEngine = SnmpEngine(OctetString(hexValue="8000000001020304"))
+        errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
+            snmpEngine,
+            UsmUserData("usr-md5-des1", "authkey1", "privkey1"),
+            await UdpTransportTarget.create(("localhost", MANAGER_PORT)),
+            ContextData(),
+            "trap",
+            NotificationType(ObjectIdentity("IF-MIB", "linkDown")),
+        )
+
+        snmpEngine.closeDispatcher()
+        await asyncio.sleep(1)
+        assert message_count == [0]

tests/manager_context.py

@@ -1,6 +1,6 @@
 # manager_context.py
 import asyncio
-from typing import Tuple
+from typing import List, Tuple
 
 
 from pysnmp.entity.rfc3413 import ntfrcv
@@ -13,11 +13,10 @@
 # privileged port and requires root access
 MANAGER_PORT = 1622
 
-# Global variable to track message count
-message_count = 0
 
-
-async def start_manager() -> Tuple[SnmpEngine, ntfrcv.NotificationReceiver]:
+async def start_manager(
+    message_count: List[int],
+) -> Tuple[SnmpEngine, ntfrcv.NotificationReceiver]:
     # Create SNMP engine
     snmpEngine = engine.SnmpEngine()
 
@@ -64,8 +63,7 @@ async def start_manager() -> Tuple[SnmpEngine, ntfrcv.NotificationReceiver]:
     def cbFun(
         snmpEngine, stateReference, contextEngineId, contextName, varBinds, cbCtx
     ):
-        global message_count
-        message_count += 1
+        message_count[0] += 1
         print(
             'Notification from ContextEngineId "{}", ContextName "{}"'.format(
                 contextEngineId.prettyPrint(), contextName.prettyPrint()
@@ -105,13 +103,12 @@ class ManagerContextManager:
 
     manager: SnmpEngine
     receiver: ntfrcv.NotificationReceiver
+    message_count: List[int]
 
     async def __aenter__(self):
-        global message_count
-        message_count = 0
-
-        self.manager, self.receiver = await start_manager()
-        return self.manager
+        self.message_count = [0]
+        self.manager, self.receiver = await start_manager(self.message_count)
+        return self.manager, self.message_count
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         self.receiver.close(self.manager)