From 039792d9de496ee3103dec14904ebd448fb683f1 Mon Sep 17 00:00:00 2001 From: Morten Brekkevold Date: Thu, 5 Oct 2023 11:54:07 +0200 Subject: [PATCH] Implement getnext capable of multi-var query 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. --- src/zino/snmp.py | 39 +++++++++++++++++++++++++++++++++++++++ tests/snmp_test.py | 9 ++++++++- 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/src/zino/snmp.py b/src/zino/snmp.py index b5ee0571c..3f1583e3e 100644 --- a/src/zino/snmp.py +++ b/src/zino/snmp.py @@ -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) + 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: diff --git a/tests/snmp_test.py b/tests/snmp_test.py index 0cd340bd3..7c6e77b3d 100644 --- a/tests/snmp_test.py +++ b/tests/snmp_test.py @@ -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") @@ -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")