Skip to content
Snippets Groups Projects
Commit 5ecfcbe5 authored by Taya Snijder's avatar Taya Snijder
Browse files

polished code

parent fd7ecb8e
No related branches found
No related tags found
1 merge request!288Resolve L2SS-446 "Extend snmp client to support mib files"
...@@ -14,7 +14,7 @@ __all__ = ["SNMP_client"] ...@@ -14,7 +14,7 @@ __all__ = ["SNMP_client"]
snmp_to_numpy_dict = { snmp_to_numpy_dict = {
hlapi.Integer32: numpy.int64, hlapi.Integer32: numpy.int64,
hlapi.TimeTicks: numpy.int64, hlapi.TimeTicks: numpy.int64,
str: str, hlapi.OctetString: str,
hlapi.ObjectIdentity: str, hlapi.ObjectIdentity: str,
hlapi.Counter32: numpy.int64, hlapi.Counter32: numpy.int64,
hlapi.Gauge32: numpy.int64, hlapi.Gauge32: numpy.int64,
...@@ -201,30 +201,30 @@ class snmp_attribute: ...@@ -201,30 +201,30 @@ class snmp_attribute:
for varBind in varBinds: for varBind in varBinds:
# Get all the types. Custom and base. Custom types are merely wrapped base types. # Some MIB's used custom types, some dont. Custom types are merely wrapped base types.
varbind_types = varBind[1].__class__.__bases__ + (type(varBind[1]),) varbind_types = varBind[1].__class__.__bases__ + (type(varBind[1]),)
snmp_type = None snmp_type = None
# find if one of the base types is present.
for i in varbind_types: for i in varbind_types:
if i in snmp_to_numpy_dict.keys(): if i in snmp_to_numpy_dict.keys():
snmp_type = i snmp_type = i
if snmp_type is None: if snmp_type is None:
TypeError(f"Error: did not find a valid snmp type. Got: {varbind_types}, expected one of: {snmp_to_numpy_dict.keys()}") raise TypeError(f"Error: did not find a valid snmp type. Got: {varbind_types}, expected one of: '{snmp_to_numpy_dict.keys()}'")
if snmp_type == hlapi.IpAddress: if snmp_type is hlapi.IpAddress:
# IpAddress values get printed as their raw value but in hex (7F 20 20 01 for 127.0.0.1 for example) # IpAddress values get printed as their raw value but in hex (7F 20 20 01 for 127.0.0.1 for example)
vals.append(varBind[1].prettyPrint()) vals.append(varBind[1].prettyPrint())
elif snmp_type == hlapi.Integer32 and self.dtype == str: elif snmp_type is hlapi.Integer32 and self.dtype == str:
# Integers can have 'named values', Where a value can be translated to a specific name. A dict basically # Integers can have 'named values', Where a value can be translated to a specific name. A dict basically
# Example: {1: "other", 2: "invalid", 3: "dynamic", 4: "static",} # Example: {1: "other", 2: "invalid", 3: "dynamic", 4: "static",}
if varBind[1].namedValues == {}: if varBind[1].namedValues == {}:
# An empty dict {} means no namedValue's are present. # An empty dict {} means no namedValue's are present.
vals.append(snmp_to_numpy_dict[snmp_type](varBind[1])) vals.append(snmp_to_numpy_dict[snmp_type](varBind[1]))
else: else:
# append the named values string instead of the raw number. # append the named values string instead of the raw number.
vals.append(varBind[1].prettyPrint()) vals.append(varBind[1].prettyPrint())
......
...@@ -13,7 +13,7 @@ class server_imitator: ...@@ -13,7 +13,7 @@ class server_imitator:
snmp_to_numpy_dict = { snmp_to_numpy_dict = {
hlapi.Integer32: numpy.int64, hlapi.Integer32: numpy.int64,
hlapi.TimeTicks: numpy.int64, hlapi.TimeTicks: numpy.int64,
str: str, hlapi.OctetString: str,
hlapi.Counter32: numpy.int64, hlapi.Counter32: numpy.int64,
hlapi.Gauge32: numpy.int64, hlapi.Gauge32: numpy.int64,
hlapi.IpAddress: str, hlapi.IpAddress: str,
...@@ -35,6 +35,8 @@ class server_imitator: ...@@ -35,6 +35,8 @@ class server_imitator:
read_val = (None, snmp_type("1.3.6.1.2.1.1.1.0")) read_val = (None, snmp_type("1.3.6.1.2.1.1.1.0"))
elif snmp_type is hlapi.IpAddress: elif snmp_type is hlapi.IpAddress:
read_val = (None, snmp_type("1.1.1.1")) read_val = (None, snmp_type("1.1.1.1"))
elif snmp_type is hlapi.OctetString:
read_val = (None, snmp_type("1"))
else: else:
read_val = (None, snmp_type(1)) read_val = (None, snmp_type(1))
...@@ -48,6 +50,10 @@ class server_imitator: ...@@ -48,6 +50,10 @@ class server_imitator:
read_val = [] read_val = []
for _i in range(dims[0]): for _i in range(dims[0]):
read_val.append((None, snmp_type(f"1.1.1.1"))) read_val.append((None, snmp_type(f"1.1.1.1")))
elif snmp_type is hlapi.OctetString:
read_val = []
for _i in range(dims[0]):
read_val.append((None, snmp_type("1")))
else: else:
read_val = [] read_val = []
for _i in range(dims[0]): for _i in range(dims[0]):
...@@ -68,7 +74,7 @@ class server_imitator: ...@@ -68,7 +74,7 @@ class server_imitator:
check_val = "1.3.6.1.2.1.1.1.0.1" check_val = "1.3.6.1.2.1.1.1.0.1"
elif snmp_type is hlapi.IpAddress: elif snmp_type is hlapi.IpAddress:
check_val = "1.1.1.1" check_val = "1.1.1.1"
elif snmp_type is str: elif snmp_type is hlapi.OctetString:
check_val = "1" check_val = "1"
else: else:
check_val = 1 check_val = 1
...@@ -78,7 +84,7 @@ class server_imitator: ...@@ -78,7 +84,7 @@ class server_imitator:
elif snmp_type is hlapi.IpAddress: elif snmp_type is hlapi.IpAddress:
check_val = ["1.1.1.1"] * dims[0] check_val = ["1.1.1.1"] * dims[0]
elif snmp_type is str: elif snmp_type is hlapi.OctetString:
check_val = ["1"] * dims[0] check_val = ["1"] * dims[0]
else: else:
check_val = [1] * dims[0] check_val = [1] * dims[0]
...@@ -128,7 +134,7 @@ class TestSNMP(base.TestCase): ...@@ -128,7 +134,7 @@ class TestSNMP(base.TestCase):
val = snmp_attr.read_function() val = snmp_attr.read_function()
checkval = server.val_check(i, server.dim_list[j]) checkval = server.val_check(i, server.dim_list[j])
self.assertEqual(checkval, val, f"Expected: {checkval}, got: {val}") self.assertEqual(checkval, val, f"During test {j} {i}; Expected: {checkval} of type {i}, got: {val} of type {type(val)}")
@mock.patch('pysnmp.hlapi.ObjectIdentity') @mock.patch('pysnmp.hlapi.ObjectIdentity')
@mock.patch('pysnmp.hlapi.ObjectType') @mock.patch('pysnmp.hlapi.ObjectType')
...@@ -163,5 +169,6 @@ class TestSNMP(base.TestCase): ...@@ -163,5 +169,6 @@ class TestSNMP(base.TestCase):
res_lst = res_lst[0] res_lst = res_lst[0]
checkval = server.val_check(i, server.dim_list[j]) checkval = server.val_check(i, server.dim_list[j])
self.assertEqual(checkval, res_lst, f"Expected: {checkval}, got: {res_lst}") self.assertEqual(checkval, res_lst, f"During test {j} {i}; Expected: {checkval}, got: {res_lst}")
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment