Add functionality for handling varbind errors in get command

changelog.d/261.fixed.md

@@ -0,0 +1 @@
+Add handling of varbind errors for snmp get actions.

src/zino/snmp.py

@@ -21,7 +21,7 @@
     nextCmd,
 )
 from pysnmp.proto import errind
-from pysnmp.proto.rfc1905 import errorStatus
+from pysnmp.proto.rfc1905 import EndOfMibView, NoSuchInstance, NoSuchObject, errorStatus
 from pysnmp.smi import builder, view
 from pysnmp.smi.error import MibNotFoundError as PysnmpMibNotFoundError
 
@@ -75,7 +75,7 @@ class SnmpError(Exception):
     """Base class for SNMP, MIB and OID specific errors"""
 
 
-class ErrorIndication(Exception):
+class ErrorIndication(SnmpError):
     """Class for SNMP errors that occur locally,
     as opposed to being reported from a different SNMP entity.
     """
@@ -96,6 +96,22 @@ class NoSuchNameError(ErrorStatus):
     """Represents the "noSuchName" error. Raised if an object could not be found at an OID."""
 
 
+class VarBindError(SnmpError):
+    """Base class for errors carried in varbinds and not in the errorStatus or errorIndication fields"""
+
+
+class NoSuchObjectError(VarBindError):
+    """Raised if an object could not be found at an OID"""
+
+
+class NoSuchInstanceError(VarBindError):
+    """Raised if an instance could not be found at an OID"""
+
+
+class EndOfMibViewError(VarBindError):
+    """Raised if end of MIB view is encountered"""
+
+
 class SNMP:
     """Represents an SNMP management session for a single device"""
 
@@ -126,7 +142,9 @@ async def get(self, *oid: str) -> MibObject:
         except PysnmpMibNotFoundError as error:
             raise MibNotFoundError(error)
         self._raise_errors(error_indication, error_status, error_index, query)
-        return self._object_type_to_mib_object(var_binds[0])
+        result = var_binds[0]
+        self._raise_varbind_errors(result)
+        return self._object_type_to_mib_object(result)
 
     def _raise_errors(
         self,
@@ -153,6 +171,17 @@ def _raise_errors(
             else:
                 raise ErrorStatus(f"SNMP operation failed with error {error_name} for {error_object.oid}")
 
+    def _raise_varbind_errors(self, object_type: ObjectType):
+        """Raises a relevant exception if an error has occurred in a varbind"""
+        oid = OID(str(object_type[0]))
+        value = object_type[1]
+        if isinstance(value, NoSuchObject):
+            raise NoSuchObjectError(f"Could not find object at {oid}")
+        if isinstance(value, NoSuchInstance):
+            raise NoSuchInstanceError(f"Could not find instance at {oid}")
+        if isinstance(value, EndOfMibView):
+            raise EndOfMibViewError("Reached end of MIB view")
+
     async def getnext(self, *oid: str) -> MibObject:
         """SNMP-GETNEXTs the given oid
         Example usage:

tests/snmp_test.py

@@ -2,12 +2,26 @@
 from unittest.mock import Mock, patch
 
 import pytest
-from pysnmp.hlapi.asyncio import Udp6TransportTarget, UdpTransportTarget
+from pysnmp.hlapi.asyncio import (
+    ObjectIdentity,
+    ObjectType,
+    Udp6TransportTarget,
+    UdpTransportTarget,
+)
 from pysnmp.proto import errind
+from pysnmp.proto.rfc1905 import EndOfMibView, NoSuchInstance, NoSuchObject
 
 from zino.config.models import PollDevice
 from zino.oid import OID
-from zino.snmp import SNMP, Identifier, MibNotFoundError, NoSuchNameError
+from zino.snmp import (
+    SNMP,
+    EndOfMibViewError,
+    Identifier,
+    MibNotFoundError,
+    NoSuchInstanceError,
+    NoSuchNameError,
+    NoSuchObjectError,
+)
 
 
 @pytest.fixture(scope="session")
@@ -241,3 +255,38 @@ def test_when_device_address_is_ipv6_then_udp6_transport_should_be_returned(self
 
     def test_when_device_address_is_ipv4_then_udp_transport_should_be_returned(self, snmp_client):
         assert isinstance(snmp_client.udp_transport_target, UdpTransportTarget)
+
+
+class TestVarBindErrors:
+    """
+    Test class for verifying the handling of varbind errors in SNMP commands.
+
+    This class contains tests that check if the correct exceptions are raised when
+    varbind errors (NoSuchObject, NoSuchInstance, EndOfMibView) are encountered
+    in the response to an SNMP command.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "error, exception",
+        [
+            (NoSuchObject(""), NoSuchObjectError),
+            (NoSuchInstance(""), NoSuchInstanceError),
+            (EndOfMibView(""), EndOfMibViewError),
+        ],
+        ids=["NoSuchObject-NoSuchObjectError", "NoSuchInstance-NoSuchInstanceError", "EndOfMibView-EndOfMibViewError"],
+    )
+    async def test_get_should_raise_exception(self, error, exception, snmp_client, monkeypatch):
+        query = ("SNMPv2-MIB", "sysDescr", 0)
+        object_type = ObjectType(ObjectIdentity(*query), error)
+
+        mock_results = None, None, None, [object_type]
+        future = asyncio.Future()
+        future.set_result(mock_results)
+        get_mock = Mock(return_value=future)
+        monkeypatch.setattr("zino.snmp.getCmd", get_mock)
+
+        snmp_client._resolve_object(object_type)
+
+        with pytest.raises(exception):
+            await snmp_client.get(*query)