Fixed etingof#270. Added more test cases.

lextudio · Aug 25, 2024 · 5e8b67b · 5e8b67b
1 parent b3f297a
commit 5e8b67b
Show file tree

Hide file tree

Showing 4 changed files with 139 additions and 53 deletions.
diff --git a/pysnmp/hlapi/asyncio/cmdgen.py b/pysnmp/hlapi/asyncio/cmdgen.py
@@ -734,9 +734,7 @@ async def walkCmd(
     maxRows = options.get("maxRows", 0)
     maxCalls = options.get("maxCalls", 0)
 
-    vbProcessor = CommandGeneratorVarBinds()
-
-    initialVars = [x[0] for x in vbProcessor.makeVarBinds(snmpEngine, varBinds)]
+    initialVars = [x[0] for x in VB_PROCESSOR.makeVarBinds(snmpEngine, varBinds)]
 
     totalRows = totalCalls = 0
 
@@ -798,7 +796,9 @@ async def walkCmd(
 
         if initialVarBinds:
             varBinds = initialVarBinds
-            initialVars = [x[0] for x in vbProcessor.makeVarBinds(snmpEngine, varBinds)]
+            initialVars = [
+                x[0] for x in VB_PROCESSOR.makeVarBinds(snmpEngine, varBinds)
+            ]
 
         if maxRows and totalRows >= maxRows:
             return
@@ -932,9 +932,7 @@ async def bulkWalkCmd(
     maxRows = options.get("maxRows", 0)
     maxCalls = options.get("maxCalls", 0)
 
-    vbProcessor = CommandGeneratorVarBinds()
-
-    initialVars = [x[0] for x in vbProcessor.makeVarBinds(snmpEngine, varBinds)]
+    initialVars = [x[0] for x in VB_PROCESSOR.makeVarBinds(snmpEngine, varBinds)]
     nullVarBinds = [False] * len(initialVars)
 
     totalRows = totalCalls = 0
@@ -1005,7 +1003,11 @@ async def bulkWalkCmd(
                         nullVarBinds[col] = True
                     if not lexicographicMode and not initialVars[col].isPrefixOf(name):
                         varBindTable[row][col] = previousVarBinds[col][0], endOfMibView
-                        nullVarBinds[col] = True
+                        if len(varBindTable) == 1:
+                            stopFlag = True
+                            break
+                        else:
+                            nullVarBinds[col] = True
                 if stopFlag:
                     varBindTable = row and varBindTable[: row - 1] or []
                     break
@@ -1034,6 +1036,6 @@ async def bulkWalkCmd(
                 if initialVarBinds:
                     varBinds = initialVarBinds
                     initialVars = [
-                        x[0] for x in vbProcessor.makeVarBinds(snmpEngine, varBinds)
+                        x[0] for x in VB_PROCESSOR.makeVarBinds(snmpEngine, varBinds)
                     ]
                     nullVarBinds = [False] * len(initialVars)
diff --git a/pysnmp/hlapi/asyncio/slim.py b/pysnmp/hlapi/asyncio/slim.py
@@ -245,7 +245,7 @@ async def bulk(
         *varBinds,
         timeout: int = 1,
         retries: int = 5,
-    ) -> "tuple[ErrorIndication, Integer32 | int, Integer32 | int, tuple[ObjectType]]":
+    ) -> "tuple[ErrorIndication, Integer32 | int, Integer32 | int, tuple[tuple[ObjectType, ...], ...]]":
         r"""Creates a generator to perform SNMP GETBULK query.
 
         When iterator gets advanced by :py:mod:`asyncio` main loop,

diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulk.py
@@ -2,62 +2,30 @@
 
 from pysnmp.hlapi.asyncio import *
 from tests.agent_context import AGENT_PORT, AgentContextManager
+import pytest
+from pysnmp.hlapi.asyncio import *
+from tests.agent_context import AGENT_PORT, AgentContextManager
 
 
 @pytest.mark.asyncio
-async def test_v2c_bulk():
+@pytest.mark.parametrize("num_bulk", [1, 2, 50])
+async def test_v2c_bulk(num_bulk):
     async with AgentContextManager():
         with Slim() as slim:
             errorIndication, errorStatus, errorIndex, varBinds = await slim.bulk(
                 "public",
                 "localhost",
                 AGENT_PORT,
                 0,
-                50,
+                num_bulk,
                 ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
             )
 
             assert errorIndication is None
             assert errorStatus == 0
-            assert len(varBinds) == 50
+            assert len(varBinds) == num_bulk
             assert varBinds[0][0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0"
-            assert varBinds[0][0][1].prettyPrint() == "PYSNMP-MIB::pysnmp"
-            # assert isinstance(varBinds[0][0][1], ObjectIdentifier)
-
-
-@pytest.mark.asyncio
-async def test_v2c_get_table_bulk():
-    async with AgentContextManager():
-        snmpEngine = SnmpEngine()
-        objects = bulkWalkCmd(
-            snmpEngine,
-            CommunityData("public"),
-            UdpTransportTarget(("localhost", AGENT_PORT)),
-            ContextData(),
-            0,
-            4,
-            ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
-        )
-
-        objects_list = [item async for item in objects]
-
-        errorIndication, errorStatus, errorIndex, varBinds = objects_list[0]
-
-        assert errorIndication is None
-        assert errorStatus == 0
-        assert len(varBinds) == 1
-        assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0"
-        assert varBinds[0][1].prettyPrint() == "PYSNMP-MIB::pysnmp"
-        # assert isinstance(varBinds[0][1], ObjectIdentifier)
-
-        errorIndication, errorStatus, errorIndex, varBinds = objects_list[1]
-
-        assert errorIndication is None
-        assert errorStatus == 0
-        assert len(varBinds) == 1
-        assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0"
-        # assert isinstance(varBinds[0][1], TimeTicks)
-
-        assert len(objects_list), 50
-
-        snmpEngine.closeDispatcher()
+            if num_bulk > 1:
+                assert varBinds[1][0][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0"
+            if num_bulk > 2:
+                assert varBinds[2][0][0].prettyPrint() == "SNMPv2-MIB::sysContact.0"
diff --git a/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulkwalk.py b/tests/hlapi/asyncio/manager/cmdgen/test_v2c_bulkwalk.py
@@ -0,0 +1,116 @@
+import pytest
+
+from pysnmp.hlapi.asyncio import *
+from tests.agent_context import AGENT_PORT, AgentContextManager
+
+
+@pytest.mark.asyncio
+async def test_v2c_get_table_bulk():
+    async with AgentContextManager():
+        snmpEngine = SnmpEngine()
+        objects = bulkWalkCmd(
+            snmpEngine,
+            CommunityData("public"),
+            UdpTransportTarget(("localhost", AGENT_PORT)),
+            ContextData(),
+            0,
+            4,
+            ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
+        )
+
+        objects_list = [obj async for obj in objects]
+
+        errorIndication, errorStatus, errorIndex, varBinds = objects_list[0]
+
+        assert errorIndication is None
+        assert errorStatus == 0
+        assert len(varBinds) == 1
+        assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0"
+        assert varBinds[0][1].prettyPrint() == "PYSNMP-MIB::pysnmp"
+        # assert isinstance(varBinds[0][1], ObjectIdentifier)
+
+        errorIndication, errorStatus, errorIndex, varBinds = objects_list[1]
+
+        assert errorIndication is None
+        assert errorStatus == 0
+        assert len(varBinds) == 1
+        assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysUpTime.0"
+        # assert isinstance(varBinds[0][1], TimeTicks)
+
+        assert len(objects_list), 50
+
+        snmpEngine.closeDispatcher()
+
+
+@pytest.mark.asyncio
+async def test_v2c_get_table_bulk_0_4():
+    async with AgentContextManager():
+        snmpEngine = SnmpEngine()
+        index = 0
+        async for errorIndication, errorStatus, errorIndex, varBinds in bulkWalkCmd(
+            snmpEngine,
+            CommunityData("public"),
+            UdpTransportTarget(("localhost", AGENT_PORT)),
+            ContextData(),
+            0,
+            4,
+            ObjectType(ObjectIdentity("SNMPv2-MIB", "snmp")),
+            lexicographicMode=False,
+        ):
+            assert errorIndication is None
+            assert errorStatus == 0
+            assert len(varBinds) == 1
+            if index == 0:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpInPkts.0"
+
+            if index == 1:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpOutPkts.0"
+
+            if index == 26:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpSilentDrops.0"
+
+            if index == 27:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpProxyDrops.0"
+
+            index += 1
+
+        assert index == 28
+
+        snmpEngine.closeDispatcher()
+
+
+@pytest.mark.asyncio
+async def test_v2c_get_table_bulk_0_1():
+    async with AgentContextManager():
+        snmpEngine = SnmpEngine()
+        index = 0
+        async for errorIndication, errorStatus, errorIndex, varBinds in bulkWalkCmd(
+            snmpEngine,
+            CommunityData("public"),
+            UdpTransportTarget(("localhost", AGENT_PORT)),
+            ContextData(),
+            0,
+            1,
+            ObjectType(ObjectIdentity("SNMPv2-MIB", "snmp")),
+            lexicographicMode=False,
+        ):
+            assert errorIndication is None
+            assert errorStatus == 0
+            assert len(varBinds) == 1
+            if index == 0:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpInPkts.0"
+
+            if index == 1:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpOutPkts.0"
+
+            if index == 26:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpSilentDrops.0"
+
+            if index == 27:
+                assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::snmpProxyDrops.0"
+
+            index += 1
+
+        assert index == 28
+
+        snmpEngine.closeDispatcher()