Adapt code to pysnmp 7 async API

Signed-off-by: Otto Moerbeek <otto.moerbeek@open-xchange.com>

diff --git a/regression-tests.recursor-dnssec/requirements.in b/regression-tests.recursor-dnssec/requirements.in
index 4a6940b49f03ff2b6e1f1024496bf8827bc08621..2d822b8af4ea7916ce960ee33e7fa5477a1ac96a 100644 (file)
--- a/regression-tests.recursor-dnssec/requirements.in
+++ b/regression-tests.recursor-dnssec/requirements.in
@@ -5,7 +5,7 @@ dnspython>=1.11
 pytest
 protobuf>=3.0
 pyasn1==0.6.3
-pysnmp>=5,<6
+pysnmp>=7
 requests>=2.1.0
 Twisted>0.15.0
 pyyaml==6.0.1

diff --git a/regression-tests.recursor-dnssec/test_SNMP.py b/regression-tests.recursor-dnssec/test_SNMP.py
index b3c2a1f9266ae1723cfcd3ebfbe64598e1597b05..d1f4db74672b2b47bb5ee25e9071e9001027d6df 100644 (file)
--- a/regression-tests.recursor-dnssec/test_SNMP.py
+++ b/regression-tests.recursor-dnssec/test_SNMP.py
@@ -1,6 +1,8 @@
+import asyncio
 import time
 
 from pysnmp.hlapi import *
+from pysnmp.hlapi.v3arch.asyncio import *
 
 from recursortests import RecursorTest
 
@@ -35,32 +37,38 @@ class SNMPTest(RecursorTest):
         # check memory usage > 0
         self.assertGreater(results["1.3.6.1.4.1.43315.2.1.76.0"], 0)
 
-    def _getSNMPStats(self, auth):
+    async def _getSNMPStats(self, auth):
         results = {}
-        for errorIndication, errorStatus, errorIndex, varBinds in nextCmd(
+        snmpEngine = SnmpEngine()
+        iterator = walk_cmd(
             SnmpEngine(),
             auth,
-            UdpTransportTarget((self._snmpServer, self._snmpPort), timeout=self._snmpTimeout),
+            await UdpTransportTarget.create((self._snmpServer, self._snmpPort), timeout=self._snmpTimeout),
             ContextData(),
             ObjectType(ObjectIdentity(self._snmpOID)),
             lookupMib=False,
-        ):
-            self.assertFalse(errorIndication)
-            self.assertFalse(errorStatus)
-            self.assertTrue(varBinds)
-            for key, value in varBinds:
-                keystr = key.prettyPrint()
-                if not keystr.startswith(self._snmpOID):
-                    continue
-                results[keystr] = value
+        )
+
+        list = [item async for item in iterator]
+
+        for errorIndication, errorStatus, errorIndex, varBinds in list:
+           self.assertFalse(errorIndication)
+           self.assertFalse(errorStatus)
+           self.assertTrue(varBinds)
+           for key, value in varBinds:
+               keystr = key.prettyPrint()
+               if not keystr.startswith(self._snmpOID):
+                   continue
+               results[keystr] = value
 
+        snmpEngine.close_dispatcher()
         return results
 
     def _checkStats(self, auth):
         # wait 1s so that the uptime is > 0
         time.sleep(1)
 
-        results = self._getSNMPStats(auth)
+        results = asyncio.run(self._getSNMPStats(auth))
         self._checkStatsValues(results)
 
     def testSNMPv2Stats(self):