diff --git a/src/sonic_ax_impl/main.py b/src/sonic_ax_impl/main.py index 281ab601b..ad062aa04 100644 --- a/src/sonic_ax_impl/main.py +++ b/src/sonic_ax_impl/main.py @@ -11,7 +11,7 @@ import ax_interface from sonic_ax_impl.mibs import ieee802_1ab, Namespace from . import logger -from .mibs.ietf import rfc1213, rfc2737, rfc2863, rfc3433, rfc4292, rfc4363 +from .mibs.ietf import rfc1213, rfc2737, rfc2863, rfc3433, rfc4292, rfc4363, rfc2790 from .mibs.vendor import dell, cisco # Background task update frequency ( in seconds ) @@ -41,6 +41,8 @@ class SonicMIB( cisco.ciscoPfcExtMIB.cpfcIfPriorityTable, cisco.ciscoSwitchQosMIB.csqIfQosGroupStatsTable, cisco.ciscoEntityFruControlMIB.cefcFruPowerStatusTable, + rfc2790.hrStorageTable, + rfc2790.hrFSTable, ): """ If SONiC was to create custom MIBEntries, they may be specified here. diff --git a/src/sonic_ax_impl/mibs/ietf/rfc2790.py b/src/sonic_ax_impl/mibs/ietf/rfc2790.py new file mode 100644 index 000000000..adfcb43c9 --- /dev/null +++ b/src/sonic_ax_impl/mibs/ietf/rfc2790.py @@ -0,0 +1,279 @@ +from enum import Enum, unique +from sonic_ax_impl import mibs +from ax_interface import MIBMeta, ValueType, SubtreeMIBEntry +import sonic_ax_impl.mibs.ietf.physical_entity_sub_oid_generator as fru_oids +import re + +@unique +class hrFSinfoDB(str, Enum): + """ + hr FS info keys + """ + TYPE = "Type" + MOUNT = "MountPoint" + +def get_FS_data(fs_info): + """ + :param chassis_info: chassis info dict + :return: tuple of chassis; + Empty string if field not in chassis_info + """ + return tuple(fs_info.get(field.value, "") for field in hrFSinfoDB) + +class fsHandler: + """ + Class to handle the SNMP request + """ + def __init__(self): + """ + init the handler + """ + self.statedb = mibs.init_db() + self.statedb.connect(self.statedb.STATE_DB) + self.init_fs() + + def init_fs(self): + fs_entries = self.statedb.keys(self.statedb.STATE_DB, + 'MOUNT_POINTS' + mibs.TABLE_NAME_SEPARATOR_VBAR + '*') + if not fs_entries: + mibs.logger.debug('No mount point found in {}'.format(fs_entries)) + return None + fs_entries = sorted(fs_entries) + self.fs_entries = fs_entries + + def get_next(self, sub_id): + """ + :param sub_id: The 1-based snmp sub-identifier query. + :return: the next sub id. + """ + if not sub_id: + self.init_fs() + return (1, ) + + index = sub_id[0] + if index >= len(self.fs_entries): + return None + + return (index + 1,) + + + # Type of FS entry + def _get_fs_type(self, oid): + """ + :return: Type of requested sub_id according to hrFSTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + fs_name = self.fs_entries[oid] + fs_info = self.statedb.get_all(self.statedb.STATE_DB, fs_name) + fs_type, fs_mount = get_FS_data(fs_info) + mibs.logger.debug('fs storage info {} name {} type {}'.format(oid, fs_name, fs_type)) + + return fs_type + + def get_fs_type(self, sub_id): + """ + :param sub_id: The 1-based sub-identifier query. Only iterate through the MOUNTPOINT + :return: Type of requested sub_id according to hrFSTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + if not sub_id: + return None + + return self._get_fs_type(sub_id[0]-1) + + # Mountpoint of FS entry + def _get_fs_mount(self, oid): + """ + :return: Mountpoint of requested sub_id according to hrFSTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + fs_name = self.fs_entries[oid] + fs_info = self.statedb.get_all(self.statedb.STATE_DB, fs_name) + + mount_str = fs_name.split('|')[1] + mibs.logger.debug('fs storage info {} name {} mountpoint {}'.format(oid, fs_name, mount_str)) + ret_str = mount_str.encode('utf-8') + return ret_str + + def get_fs_mount(self, sub_id): + """ + :param sub_id: The 1-based sub-identifier query. Only iterate through the MOUNTPOINT + :return: Mountpoint of requested sub_id according to hrFSTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + if not sub_id: + return None + return self._get_fs_mount(sub_id[0]-1) + + +class hrFSTable(metaclass=MIBMeta, prefix='.1.3.6.1.2.1.25.3.8'): + """ + 'hrFSTable' http://oidref.com/1.3.6.1.2.1.25.3.8 + """ + handler = fsHandler() + fsMountpoint = SubtreeMIBEntry('1.2', handler, ValueType.OCTET_STRING, handler.get_fs_mount) + fsType = SubtreeMIBEntry('1.4', handler, ValueType.OCTET_STRING, handler.get_fs_type) + + +# ======================= +@unique +class hrStorageInfoDB(str, Enum): + """ + hr Storage info keys + """ + KBLOCKS = "1K-blocks" + USED = "Used" + FILESYSTEM = "Filesystem" + +def get_hrStorage_data(hrStorage_info): + """ + :param chassis_info: chassis info dict + :return: tuple of chassis; + Empty string if field not in chassis_info + """ + return tuple(hrStorage_info.get(field.value, "") for field in hrStorageInfoDB) + +class hrStorageHandler: + """ + Class to handle the SNMP request + """ + def __init__(self): + """ + init the handler + """ + self.statedb = mibs.init_db() + self.statedb.connect(self.statedb.STATE_DB) + self.init_hr_storage() + + def init_hr_storage(self): + hr_storage_entries = self.statedb.keys(self.statedb.STATE_DB, + 'MOUNT_POINTS' + mibs.TABLE_NAME_SEPARATOR_VBAR + '*') + if not hr_storage_entries: + mibs.logger.debug('No mount point found in {}'.format(hr_storage_entries)) + return None + hr_storage_entries = sorted(hr_storage_entries) + self.hr_storage_entries = hr_storage_entries + + def get_next(self, sub_id): + """ + :param sub_id: The 1-based snmp sub-identifier query. + :return: the next sub id. + """ + if not sub_id: + self.init_hr_storage() + return (1, ) + + index = sub_id[0] + if index >= len(self.hr_storage_entries): + return None + + return (index + 1,) + + + # Used space of HRStorage + def _get_hrstorage_used(self, oid): + """ + :return: Used space of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + hrstorage_name = self.hr_storage_entries[oid] + hrstorage_info = self.statedb.get_all(self.statedb.STATE_DB, hrstorage_name) + kblocks, used, filesystem = get_hrStorage_data(hrstorage_info) + mibs.logger.debug('hr storage info {} name {} used {}'.format(oid, hrstorage_name, used)) + + used_num = int(used) + return used_num + + def get_hrstorage_used(self, sub_id): + """ + :param sub_id: The 1-based sub-identifier query. Only iterate through the MOUNTPOINT + :return: Used space of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + if not sub_id: + return None + + return self._get_hrstorage_used(sub_id[0]-1) + + + """ + get_hrstorage_size will return 65536 if the output of 1k blocks (in df -T) is 65536 1k-blocks. + hrStorageAllocationUnits (1.3.6.1.2.1.25.2.3.1.4) is an int_32 represents the size in bytes + hrStorageSize (1.3.6.1.2.1.25.2.3.1.5) is an int_32 represents the size of storage in units of hrStorageAllocationUnits + """ + # Size of HRStorage + def _get_hrstorage_size(self, oid): + """ + :return: Size of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + hrstorage_name = self.hr_storage_entries[oid] + hrstorage_info = self.statedb.get_all(self.statedb.STATE_DB, hrstorage_name) + kblocks, used, filesystem = get_hrStorage_data(hrstorage_info) + mibs.logger.debug('hr storage info {} name {} size {}'.format(oid, hrstorage_name, kblocks)) + + size_num = int(kblocks) + return size_num + + def get_hrstorage_size(self, sub_id): + """ + :param sub_id: The 1-based sub-identifier query. Only iterate through the MOUNTPOINT + :return: Size of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + if not sub_id: + return None + return self._get_hrstorage_size(sub_id[0]-1) + + + # Description of the filesystem of HRStorage + def _get_hrstorage_descr(self, oid): + """ + :return: Description of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + hrstorage_name = self.hr_storage_entries[oid] + hrstorage_info = self.statedb.get_all(self.statedb.STATE_DB, hrstorage_name) + kblocks, used, filesystem = get_hrStorage_data(hrstorage_info) + mibs.logger.debug('hr storage info {} name {} filesystem {}'.format(oid, hrstorage_name, filesystem)) + return filesystem + + def get_hrstorage_descr (self, sub_id): + """ + :param sub_id: The 1-based sub-identifier query. Only iterate through the MOUNTPOINT + :return: Descr of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + if not sub_id: + return None + return self._get_hrstorage_descr(sub_id[0]-1) + + + # Alloc units of the filesystem of HRStorage + def _get_hrstorage_alloc(self, oid): + """ + :return: Allocation units of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + return 1024 + + def get_hrstorage_alloc(self, sub_id): + """ + :param sub_id: The 1-based sub-identifier query. Only iterate through the MOUNTPOINT + :return: Alloc units of requested sub_id according to hrStorageTable + :ref: https://mibbrowser.online/mibdb_search.php?mib=HOST-RESOURCES-V2-MIB + """ + if not sub_id: + return None + return self._get_hrstorage_alloc(sub_id[0]-1) + + +class hrStorageTable(metaclass=MIBMeta, prefix='.1.3.6.1.2.1.25.2.3'): + """ + 'hrStorageTable' http://oidref.com/1.3.6.1.2.1.25.2.3 + """ + handler = hrStorageHandler() + hrStorageDescr = SubtreeMIBEntry('1.3', handler, ValueType.OCTET_STRING, handler.get_hrstorage_descr) + hrStorageAlloc = SubtreeMIBEntry('1.4', handler, ValueType.INTEGER, handler.get_hrstorage_alloc) + hrStorageSize = SubtreeMIBEntry('1.5', handler, ValueType.INTEGER, handler.get_hrstorage_size) + hrStorageUsed = SubtreeMIBEntry('1.6', handler, ValueType.INTEGER, handler.get_hrstorage_used) diff --git a/tests/mock_tables/state_db.json b/tests/mock_tables/state_db.json index 7de231f3d..287b95567 100644 --- a/tests/mock_tables/state_db.json +++ b/tests/mock_tables/state_db.json @@ -148,6 +148,27 @@ "temperature": "20.5", "is_replaceable": "False" }, + "MOUNT_POINTS|/dev": { + "Filesystem": "/host", + "Type": "ext4", + "1K-blocks": "52829740", + "Used": "0", + "Available": "52829740" + }, + "MOUNT_POINTS|/dev2": { + "Filesystem": "/host2", + "Type": "ext3", + "1K-blocks": "12345", + "Used": "10", + "Available": "12335" + }, +"MOUNT_POINTS|/dev3": { +"Filesystem": "/host3", +"Type": "nfs", +"1K-blocks": "1000", +"Used": "100", +"Available": "900" +}, "NEIGH_STATE_TABLE|10.0.0.57": { "state" : "6402" }, diff --git a/tests/test_rfc2790_fs.py b/tests/test_rfc2790_fs.py new file mode 100644 index 000000000..f91b210d4 --- /dev/null +++ b/tests/test_rfc2790_fs.py @@ -0,0 +1,223 @@ +import os +import sys + +# noinspection PyUnresolvedReferences +import tests.mock_tables.dbconnector + +modules_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(modules_path, 'src')) + +from unittest import TestCase + +from ax_interface import ValueType +from ax_interface.pdu_implementations import GetPDU, GetNextPDU +from ax_interface.encodings import ObjectIdentifier +from ax_interface.constants import PduTypes +from ax_interface.pdu import PDU, PDUHeader +from ax_interface.mib import MIBTable +from sonic_ax_impl.mibs.ietf import rfc2790 + +class TestMountpoints_fs(TestCase): + @classmethod + def setUpClass(cls): + cls.lut = MIBTable(rfc2790.hrFSTable) + + # ======= Filesystem Type ======= + def test_getNextFSType0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 1)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "ext4") + + def test_getFSType0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 1)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "ext4") + + def test_getNextFSType1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 1)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 2)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "ext3") + + def test_getFSType1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 2)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "ext3") + + def test_getNextFSType2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 3)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "nfs") + + def test_getFSType2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 3)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "nfs") + + # ======= Filesystem Mount ======= + def test_getNextMountFS0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 1)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/dev") + + def test_getMountFS0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 1)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/dev") + + def test_getNextMountFS1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 1)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 2)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/dev2") + + def test_getMountFS1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 2)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/dev2") + + def test_getNextMountFS2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 3)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/dev3") + + def test_getMountFS2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 3)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/dev3") diff --git a/tests/test_rfc2790_hrstorage.py b/tests/test_rfc2790_hrstorage.py new file mode 100644 index 000000000..1b2a1d9a6 --- /dev/null +++ b/tests/test_rfc2790_hrstorage.py @@ -0,0 +1,407 @@ +import os +import sys + +# noinspection PyUnresolvedReferences +import tests.mock_tables.dbconnector + +modules_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(modules_path, 'src')) + +from unittest import TestCase + +from ax_interface import ValueType +from ax_interface.pdu_implementations import GetPDU, GetNextPDU +from ax_interface.encodings import ObjectIdentifier +from ax_interface.constants import PduTypes +from ax_interface.pdu import PDU, PDUHeader +from ax_interface.mib import MIBTable +from sonic_ax_impl.mibs.ietf import rfc2790 + +class TestMountpoints(TestCase): + @classmethod + def setUpClass(cls): + cls.lut = MIBTable(rfc2790.hrStorageTable) + + # ====== Testing Used ====== + def test_getNextHrStorageUsed0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 1)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 0) + + def test_getHrStorageUsed0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 1)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 0) + + def test_getNextHrStorageUsed1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 1)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 2)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 10) + + def test_getHrStorageUsed1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 2)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 10) + + def test_getNextHrStorageUsed2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 3)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 100) + + def test_getHrStorageUsed2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 3)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 100) + + # ====== Testing Size ====== + def test_getNextHrStorageSize0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 1)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 52829740) + + def test_getHrStorageSize0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 1)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 52829740) + + def test_getNextHrStorageSize1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 1)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 2)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 12345) + + def test_getHrStorageSize1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 2)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 12345) + + def test_getNextHrStorageSize2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 3)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 1000) + + def test_getHrStorageSize2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 3)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 1000) + + # ====== Testing Description ====== + def test_getNextHrStorageDescr0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 1)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/host") + + def test_getHrStorageDescr0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 1)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/host") + + def test_getNextHrStorageDescr1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 1)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 2)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/host2") + + def test_getHrStorageDescr1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 2)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/host2") + + def test_getNextHrStorageDescr2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 3)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(expected_oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/host3") + + def test_getHrStorageDescr2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 3)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.OCTET_STRING) + self.assertEqual(str(value0.name), str(oid)) + decoded_str = value0.data.string.decode('utf-8') + self.assertEqual(decoded_str, "/host3") + + + # ====== Testing Allocation ====== + def test_getNextHrStorageAlloc0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 1)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 1024) + + def test_getHrStorageAlloc0(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 1)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 1024) + + def test_getNextHrStorageAlloc1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 1)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 2)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 1024) + + def test_getHrStorageAlloc1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 2)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 1024) + + def test_getNextHrStorageAlloc2(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 2)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 3)) + get_pdu = GetNextPDU( + header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(expected_oid)) + self.assertEqual(value0.data, 1024) + + def test_getHrStorageAlloc1(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 3)) + get_pdu = GetPDU( + header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0), + oids=[oid] + ) + + encoded = get_pdu.encode() + response = get_pdu.make_response(self.lut) + + value0 = response.values[0] + self.assertEqual(value0.type_, ValueType.INTEGER) + self.assertEqual(str(value0.name), str(oid)) + self.assertEqual(value0.data, 1024) +