Skip to content

Commit

Permalink
Update funcs so they do not return empty values
Browse files Browse the repository at this point in the history
  • Loading branch information
stveit committed Oct 4, 2023
1 parent 9985797 commit ad78a59
Showing 1 changed file with 9 additions and 15 deletions.
24 changes: 9 additions & 15 deletions src/zino/snmp.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,15 +103,15 @@ def __init__(self, device: PollDevice, retries=5):
self.device = device
self.retries = retries

async def get(self, *oid: str) -> Union[MibObject, None]:
async def get(self, *oid: str) -> MibObject:
"""SNMP-GETs the given oid
Example usage:
get("SNMPv2-MIB", "sysUpTime", 0)
get("1.3.6.1.2.1.1.3.0")
:param oid: Values for defining an OID. For detailed use see
https://github.com/pysnmp/pysnmp/blob/bc1fb3c39764f36c1b7c9551b52ef8246b9aea7c/pysnmp/smi/rfc1902.py#L35-L49
:return: A MibObject representing the resulting MIB variable or None if nothing could be found
:return: A MibObject representing the resulting MIB variable
"""
query = self._oid_to_object_type(*oid)
try:
Expand All @@ -125,10 +125,9 @@ async def get(self, *oid: str) -> Union[MibObject, None]:
except PysnmpMibNotFoundError as error:
raise MibNotFoundError(error)
self._raise_errors(error_indication, error_status, error_index, query)
for var_bind in var_binds:
return self._object_type_to_mib_object(var_bind)
return self._object_type_to_mib_object(var_binds[0])

def _raise_errors(self, error_indication: str, error_status: str, error_index: int, *query: ObjectType) -> bool:
def _raise_errors(self, error_indication: str, error_status: str, error_index: int, *query: ObjectType):
"""Raises a relevant exception if an error has occurred"""
# Local errors (timeout, config errors etc)
if error_indication:
Expand All @@ -147,7 +146,7 @@ def _raise_errors(self, error_indication: str, error_status: str, error_index: i
else:
raise ErrorStatus(f"SNMP operation failed with error {error_name} for {error_object.oid}")

async def getnext(self, *oid: str) -> Union[MibObject, None]:
async def getnext(self, *oid: str) -> MibObject:
"""SNMP-GETNEXTs the given oid
Example usage:
getnext("SNMPv2-MIB", "sysUpTime")
Expand All @@ -159,11 +158,9 @@ async def getnext(self, *oid: str) -> Union[MibObject, None]:
"""
query = self._oid_to_object_type(*oid)
object_type = await self._getnext(query)
if not object_type:
return None
return self._object_type_to_mib_object(object_type)

async def _getnext(self, object_type: ObjectType) -> Union[ObjectType, None]:
async def _getnext(self, object_type: ObjectType) -> ObjectType:
"""SNMP-GETNEXTs the given object_type
:param object_type: An ObjectType representing the object you want to query
Expand All @@ -181,8 +178,7 @@ async def _getnext(self, object_type: ObjectType) -> Union[ObjectType, None]:
raise MibNotFoundError(error)
self._raise_errors(error_indication, error_status, error_index, object_type)
# var_binds should be a sequence of sequences with one inner sequence that contains the result.
if var_binds and var_binds[0]:
return var_binds[0][0]
return var_binds[0][0]

async def walk(self, *oid: str) -> list[MibObject]:
"""Uses SNMP-GETNEXT calls to get all objects in the subtree with oid as root
Expand All @@ -200,7 +196,7 @@ async def walk(self, *oid: str) -> list[MibObject]:
original_oid = OID(str(current_object[0]))
while True:
current_object = await self._getnext(current_object)
if not current_object or not original_oid.is_a_prefix_of(str(current_object[0])):
if not original_oid.is_a_prefix_of(str(current_object[0])):
break
mib_object = self._object_type_to_mib_object(current_object)
results.append(mib_object)
Expand Down Expand Up @@ -240,8 +236,6 @@ async def _getbulk(self, object_type: ObjectType, max_repetitions: int) -> list[
except PysnmpMibNotFoundError as error:
raise MibNotFoundError(error)
self._raise_errors(error_indication, error_status, error_index, object_type)
if not var_binds:
return []
return var_binds[0]

async def getbulk2(self, *variables: Sequence[str], max_repetitions: int = 10) -> Sequence[Sequence[SNMPVarBind]]:
Expand Down Expand Up @@ -282,7 +276,7 @@ async def _getbulk2(
except PysnmpMibNotFoundError as error:
raise MibNotFoundError(error)
self._raise_errors(error_indication, error_status, error_index, *variables)
return var_bind_table or []
return var_bind_table

async def bulkwalk(self, *oid: str, max_repetitions: int = 10) -> list[MibObject]:
"""Uses SNMP-BULK calls to get all objects in the subtree with oid as root
Expand Down

0 comments on commit ad78a59

Please sign in to comment.