Fully implemented TRAP test cases.

lextudio · Sep 6, 2024 · 37a5878 · 37a5878
1 parent ce8faf0
commit 37a5878
Show file tree

Hide file tree

Showing 3 changed files with 38 additions and 20 deletions.
diff --git a/tests/hlapi/asyncio/agent/ntforg/test_default-v1-trap.py b/tests/hlapi/asyncio/agent/ntforg/test_default-v1-trap.py
@@ -1,18 +1,22 @@
+import asyncio
 import pytest
+
+
 from pysnmp.hlapi.v3arch.asyncio import *
 from pysnmp.proto.api import v2c
-from tests.manager_context import MANAGER_PORT, ManagerContextManager
+import tests.manager_context
 
 
 @pytest.mark.asyncio
 async def test_send_trap_enterprise_specific():
-    async with ManagerContextManager():
+    async with tests.manager_context.ManagerContextManager():
         snmpEngine = SnmpEngine()
         errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
             snmpEngine,
             CommunityData("public", mpModel=0),
-            # await UdpTransportTarget.create(("localhost", MANAGER_PORT)), # TODO: Fix this
-            await UdpTransportTarget.create(("demo.pysnmp.com", 162)),
+            await UdpTransportTarget.create(
+                ("localhost", tests.manager_context.MANAGER_PORT)
+            ),
             ContextData(),
             "trap",
             NotificationType(
@@ -27,22 +31,22 @@ async def test_send_trap_enterprise_specific():
                 ),
             ),
         )
-        assert errorIndication is None
-        assert errorStatus == 0
-        assert errorIndex == 0
-        assert varBinds == []
+
         snmpEngine.closeDispatcher()
+        await asyncio.sleep(1)
+        assert tests.manager_context.message_count == 1
 
 
 @pytest.mark.asyncio
 async def test_send_trap_generic():
-    async with ManagerContextManager():
+    async with tests.manager_context.ManagerContextManager():
         snmpEngine = SnmpEngine()
         errorIndication, errorStatus, errorIndex, varBinds = await sendNotification(
             snmpEngine,
             CommunityData("public", mpModel=0),
-            # await UdpTransportTarget.create(("localhost", MANAGER_PORT)), # TODO: Fix this
-            await UdpTransportTarget.create(("demo.pysnmp.com", 162)),
+            await UdpTransportTarget.create(
+                ("localhost", tests.manager_context.MANAGER_PORT)
+            ),
             ContextData(),
             "trap",
             NotificationType(ObjectIdentity("1.3.6.1.6.3.1.1.5.2"))
@@ -55,8 +59,7 @@ async def test_send_trap_generic():
                 ("1.3.6.1.2.1.1.1.0", OctetString("my system")),
             ),
         )
-        assert errorIndication is None
-        assert errorStatus == 0
-        assert errorIndex == 0
-        assert varBinds == []
+
         snmpEngine.closeDispatcher()
+        await asyncio.sleep(1)
+        assert tests.manager_context.message_count == 1
diff --git a/tests/hlapi/asyncio/agent/ntforg/test_v3-trap.py b/tests/hlapi/asyncio/agent/ntforg/test_v3-trap.py
@@ -1,19 +1,27 @@
+import asyncio
 import pytest
+
+
 from pysnmp.hlapi.v3arch.asyncio import *
-from tests.manager_context import MANAGER_PORT, ManagerContextManager
+import tests.manager_context
 
 
 @pytest.mark.asyncio
 async def test_send_v3_trap_notification():
-    async with ManagerContextManager():
+    async with tests.manager_context.ManagerContextManager():
+        # 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", MANAGER_PORT)),
+            await UdpTransportTarget.create(
+                ("localhost", tests.manager_context.MANAGER_PORT)
+            ),
             ContextData(),
             "trap",
             NotificationType(ObjectIdentity("IF-MIB", "linkDown")),
         )
 
         snmpEngine.closeDispatcher()
+        await asyncio.sleep(1)
+        assert tests.manager_context.message_count == 1
diff --git a/tests/manager_context.py b/tests/manager_context.py
@@ -1,4 +1,5 @@
 # manager_context.py
+import asyncio
 from typing import Tuple
 
 
@@ -8,12 +9,13 @@
 from pysnmp.entity import engine, config
 from pysnmp.proto.api import v2c
 
-import asyncio
-
 # Set the port to 1622 instead of 162, because 162 is a
 # 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]:
     # Create SNMP engine
@@ -62,6 +64,8 @@ async def start_manager() -> Tuple[SnmpEngine, ntfrcv.NotificationReceiver]:
     def cbFun(
         snmpEngine, stateReference, contextEngineId, contextName, varBinds, cbCtx
     ):
+        global message_count
+        message_count += 1
         print(
             'Notification from ContextEngineId "{}", ContextName "{}"'.format(
                 contextEngineId.prettyPrint(), contextName.prettyPrint()
@@ -103,6 +107,9 @@ class ManagerContextManager:
     receiver: ntfrcv.NotificationReceiver
 
     async def __aenter__(self):
+        global message_count
+        message_count = 0
+
         self.manager, self.receiver = await start_manager()
         return self.manager