Skip to content

Commit

Permalink
Implement getnext capable of multi-var query
Browse files Browse the repository at this point in the history
This adds an alternative getnext implementation (getnext2), which is
capable of dispatching multi-varbind queries, and capable of translating
the response varbinds to symbolic names.
  • Loading branch information
lunkwill42 committed Oct 6, 2023
1 parent 04c37f0 commit 039792d
Show file tree
Hide file tree
Showing 2 changed files with 47 additions and 1 deletion.
39 changes: 39 additions & 0 deletions src/zino/snmp.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,45 @@ async def _getnext(self, object_type: ObjectType) -> ObjectType:
# var_binds should be a sequence of sequences with one inner sequence that contains the result.
return var_binds[0][0]

async def getnext2(self, *variables: Sequence[str]) -> Sequence[SNMPVarBind]:
"""Dispatches a GET-NEXT query for the given variables and returns a result where the fetched variables are
identified symbolically.
Example usage:
>>> await s.getnext2(("IF-MIB", "ifName", "1"), ("IF-MIB", "ifAlias", "1"])
[(Identifier("IF-MIB", "ifName", OID('.2')), "Gi2/3"),
(Identifier("IF-MIB", "ifName", OID('.2')), "Uplink to somewhere")]
>>>
:param variables: Values for defining OIDs. For detailed use see
https://github.com/pysnmp/pysnmp/blob/bc1fb3c39764f36c1b7c9551b52ef8246b9aea7c/pysnmp/smi/rfc1902.py#L35-L49
:return: A sequence of MibObject instances representing the resulting MIB variables, or None if nothing could
be found
"""
query = [self._oid_to_object_type(*var) for var in variables]
response = await self._getnext2(*query)
return [_convert_varbind(*r) for r in response]

async def _getnext2(self, *variables: ObjectType) -> Sequence[ObjectType]:
"""SNMP-GETNEXTs the given variables
:param variables: An sequence of ObjectTypes representing the objects you want to query
:return: A sequence of ObjectTypes representing the resulting MIB variables, or None if nothing could be found
"""
try:
error_indication, error_status, error_index, var_bind_table = await nextCmd(
_get_engine(),
self.community_data,
self.udp_transport_target,
ContextData(),
*variables,
)
except PysnmpMibNotFoundError as error:
raise MibNotFoundError(error)

Check warning on line 222 in src/zino/snmp.py

View check run for this annotation

Codecov / codecov/patch

src/zino/snmp.py#L221-L222

Added lines #L221 - L222 were not covered by tests
self._raise_errors(error_indication, error_status, error_index, *variables)
# The table should contain only one set of results for our query
return var_bind_table[0]

async def walk(self, *oid: str) -> list[MibObject]:
"""Uses SNMP-GETNEXT calls to get all objects in the subtree with oid as root
Example usage:
Expand Down
9 changes: 8 additions & 1 deletion tests/snmp_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

from zino.config.models import PollDevice
from zino.oid import OID
from zino.snmp import SNMP, MibNotFoundError, NoSuchNameError
from zino.snmp import SNMP, Identifier, MibNotFoundError, NoSuchNameError


@pytest.fixture(scope="session")
Expand Down Expand Up @@ -30,6 +30,13 @@ async def test_getnext(self, snmp_client):
assert isinstance(response.oid, OID)
assert isinstance(response.value, int)

@pytest.mark.asyncio
async def test_getnext2_should_return_symbolic_identifiers(self, snmp_client):
response = await snmp_client.getnext2(("IF-MIB", "ifName", "1"), ("IF-MIB", "ifAlias", "1"))
assert len(list(response)) == 2
assert any(k == Identifier("IF-MIB", "ifName", OID(".2")) for k, v in response)
assert any(k == Identifier("IF-MIB", "ifAlias", OID(".2")) for k, v in response)

@pytest.mark.asyncio
async def test_walk(self, snmp_client):
response = await snmp_client.walk("SNMPv2-MIB", "sysUpTime")
Expand Down

0 comments on commit 039792d

Please sign in to comment.