From 8739a06d297adb447de6f2464920dd9d62574921 Mon Sep 17 00:00:00 2001 From: Lex Li Date: Sun, 3 Nov 2024 15:17:16 -0500 Subject: [PATCH] Added display hint test cases. --- .github/workflows/build-test.yml | 1 + pysnmp/hlapi/v1arch/dispatch.py | 2 + tests/agent_context.py | 47 +++++++++++++ .../manager/cmdgen/test_v1arch_v1_get.py | 69 +++++++++++++++++++ 4 files changed, 119 insertions(+) diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml index f58e44d0..ed1b3210 100644 --- a/.github/workflows/build-test.yml +++ b/.github/workflows/build-test.yml @@ -35,6 +35,7 @@ jobs: run: | poetry run mibdump NET-SNMP-EXAMPLES-MIB poetry run mibdump IF-MIB + poetry run mibdump LEXTUDIO-TEST-MIB poetry run pytest --junitxml=junit/test-results-${{ matrix.python-version }}.xml --cov=com --cov-report=xml --cov-report=html - name: Test summary uses: test-summary/action@v2 diff --git a/pysnmp/hlapi/v1arch/dispatch.py b/pysnmp/hlapi/v1arch/dispatch.py index 8a484f98..f1ed2c76 100644 --- a/pysnmp/hlapi/v1arch/dispatch.py +++ b/pysnmp/hlapi/v1arch/dispatch.py @@ -6,6 +6,7 @@ # import warnings from time import time +from typing import Any from pyasn1.codec.ber import decoder, encoder from pysnmp import debug @@ -38,6 +39,7 @@ class AbstractSnmpDispatcher: PROTO_DISPATCHER = None transport_dispatcher: AbstractTransportDispatcher + cache: dict[str, Any] def __init__(self, transportDispatcher: AbstractTransportDispatcher = None): # type: ignore if transportDispatcher: diff --git a/tests/agent_context.py b/tests/agent_context.py index 957f9142..4384d293 100644 --- a/tests/agent_context.py +++ b/tests/agent_context.py @@ -83,6 +83,19 @@ async def start_agent( "SNMPv2-SMI", "MibScalar", "MibScalarInstance" ) + ( + DisplayString, + PhysAddress, + DateAndTime, + TextualConvention, + ) = mibBuilder.importSymbols( + "SNMPv2-TC", + "DisplayString", + "PhysAddress", + "DateAndTime", + "TextualConvention", + ) + class SlowMibScalarInstance(MibScalarInstance): def getValue(self, name, **context): time.sleep(2) # Add a 2-second sleep @@ -93,6 +106,26 @@ def setValue(self, value, name, **context): print(f"SET operation received. New value: {value}") return self.getSyntax().clone(value) + # Define a fixed date and time in the DateAndTime format + def get_fixed_date_and_time(): + # Fixed date: 2024-11-03, 19:52:40.0, UTC offset +0:0 + return [ + 2024 // 256, + 2024 % 256, # Year (2024) + 11, # Month (November) + 3, # Day + 19, # Hour (7 PM UTC) + 52, # Minute + 40, # Second + 0, # Deciseconds (0 for simplicity) + ord("+"), # UTC offset direction ('+') + 0, # Hours offset from UTC + 0, # Minutes offset from UTC + ] + + # Initialize the PhysAddress object with a sample MAC address (e.g., 00:11:22:33:44:55) + initial_phys_address = [0x00, 0x11, 0x22, 0x33, 0x44, 0x55] + mibBuilder.export_symbols( "__MY_MIB", MibScalar((1, 3, 6, 1, 4, 1, 60069, 9, 1), v2c.OctetString()), @@ -105,6 +138,20 @@ def setValue(self, value, name, **context): MibScalar((1, 3, 6, 1, 4, 1, 60069, 9, 4), v2c.OctetString()).setMaxAccess( "readonly" ), # PySMI <1.3.0 generates such objects + MibScalar((1, 3, 6, 1, 4, 1, 60069, 9, 5), DateAndTime()).setMaxAccess( + "read-write" + ), + MibScalarInstance( + (1, 3, 6, 1, 4, 1, 60069, 9, 5), + (0,), + DateAndTime(get_fixed_date_and_time()), + ), + MibScalar((1, 3, 6, 1, 4, 1, 60069, 9, 6), PhysAddress()).setMaxAccess( + "read-write" + ), + MibScalarInstance( + (1, 3, 6, 1, 4, 1, 60069, 9, 6), (0,), PhysAddress(initial_phys_address) + ), ) # --- end of Managed Object Instance initialization ---- diff --git a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py index ed0f8cd0..2d1336bb 100644 --- a/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py +++ b/tests/hlapi/v1arch/asyncio/manager/cmdgen/test_v1arch_v1_get.py @@ -3,6 +3,7 @@ import pytest from pysnmp.hlapi.v1arch.asyncio import * from pysnmp.proto.rfc1905 import errorStatus as pysnmp_errorStatus +from pysnmp.smi import builder, compiler, view from tests.agent_context import AGENT_PORT, AgentContextManager @@ -132,3 +133,71 @@ async def test_v1_get_no_access_object(): assert errorIndication is None assert errorStatus.prettyPrint() == "noSuchName" # v1 does not have noAccess snmpDispatcher.transport_dispatcher.close_dispatcher() + + +@pytest.mark.asyncio +async def test_v1_get_dateandtime_object(): + async with AgentContextManager(enable_custom_objects=True): + snmpDispatcher = SnmpDispatcher() + # Step 1: Set up MIB builder and add custom MIB directory + mibBuilder = builder.MibBuilder() + compiler.addMibCompiler(mibBuilder) + mibViewController = view.MibViewController(mibBuilder) + + # Load the custom MIB + mibBuilder.loadModules("LEXTUDIO-TEST-MIB") + snmpDispatcher.cache["mibViewController"] = mibViewController + + errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create( + ("localhost", AGENT_PORT), timeout=1, retries=0 + ), + ObjectType( + ObjectIdentity("LEXTUDIO-TEST-MIB", "testDateAndTime", 0) + ), # "1.3.6.1.4.1.60069.9.5.0" + ) + assert errorIndication is None + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "LEXTUDIO-TEST-MIB::testDateAndTime.0" + assert ( + varBinds[0][1].prettyPrint() == "2024-11-3,19:52:40.0,+0:0" + ) # IMPORTANT: test DateAndTime display hint. + + +@pytest.mark.asyncio +async def test_v1_get_physaddress_object(): + async with AgentContextManager(enable_custom_objects=True): + snmpDispatcher = SnmpDispatcher() + # Step 1: Set up MIB builder and add custom MIB directory + mibBuilder = builder.MibBuilder() + compiler.addMibCompiler(mibBuilder) + mibViewController = view.MibViewController(mibBuilder) + + # Load the custom MIB + mibBuilder.loadModules("LEXTUDIO-TEST-MIB") + snmpDispatcher.cache["mibViewController"] = mibViewController + + errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( + snmpDispatcher, + CommunityData("public", mpModel=0), + await UdpTransportTarget.create( + ("localhost", AGENT_PORT), timeout=1, retries=0 + ), + ObjectType( + ObjectIdentity("LEXTUDIO-TEST-MIB", "testPhysAddress", 0) + ), # "1.3.6.1.4.1.60069.9.6.0" + ) + assert errorIndication is None + assert errorIndication is None + assert errorStatus == 0 + assert errorIndex == 0 + assert len(varBinds) == 1 + assert varBinds[0][0].prettyPrint() == "LEXTUDIO-TEST-MIB::testPhysAddress.0" + assert ( + varBinds[0][1].prettyPrint() == "00:11:22:33:44:55" + ) # IMPORTANT: test PhysAddress display hint.