diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py index 99f5e6c4..72c741f3 100644 --- a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v2c_bulkwalk.py @@ -1,8 +1,11 @@ +import math import pytest from pysnmp.hlapi.v1arch.asyncio import * from tests.agent_context import AGENT_PORT, AgentContextManager +total_count = 68 + @pytest.mark.asyncio @pytest.mark.parametrize("max_repetitions", [1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30]) @@ -35,7 +38,7 @@ async def test_v2c_get_table_bulk(max_repetitions): assert len(varBinds) == max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 60 / max_repetitions + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpDispatcher.transport_dispatcher.close_dispatcher() @@ -146,7 +149,7 @@ async def test_v2c_get_table_bulk_0_7(): assert len(varBinds) == max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 9 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpDispatcher.transport_dispatcher.close_dispatcher() @@ -180,15 +183,15 @@ async def test_v2c_get_table_bulk_0_8(): assert len(varBinds) == max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 8 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpDispatcher.transport_dispatcher.close_dispatcher() @pytest.mark.asyncio -async def test_v2c_get_table_bulk_0_31(): +async def test_v2c_get_table_bulk_0_35(): async with AgentContextManager(): snmpDispatcher = SnmpDispatcher() - max_repetitions = 31 + max_repetitions = 35 # 68/2 + 1 objects = bulk_walk_cmd( snmpDispatcher, CommunityData("public"), @@ -212,10 +215,10 @@ async def test_v2c_get_table_bulk_0_31(): assert errorIndication is None assert errorStatus == 0 - assert len(varBinds) == 60 - max_repetitions + assert len(varBinds) == total_count - max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 2 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpDispatcher.transport_dispatcher.close_dispatcher() @@ -243,7 +246,7 @@ async def test_v2c_get_table_bulk_0_60(): assert len(varBinds) == max_repetitions assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" - assert len(objects_list) == 1 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpDispatcher.transport_dispatcher.close_dispatcher() diff --git a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulkwalk.py b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulkwalk.py index ab4f39ef..19298252 100644 --- a/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulkwalk.py +++ b/tests/hlapi/v3arch/asyncio/manager/cmdgen/test_v2c_bulkwalk.py @@ -1,8 +1,11 @@ +import math import pytest from pysnmp.hlapi.v3arch.asyncio import * from tests.agent_context import AGENT_PORT, AgentContextManager +total_count = 68 # Ensure this matches the actual total count of SNMP objects + @pytest.mark.asyncio @pytest.mark.parametrize("max_repetitions", [1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30]) @@ -36,7 +39,7 @@ async def test_v2c_get_table_bulk(max_repetitions): assert len(varBinds) == max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 60 / max_repetitions + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpEngine.close_dispatcher() @@ -150,7 +153,7 @@ async def test_v2c_get_table_bulk_0_7(): assert len(varBinds) == max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 9 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpEngine.close_dispatcher() @@ -185,15 +188,15 @@ async def test_v2c_get_table_bulk_0_8(): assert len(varBinds) == max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 8 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpEngine.close_dispatcher() @pytest.mark.asyncio -async def test_v2c_get_table_bulk_0_31(): +async def test_v2c_get_table_bulk_0_35(): async with AgentContextManager(): snmpEngine = SnmpEngine() - max_repetitions = 31 + max_repetitions = 35 # (68/2) + 1 objects = bulk_walk_cmd( snmpEngine, CommunityData("public"), @@ -218,10 +221,10 @@ async def test_v2c_get_table_bulk_0_31(): assert errorIndication is None assert errorStatus == 0 - assert len(varBinds) == 60 - max_repetitions + assert len(varBinds) == total_count - max_repetitions # assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysName.0" - assert len(objects_list) == 2 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpEngine.close_dispatcher() @@ -250,7 +253,7 @@ async def test_v2c_get_table_bulk_0_60(): assert len(varBinds) == max_repetitions assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysObjectID.0" - assert len(objects_list) == 1 + assert len(objects_list) == math.ceil(total_count / max_repetitions) snmpEngine.close_dispatcher()