1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
import pytest
from pysnmp.entity.engine import SnmpEngine
from pysnmp.hlapi.v1arch.asyncio.cmdgen import get_cmd, set_cmd, walk_cmd
from pysnmp.hlapi.v1arch.asyncio.dispatch import SnmpDispatcher
from pysnmp.hlapi.v1arch.asyncio.transport import UdpTransportTarget
from pysnmp.hlapi.v1arch.asyncio.auth import CommunityData
from pysnmp.proto.rfc1902 import Integer, OctetString
from pysnmp.smi import builder, compiler, view
from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType
from tests.agent_context import AGENT_PORT, AgentContextManager
@pytest.mark.asyncio
async def test_v1_set():
async with AgentContextManager():
with SnmpDispatcher() as snmpDispatcher:
iterator = await set_cmd(
snmpDispatcher,
CommunityData("public", mpModel=0),
await UdpTransportTarget.create(("localhost", AGENT_PORT)),
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai"),
)
errorIndication, errorStatus, errorIndex, varBinds = iterator
assert errorIndication is None
assert errorStatus == 0
assert len(varBinds) == 1
assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysLocation.0"
assert varBinds[0][1].prettyPrint() == "Shanghai"
assert isinstance(varBinds[0][1], OctetString)
@pytest.mark.asyncio
async def test_v1_set_mac_address():
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", "testMacAddress", 0)
), # "1.3.6.1.4.1.60069.9.9.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::testMacAddress.0"
assert (
varBinds[0][1].prettyPrint() == "00:11:22:33:44:55"
) # IMPORTANT: OCTET STRING Size constraint 6..6
errorIndication, errorStatus, errorIndex, varBinds = await set_cmd(
snmpDispatcher,
CommunityData("public", mpModel=0),
await UdpTransportTarget.create(("localhost", AGENT_PORT)),
ObjectType(
ObjectIdentity("LEXTUDIO-TEST-MIB", "testMacAddress", 0),
OctetString("00:11:22:33:44:55:66"), # GitHub issue #141
),
)
assert errorIndication is None
assert errorStatus == 3 # bad value
assert errorIndex == 1
assert len(varBinds) == 1
@pytest.mark.asyncio
async def test_v1_set_table_creation():
async with AgentContextManager(enable_table_creation=True):
with SnmpDispatcher() as snmpDispatcher:
# Perform a SNMP walk to get all object counts
objects = walk_cmd(
snmpDispatcher,
CommunityData("public", mpModel=0),
await UdpTransportTarget.create(("localhost", AGENT_PORT)),
ObjectType(ObjectIdentity("1.3.6")),
)
objects_list = [item async for item in objects]
assert len(objects_list) > 0
object_counts = len(objects_list)
errorIndication, errorStatus, errorIndex, varBinds = await set_cmd(
snmpDispatcher,
CommunityData("public", mpModel=0),
await UdpTransportTarget.create(("localhost", AGENT_PORT)),
ObjectType(
ObjectIdentity("1.3.6.6.1.5.2.97.98.99"), OctetString("My value")
),
)
assert errorIndication is None
assert errorStatus == 0
assert len(varBinds) == 1
assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.2.97.98.99"
assert varBinds[0][1].prettyPrint() == "My value"
assert type(varBinds[0][1]).__name__ == "OctetString"
errorIndication, errorStatus, errorIndex, varBinds = await set_cmd(
snmpDispatcher,
CommunityData("public", mpModel=0),
await UdpTransportTarget.create(("localhost", AGENT_PORT)),
ObjectType(ObjectIdentity("1.3.6.6.1.5.4.97.98.99"), Integer(4)),
)
assert errorIndication is None
assert errorStatus == 0
assert len(varBinds) == 1
assert varBinds[0][0].prettyPrint() == "SNMPv2-SMI::dod.6.1.5.4.97.98.99"
assert varBinds[0][1].prettyPrint() == "1"
# assert isinstance(varBinds[0][1], Integer)
# Perform a SNMP walk to get all object counts
objects = walk_cmd(
snmpDispatcher,
CommunityData("public", mpModel=0),
await UdpTransportTarget.create(("localhost", AGENT_PORT)),
ObjectType(ObjectIdentity("1.3.6")),
)
objects_list = [item async for item in objects]
assert len(objects_list) > 0
assert len(objects_list) == object_counts + 4
|
