diff --git a/tangostationcontrol/tangostationcontrol/clients/snmp_client.py b/tangostationcontrol/tangostationcontrol/clients/snmp_client.py
index 0b2bbc0480fb6d763e8b415778009f5aca7c40f1..28c0e2a64200d27cb4d5a8f3b20275507b1abdf6 100644
--- a/tangostationcontrol/tangostationcontrol/clients/snmp_client.py
+++ b/tangostationcontrol/tangostationcontrol/clients/snmp_client.py
@@ -20,6 +20,7 @@ snmp_to_numpy_dict = {
hlapi.IpAddress: str,
}
+
class SNMP_client(CommClient):
"""
messages to keep a check on the connection. On connection failure, reconnects once.
@@ -62,35 +63,8 @@ class SNMP_client(CommClient):
index (optional) the index if the value thats being read from is a table.
"""
- # return values start as None because we have a way too complicated interface
- oids = None
- mib = None
- name = None
- idx = None
-
- # check if the 'oids' key is used and not the 'mib' and 'name' keys
-
- if 'oids' in annotation and 'mib' not in annotation and 'name' not in annotation:
- uses_oid = True
- oids = annotation["oids"]
-
- # checks to make sure this isn't present
- if 'index' in annotation:
- raise ValueError(f"SNMP attribute annotation doesn't support oid type declarations with an index present.")
-
-
- # check if the 'oids' key is NOT used but instead the 'mib' and 'name' keys
- elif 'oids' not in annotation and 'mib' in annotation and 'name' in annotation:
- mib = annotation["mib"]
- name = annotation["name"]
-
- # SNMP has tables that require an index number to access them. regular non-table variable have an index of 0
- idx = annotation.get('index', 0)
-
- else:
- raise ValueError(f"SNMP attribute annotation requires a dict argument with either a 'oids' key or both a 'name' and 'mib' key. Not both. Instead got: {annotation}")
-
- return oids, mib, name, idx
+ wrapper = annotation_wrapper(annotation)
+ return wrapper
def setup_value_conversion(self, attribute):
"""
@@ -113,11 +87,11 @@ class SNMP_client(CommClient):
"""
# process the annotation
- oids, mib, name, idx = self._setup_annotation(annotation)
+ wrapper = self._setup_annotation(annotation)
# get all the necessary data to set up the read/write functions from the attribute_wrapper
dim_x, dim_y, dtype = self.setup_value_conversion(attribute)
- snmp_attr = snmp_attribute(self, oids, mib, name, idx, dtype, dim_x, dim_y)
+ snmp_attr = snmp_attribute(self, wrapper, dtype, dim_x, dim_y)
# return the read/write functions
def read_function():
@@ -129,45 +103,125 @@ class SNMP_client(CommClient):
return read_function, write_function
-class snmp_attribute:
+class annotation_wrapper:
+ def __init__(self, annotation):
+ """
+ The SNMP client uses a dict and takes the following keys:
- def __init__(self, client : SNMP_client, oids, mib, name, idx, dtype, dim_x, dim_y):
+ either
+ oids: Required. An oid string of the object
+ or
+ mib: the mib name
+ name: name of the value to read
+ index (optional) the index if the value thats being read from is a table.
+ """
- self.client = client
- self.dtype = dtype
- self.dim_x = dim_x
- self.dim_y = dim_y
- self.is_scalar = (self.dim_x + self.dim_y) == 1
+ # values start as None because we have a way too complicated interface
+ self.oids = None
+ self.mib = None
+ self.name = None
+ self.idx = None
+
+ # check if the 'oids' key is used and not the 'mib' and 'name' keys
+
+ if 'oids' in annotation and 'mib' not in annotation and 'name' not in annotation:
+ self.oids = annotation["oids"]
+
+ # checks to make sure this isn't present
+ if 'index' in annotation:
+ raise ValueError(f"SNMP attribute annotation doesn't support oid type declarations with an index present.")
+
+
+ # check if the 'oids' key is NOT used but instead the 'mib' and 'name' keys
+ elif 'oids' not in annotation and 'mib' in annotation and 'name' in annotation:
+ self.mib = annotation["mib"]
+ self.name = annotation["name"]
+
+ # SNMP has tables that require an index number to access them. regular non-table variable have an index of 0
+ self.idx = annotation.get('index', 0)
+
+ else:
+ raise ValueError(
+ f"SNMP attribute annotation requires a dict argument with either a 'oids' key or both a 'name' and 'mib' key. Not both. Instead got: {annotation}")
+
+ def create_objID(self, x, y):
+ is_scalar = (x + y) == 1
# if oids are used
- if oids is not None:
+ if self.oids is not None:
# get a list of str of the oids
- oids = self.get_oids(dim_x, dim_y, oids)
+ self.oids = self._get_oids(x, y, self.oids)
# turn the list of oids in to a tuple of pysnmp object identities. These are used for the
- objID = tuple(hlapi.ObjectIdentity(oids[i]) for i in range(len(oids)))
+ objID = tuple(hlapi.ObjectIdentity(self.oids[i]) for i in range(len(self.oids)))
# if mib + name is used
else:
# only scalars can be used at the present time.
- if not self.is_scalar:
- #tuple(hlapi.ObjectIdentity(mib, name, idx) for i in range(len(oids)))
+ if not is_scalar:
+ # tuple(hlapi.ObjectIdentity(mib, name, idx) for i in range(len(oids)))
- raise ValueError(f"MIB + name type attributes can only be scalars, got dimensions of: ({self.dim_y}, {self.dim_x})")
+ raise ValueError(f"MIB + name type attributes can only be scalars, got dimensions of: ({x}, {y})")
else:
- objID = hlapi.ObjectIdentity(mib, name, idx)
+ objID = hlapi.ObjectIdentity(self.mib, self.name, self.idx)
+
+ return objID
+
+ def _get_oids(self, x, y, in_oid):
- self.objID = objID
+ if x == 0:
+ x = 1
+ if y == 0:
+ y = 1
+
+ is_scalar = (x * y) == 1
+ nof_oids = x * y
+
+ # if scalar
+ if is_scalar:
+ if type(in_oid) is str:
+ # for ease of handling put single oid in a 1 element list
+ in_oid = [in_oid]
+
+ return in_oid
+
+ else:
+ # if we got a single str oid, make a list of sequential oids
+ if type(in_oid) is str:
+ return ["{}.{}".format(in_oid, i + 1) for i in range(nof_oids)]
+ # if its an already expanded list of all oids
+ elif type(in_oid) is list and len(in_oid) == nof_oids:
+ # already is an list and of the right length
+ return in_oid
+ # if its a list of attributes with the wrong length.
+ else:
+ raise ValueError(
+ "SNMP oids need to either be a single value or an array the size of the attribute dimensions. got: {} expected: {}x{}={}".format(
+ len(in_oid), x, y, x * y))
+
+
+class snmp_attribute:
+
+ def __init__(self, client : SNMP_client, wrapper, dtype, dim_x, dim_y):
+
+ self.client = client
+ self.wrapper = wrapper
+ self.dtype = dtype
+ self.dim_x = dim_x
+ self.dim_y = dim_y
+ self.is_scalar = (self.dim_x + self.dim_y) == 1
+
+ self.objID = self.wrapper.create_objID(self.dim_x, self.dim_y)
def next_wrap(self, cmd):
"""
This function exists to allow the next(cmd) call to be mocked for unit testing. As the
"""
-
return next(cmd)
def read_function(self):
+ # must be recreated for each read it seems
self.objs = tuple(hlapi.ObjectType(i) for i in self.objID)
# get the thingy to get the values
@@ -193,37 +247,6 @@ class snmp_attribute:
set_cmd = hlapi.setCmd(self.client.engine, self.client.community, self.client.trasport, self.client.ctx_data, *write_obj)
errorIndication, errorStatus, errorIndex, *varBinds = self.next_wrap(set_cmd)
-
- def get_oids(self, x, y, in_oid):
-
- if x == 0:
- x = 1
- if y == 0:
- y = 1
-
- nof_oids = x * y
-
- # if scalar
- if self.is_scalar:
- if type(in_oid) is str:
- # for ease of handling put single oid in a 1 element list
- in_oid = [in_oid]
-
- return in_oid
-
- else:
- # if we got a single str oid, make a list of sequential oids
- if type(in_oid) is str:
- return ["{}.{}".format(in_oid, i + 1) for i in range(nof_oids)]
- # if its an already expanded list of all oids
- elif type(in_oid) is list and len(in_oid) == nof_oids:
- # already is an list and of the right length
- return in_oid
- # if its a list of attributes with the wrong length.
- else:
- raise ValueError("SNMP oids need to either be a single value or an array the size of the attribute dimensions. got: {} expected: {}x{}={}".format(len(in_oid),x,y,x*y))
-
-
def convert(self, varBinds):
"""
get all the values in a list, make sure to convert specific types that dont want to play nicely
diff --git a/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py b/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py
index 2bb4cd60f28bdd1be32924f85dea9b64dd49f5b2..05da065b95b34387da517a66083dd6c672994643 100644
--- a/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py
+++ b/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py
@@ -5,7 +5,7 @@ from unittest import mock
from tangostationcontrol.test import base
-from tangostationcontrol.clients.snmp_client import SNMP_client, snmp_attribute
+from tangostationcontrol.clients.snmp_client import SNMP_client, snmp_attribute, annotation_wrapper
snmp_to_numpy_dict = {
hlapi.Integer32: numpy.int64,
@@ -81,60 +81,61 @@ def val_check(snmp_type : type, dims : tuple):
class TestSNMP(base.TestCase):
- @mock.patch('pysnmp.hlapi.ObjectIdentity')
- @mock.patch('pysnmp.hlapi.ObjectType')
- @mock.patch('tangostationcontrol.clients.snmp_client.snmp_attribute.next_wrap')
- def test_snmp_obj_get(self, m_next, m_obj_T, m_obj_i):
+
+ def test_annotation_success(self):
"""
- Attempts to read a fake SNMP variable and checks whether it got what it expected
+ unit test for the processing of annotation. Has 2 lists. 1 with things that should succeed and 1 with things that should fail.
"""
- for j in dim_list:
- for i in snmp_to_numpy_dict:
- m_next.return_value = (None, None, None, get_return_val(i, dim_list[j]))
-
- m_client = mock.Mock()
+ test_name = "test_name"
+ test_mib = "test_mib"
- a = snmp_attribute(client=m_client, oids="1.3.6.1.2.1.2.2.1.2.31", mib=None, name=None, idx=None, dtype=snmp_to_numpy_dict[i], dim_x=dim_list[j][0], dim_y=dim_list[j][1])
- val = a.read_function()
+ client = SNMP_client(community='public', host='localhost', timeout=10, fault_func=None, try_interval=2)
- checkval = val_check(i, dim_list[j])
- self.assertEqual(checkval, val, f"Expected: {checkval}, got: {val}")
+ test_list = [
+ # test name nad MIB type annotation
+ {"mib": "SNMPv2-MIB", "name": "sysDescr"},
- @mock.patch('pysnmp.hlapi.ObjectIdentity')
- @mock.patch('pysnmp.hlapi.setCmd')
- @mock.patch('tangostationcontrol.clients.snmp_client.snmp_attribute.next_wrap')
- def test_snmp_obj_set(self, m_next, m_nextCmd, m_obj_i):
- """
- Attempts to write a value to an SNMP server, but instead intercepts it and compared whether the values is as expected.
- """
+ # test name nad MIB type annotation with index
+ {"mib": "RFC1213-MIB", "name": "ipAdEntAddr", "index": (127, 0, 0, 1)},
+ {"mib": "random-MIB", "name": "aName", "index": 2},
- for j in dim_list:
- for i in snmp_to_numpy_dict:
- m_next.return_value = (None, None, None, get_return_val(i, dim_list[j]))
+ #oid
+ {"oids": "1.3.6.1.2.1.2.2.1.2.31"}
+ ]
- m_client = mock.Mock()
- set_val = val_check(i, dim_list[j])
+ for i in test_list:
+ wrapper = client._setup_annotation(annotation=i)
- a = snmp_attribute(client=m_client, oids="1.3.6.1.2.1.2.2.1.2.31", mib=None, name=None, idx=None, dtype=snmp_to_numpy_dict[i], dim_x=dim_list[j][0],
- dim_y=dim_list[j][1])
+ if wrapper.oids is not None:
+ self.assertEqual(wrapper.oids, i["oids"])
- res_lst = []
- def test(*value):
- res_lst.append(value[1])
- return None, None, None, get_return_val(i, dim_list[j])
+ else:
+ self.assertEqual(wrapper.mib, i["mib"], f"expected mib with: {i['mib']}, got: {wrapper.idx} from: {i}")
+ self.assertEqual(wrapper.name, i["name"], f"expected name with: {i['name']}, got: {wrapper.idx} from: {i}")
+ self.assertEqual(wrapper.idx, i.get('index', 0), f"expected idx with: {i.get('index', 0)}, got: {wrapper.idx} from: {i}")
- hlapi.ObjectType = test
- a.write_function(set_val)
+ def test_annotation_fail(self):
+ """
+ unit test for the processing of annotation. Has 2 lists. 1 with things that should succeed and 1 with things that should fail.
+ """
- if len(res_lst) == 1:
- res_lst = res_lst[0]
+ client = SNMP_client(community='public', host='localhost', timeout=10, fault_func=None, try_interval=2)
- checkval = val_check(i, dim_list[j])
- self.assertEqual(checkval, res_lst, f"Expected: {checkval}, got: {res_lst}")
+ fail_list = [
+ # OIDS cant use the index
+ {"oids": "1.3.6.1.2.1.2.2.1.2.31", "index": 2},
+ # mixed annotation is not allowed
+ {"oids": "1.3.6.1.2.1.2.2.1.2.31", "name": "thisShouldFail"},
+ # no 'name'
+ {"mib": "random-MIB", "index": 2},
+ ]
+ for i in fail_list:
+ with self.assertRaises(ValueError):
+ client._setup_annotation(annotation=i)
def test_oids_scalar(self):
@@ -145,12 +146,10 @@ class TestSNMP(base.TestCase):
m_client = mock.Mock()
# we just need the object to call another function
- a = snmp_attribute(client=m_client, oids="Not None", mib=None, name=None, idx=None, dtype=str, dim_x=x, dim_y=y)
-
+ wrapper = annotation_wrapper(annotation = {"oids": "Not None lol"})
# scalar
scalar_expected = [test_oid]
- scalar_dims = (x, y)
- ret_oids = a.get_oids(scalar_dims[0], scalar_dims[1], test_oid)
+ ret_oids = wrapper._get_oids(x, y, test_oid)
self.assertEqual(ret_oids, scalar_expected, f"Expected: {scalar_expected}, got: {ret_oids}")
def test_oids_spectrum(self):
@@ -167,66 +166,68 @@ class TestSNMP(base.TestCase):
m_client = mock.Mock()
# we just need the object to call another function
- a = snmp_attribute(client=m_client, oids="Not None", mib=None, name=None, idx=None, dtype=str, dim_x=x, dim_y=y)
+ wrapper = annotation_wrapper(annotation={"oids": "Not None lol"})
# spectrum
spectrum_expected = [test_oid + ".1", test_oid + ".2", test_oid + ".3", test_oid + ".4"]
- spectrum_dims = (x, y)
- ret_oids = a.get_oids(spectrum_dims[0], spectrum_dims[1], test_oid)
+ ret_oids = wrapper._get_oids(x, y, test_oid)
self.assertListEqual(ret_oids, spectrum_expected, f"Expected: {spectrum_expected}, got: {ret_oids}")
-
- def test_annotation_success(self):
+ @mock.patch('pysnmp.hlapi.ObjectIdentity')
+ @mock.patch('pysnmp.hlapi.ObjectType')
+ @mock.patch('tangostationcontrol.clients.snmp_client.snmp_attribute.next_wrap')
+ def test_snmp_obj_get(self, m_next, m_obj_T, m_obj_i):
"""
- unit test for the processing of annotation. Has 2 lists. 1 with things that should succeed and 1 with things that should fail.
+ Attempts to read a fake SNMP variable and checks whether it got what it expected
"""
- test_name = "test_name"
- test_mib = "test_mib"
+ for j in dim_list:
+ for i in snmp_to_numpy_dict:
+ m_next.return_value = (None, None, None, get_return_val(i, dim_list[j]))
- client = SNMP_client(community='public', host='localhost', timeout=10, fault_func=None, try_interval=2)
+ m_client = mock.Mock()
- test_list = [
- # test name nad MIB type annotation
- {"mib": "SNMPv2-MIB", "name": "sysDescr"},
+ wrapper = annotation_wrapper(annotation={"oids": "1.3.6.1.2.1.2.2.1.2.31"})
+ a = snmp_attribute(client=m_client, wrapper=wrapper, dtype=snmp_to_numpy_dict[i], dim_x=dim_list[j][0], dim_y=dim_list[j][1])
- # test name nad MIB type annotation with index
- {"mib": "RFC1213-MIB", "name": "ipAdEntAddr", "index": (127, 0, 0, 1)},
- {"mib": "random-MIB", "name": "aName", "index": 2},
+ val = a.read_function()
- #oid
- {"oids": "1.3.6.1.2.1.2.2.1.2.31"}
- ]
+ checkval = val_check(i, dim_list[j])
+ self.assertEqual(checkval, val, f"Expected: {checkval}, got: {val}")
+ @mock.patch('pysnmp.hlapi.ObjectIdentity')
+ @mock.patch('pysnmp.hlapi.setCmd')
+ @mock.patch('tangostationcontrol.clients.snmp_client.snmp_attribute.next_wrap')
+ def test_snmp_obj_set(self, m_next, m_nextCmd, m_obj_i):
+ """
+ Attempts to write a value to an SNMP server, but instead intercepts it and compared whether the values is as expected.
+ """
- for i in test_list:
- oids, mib, name, idx = client._setup_annotation(annotation=i)
+ for j in dim_list:
+ for i in snmp_to_numpy_dict:
+ m_next.return_value = (None, None, None, get_return_val(i, dim_list[j]))
- if oids is not None:
- self.assertEqual(oids, i["oids"])
+ m_client = mock.Mock()
+ set_val = val_check(i, dim_list[j])
- else:
- self.assertEqual(mib, i["mib"], f"expected mib with: {i['mib']}, got: {idx} from: {i}")
- self.assertEqual(name, i["name"], f"expected name with: {i['name']}, got: {idx} from: {i}")
- self.assertEqual(idx, i.get('index', 0), f"expected idx with: {i.get('index', 0)}, got: {idx} from: {i}")
+ wrapper = annotation_wrapper(annotation={"oids": "1.3.6.1.2.1.2.2.1.2.31"})
+ a = snmp_attribute(client=m_client, wrapper=wrapper, dtype=snmp_to_numpy_dict[i], dim_x=dim_list[j][0], dim_y=dim_list[j][1])
+ res_lst = []
+ def test(*value):
+ res_lst.append(value[1])
+ return None, None, None, get_return_val(i, dim_list[j])
+
+ hlapi.ObjectType = test
+
+ a.write_function(set_val)
+
+ if len(res_lst) == 1:
+ res_lst = res_lst[0]
+
+ checkval = val_check(i, dim_list[j])
+ self.assertEqual(checkval, res_lst, f"Expected: {checkval}, got: {res_lst}")
- def test_annotation_fail(self):
- """
- unit test for the processing of annotation. Has 2 lists. 1 with things that should succeed and 1 with things that should fail.
- """
- client = SNMP_client(community='public', host='localhost', timeout=10, fault_func=None, try_interval=2)
- fail_list = [
- # OIDS cant use the index
- {"oids": "1.3.6.1.2.1.2.2.1.2.31", "index": 2},
- # mixed annotation is not allowed
- {"oids": "1.3.6.1.2.1.2.2.1.2.31", "name": "thisShouldFail"},
- # no 'name'
- {"mib": "random-MIB", "index": 2},
- ]
- for i in fail_list:
- with self.assertRaises(ValueError):
- client._setup_annotation(annotation=i)