diff --git a/src/sonic_ax_impl/main.py b/src/sonic_ax_impl/main.py index c798ab690..93669137f 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..54b0c3271 --- /dev/null +++ b/src/sonic_ax_impl/mibs/ietf/rfc2790.py @@ -0,0 +1,397 @@ +from enum import Enum, unique +from sonic_ax_impl import mibs +from ax_interface import MIBMeta, MIBUpdater, 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 _db_call(func, context, *args): + """Execute a STATE_DB operation, returning None on access failure.""" + try: + return func(*args) + except Exception as e: + mibs.logger.error('Failed to access STATE_DB for {}: Reason: {}'.format(context, e)) + return None + +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(MIBUpdater): + """ + Class to handle the SNMP request + """ + def __init__(self): + """ + init the handler + """ + super().__init__() + self.fs_entries = [] + self.statedb = mibs.init_db() + _db_call(self.statedb.connect, 'STATE_DB connect', self.statedb.STATE_DB) + self.init_fs() + + def reinit_data(self): + self.init_fs() + + def update_data(self): + pass + + def init_fs(self): + self.fs_entries = [] + + fs_entries = _db_call(self.statedb.keys, 'MOUNT_POINTS', + self.statedb.STATE_DB, 'MOUNT_POINTS' + mibs.TABLE_NAME_SEPARATOR_VBAR + '*') + if fs_entries is None: + fs_entries = [] + elif not fs_entries: + mibs.logger.info('init_fs - No mount point found in STATE_DB MOUNT_POINTS table') + + sorted_fs_entries = sorted(fs_entries) + + for entry in fs_entries: + fs_info = _db_call(self.statedb.get_all, entry, self.statedb.STATE_DB, entry) + if not fs_info: + mibs.logger.info('Failed to get data for entry {}'.format(entry)) + sorted_fs_entries.remove(entry) + continue + fs_type, fs_mount = get_FS_data(fs_info) + if fs_type == "" and fs_mount == "": + sorted_fs_entries.remove(entry) + + physical_memory_entries = _db_call(self.statedb.keys, 'MEMORY_STATS', + self.statedb.STATE_DB, 'MEMORY_STATS' + mibs.TABLE_NAME_SEPARATOR_VBAR + '*') + if physical_memory_entries is None: + physical_memory_entries = [] + elif not physical_memory_entries: + mibs.logger.info('init_fs - No memory stats found in STATE_DB MEMORY_STATS table') + + sorted_fs_entries += physical_memory_entries + self.fs_entries = sorted_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() + if not self.fs_entries: + return None + 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 + """ + if oid >= len(self.fs_entries): + mibs.logger.info('fs_type - Index out of range. Size of fs_entries: {}'.format(len(self.fs_entries))) + return None + fs_name = self.fs_entries[oid] + fs_info = _db_call(self.statedb.get_all, fs_name, self.statedb.STATE_DB, fs_name) + if not fs_info: + mibs.logger.info('get_all() returned empty data for {}'.format(fs_name)) + return None + 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 + """ + if oid >= len(self.fs_entries): + mibs.logger.info('fs_mount - Index out of range. Size of fs_entries: {}'.format(len(self.fs_entries))) + return None + fs_name = self.fs_entries[oid] + fs_info = _db_call(self.statedb.get_all, fs_name, self.statedb.STATE_DB, fs_name) + if not fs_info: + mibs.logger.info('get_all() returned empty data for {}'.format(fs_name)) + return None + + 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') + + if fs_name.split('|')[0] == "MEMORY_STATS": + ret_str = "" + + 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(MIBUpdater): + """ + Class to handle the SNMP request + """ + def __init__(self): + """ + init the handler + """ + super().__init__() + self.hr_storage_entries = [] + self.statedb = mibs.init_db() + _db_call(self.statedb.connect, 'STATE_DB connect', self.statedb.STATE_DB) + self.init_hr_storage() + + def reinit_data(self): + self.init_hr_storage() + + def update_data(self): + pass + + def init_hr_storage(self): + self.hr_storage_entries = [] + + hr_storage_entries = _db_call(self.statedb.keys, 'MOUNT_POINTS', + self.statedb.STATE_DB, 'MOUNT_POINTS' + mibs.TABLE_NAME_SEPARATOR_VBAR + '*') + if hr_storage_entries is None: + hr_storage_entries = [] + elif not hr_storage_entries: + mibs.logger.info('init_hr_storage - No mount point found in STATE_DB MOUNT_POINTS table') + + sorted_hr_storage_entries = sorted(hr_storage_entries) + + for entry in hr_storage_entries: + hr_info = _db_call(self.statedb.get_all, entry, self.statedb.STATE_DB, entry) + if not hr_info: + mibs.logger.info('Failed to get data for entry {}'.format(entry)) + sorted_hr_storage_entries.remove(entry) + continue + hr_block, hr_used, hr_fs = get_hrStorage_data(hr_info) + if hr_block == "" and hr_used == "" and hr_fs == "": + sorted_hr_storage_entries.remove(entry) + + physical_memory_entries = _db_call(self.statedb.keys, 'MEMORY_STATS', + self.statedb.STATE_DB, 'MEMORY_STATS' + mibs.TABLE_NAME_SEPARATOR_VBAR + '*') + if physical_memory_entries is None: + physical_memory_entries = [] + elif not physical_memory_entries: + mibs.logger.info('init_hr_storage - No memory stats found in STATE_DB MEMORY_STATS table') + + sorted_hr_storage_entries += physical_memory_entries + self.hr_storage_entries = sorted_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() + if not self.hr_storage_entries: + return None + 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 + """ + if oid >= len(self.hr_storage_entries): + mibs.logger.info('hrstorage_used - Index out of range. Size of hr_storage_entries: {}'.format(len(self.hr_storage_entries))) + return None + hrstorage_name = self.hr_storage_entries[oid] + hrstorage_info = _db_call(self.statedb.get_all, hrstorage_name, + self.statedb.STATE_DB, hrstorage_name) + if not hrstorage_info: + mibs.logger.info('get_all() returned empty data for {}'.format(hrstorage_name)) + return None + kblocks, used, filesystem = get_hrStorage_data(hrstorage_info) + mibs.logger.debug('hr storage info {} name {} used {}'.format(oid, hrstorage_name, used)) + + if not used: + mibs.logger.info('No "Used" field for {}'.format(hrstorage_name)) + return None + return int(used) + + 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 + """ + if oid >= len(self.hr_storage_entries): + mibs.logger.info('hrstorage_size - Index out of range. Size of hr_storage_entries: {}'.format(len(self.hr_storage_entries))) + return None + hrstorage_name = self.hr_storage_entries[oid] + hrstorage_info = _db_call(self.statedb.get_all, hrstorage_name, + self.statedb.STATE_DB, hrstorage_name) + if not hrstorage_info: + mibs.logger.info('get_all() returned empty data for {}'.format(hrstorage_name)) + return None + kblocks, used, filesystem = get_hrStorage_data(hrstorage_info) + mibs.logger.debug('hr storage info {} name {} size {}'.format(oid, hrstorage_name, kblocks)) + + if not kblocks: + mibs.logger.info('No "1K-blocks" field for {}'.format(hrstorage_name)) + return None + return int(kblocks) + + 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 + """ + if oid >= len(self.hr_storage_entries): + mibs.logger.info('hrstorage_descr - Index out of range. Size of hr_storage_entries: {}'.format(len(self.hr_storage_entries))) + return None + hrstorage_name = self.hr_storage_entries[oid] + hrstorage_info = _db_call(self.statedb.get_all, hrstorage_name, + self.statedb.STATE_DB, hrstorage_name) + if not hrstorage_info: + mibs.logger.info('get_all() returned empty data for {}'.format(hrstorage_name)) + return None + kblocks, used, fs_descr = get_hrStorage_data(hrstorage_info) + mibs.logger.debug('hr storage info {} name {} filesystem {}'.format(oid, hrstorage_name, fs_descr)) + if not fs_descr: + fs_descr = hrstorage_name.split('|')[1] + " Memory" + + return fs_descr + + 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..ebbd20fa9 100644 --- a/tests/mock_tables/state_db.json +++ b/tests/mock_tables/state_db.json @@ -148,6 +148,54 @@ "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" + }, + "MOUNT_POINTS|LastUpdateTime": { + "lastupdate": "2024-09-06 17:28:22.959412" + }, + "MEMORY_STATS|Physical": { + "1K-blocks": "100000", + "Used": "1000" + }, + "MEMORY_STATS|Virtual": { + "1K-blocks": "100000", + "Used": "0" + }, + "MEMORY_STATS|Swap": { + "1K-blocks": "0", + "Used": "0" + }, + "MEMORY_STATS|Cached": { + "1K-blocks": "0", + "Used": "0" + }, + "MEMORY_STATS|Shared": { + "1K-blocks": "200", + "Used": "200" + }, + "MEMORY_STATS|Buffer": { + "1K-blocks": "678", + "Used": "300" + }, "NEIGH_STATE_TABLE|10.0.0.57": { "state" : "6402" }, diff --git a/tests/test_rfc2790_empty_tables.py b/tests/test_rfc2790_empty_tables.py new file mode 100644 index 000000000..130c15a02 --- /dev/null +++ b/tests/test_rfc2790_empty_tables.py @@ -0,0 +1,790 @@ +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 unittest.mock import patch, MagicMock + +from sonic_ax_impl.mibs.ietf import rfc2790 +from sonic_ax_impl import mibs + + +class TestFsHandlerNoMountPoints(TestCase): + """Test fsHandler when no MOUNT_POINTS entries exist in STATE_DB.""" + + def _create_handler_with_empty_keys(self): + """Helper to create an fsHandler with statedb.keys returning None.""" + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = None + handler = rfc2790.fsHandler() + return handler, mock_db + + def test_init_fs_no_mount_points_logs_debug(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = [] + + with patch.object(mibs.logger, 'info') as mock_info: + handler = rfc2790.fsHandler() + mock_info.assert_any_call( + 'init_fs - No mount point found in STATE_DB MOUNT_POINTS table' + ) + + def test_fs_entries_empty_when_no_mount_points(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertEqual(handler.fs_entries, []) + + def test_get_next_none_returns_none_when_no_mount_points(self): + handler, mock_db = self._create_handler_with_empty_keys() + mock_db.keys.return_value = None + result = handler.get_next(None) + self.assertIsNone(result) + + def test_get_next_with_sub_id_returns_none_when_no_mount_points(self): + handler, mock_db = self._create_handler_with_empty_keys() + mock_db.keys.return_value = None + result = handler.get_next((1,)) + self.assertIsNone(result) + + def test_get_fs_type_returns_none_for_none_sub_id(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_fs_type(None)) + + def test_get_fs_mount_returns_none_for_none_sub_id(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_fs_mount(None)) + + +class TestFsHandlerNoMemoryStats(TestCase): + """Test fsHandler when MOUNT_POINTS exist but no MEMORY_STATS entries.""" + + def _keys_side_effect(self, db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + if 'MEMORY_STATS' in pattern: + return [] + return None + + def _create_handler_no_memory_stats(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = self._keys_side_effect + mock_db.get_all.return_value = { + 'Type': 'ext4', 'MountPoint': '/dev' + } + handler = rfc2790.fsHandler() + return handler, mock_db + + def test_init_fs_no_memory_stats_logs_debug(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = self._keys_side_effect + mock_db.get_all.return_value = { + 'Type': 'ext4', 'MountPoint': '/dev' + } + + with patch.object(mibs.logger, 'info') as mock_info: + handler = rfc2790.fsHandler() + mock_info.assert_any_call( + 'init_fs - No memory stats found in STATE_DB MEMORY_STATS table' + ) + + def test_fs_entries_has_mount_points_when_no_memory_stats(self): + handler, _ = self._create_handler_no_memory_stats() + self.assertEqual(handler.fs_entries, ['MOUNT_POINTS|/dev']) + + def test_get_next_none_returns_first_when_no_memory_stats(self): + handler, mock_db = self._create_handler_no_memory_stats() + mock_db.keys.side_effect = self._keys_side_effect + mock_db.get_all.return_value = { + 'Type': 'ext4', 'MountPoint': '/dev' + } + result = handler.get_next(None) + self.assertEqual(result, (1,)) + + def test_get_next_with_sub_id_returns_none_past_end_when_no_memory_stats(self): + handler, _ = self._create_handler_no_memory_stats() + result = handler.get_next((len(handler.fs_entries),)) + self.assertIsNone(result) + + +class TestFsHandlerNeitherMountPointsNorMemoryStats(TestCase): + """Test fsHandler when neither MOUNT_POINTS nor MEMORY_STATS exist.""" + + def test_init_fs_logs_mount_points_missing_and_continues(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = [] + + with patch.object(mibs.logger, 'info') as mock_info: + handler = rfc2790.fsHandler() + mock_info.assert_any_call( + 'init_fs - No mount point found in STATE_DB MOUNT_POINTS table' + ) + mock_info.assert_any_call( + 'init_fs - No memory stats found in STATE_DB MEMORY_STATS table' + ) + + self.assertEqual(handler.fs_entries, []) + + +class TestHrStorageHandlerNoMountPoints(TestCase): + """Test hrStorageHandler when no MOUNT_POINTS entries exist in STATE_DB.""" + + def _create_handler_with_empty_keys(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = None + handler = rfc2790.hrStorageHandler() + return handler, mock_db + + def test_init_hr_storage_no_mount_points_logs_debug(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = [] + + with patch.object(mibs.logger, 'info') as mock_info: + handler = rfc2790.hrStorageHandler() + mock_info.assert_any_call( + 'init_hr_storage - No mount point found in STATE_DB MOUNT_POINTS table' + ) + + def test_hr_storage_entries_empty_when_no_mount_points(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertEqual(handler.hr_storage_entries, []) + + def test_get_next_none_returns_none_when_no_mount_points(self): + handler, mock_db = self._create_handler_with_empty_keys() + mock_db.keys.return_value = None + result = handler.get_next(None) + self.assertIsNone(result) + + def test_get_next_with_sub_id_returns_none_when_no_mount_points(self): + handler, mock_db = self._create_handler_with_empty_keys() + mock_db.keys.return_value = None + result = handler.get_next((1,)) + self.assertIsNone(result) + + def test_get_hrstorage_used_returns_none_for_none_sub_id(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_used(None)) + + def test_get_hrstorage_size_returns_none_for_none_sub_id(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_size(None)) + + def test_get_hrstorage_descr_returns_none_for_none_sub_id(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_descr(None)) + + def test_get_hrstorage_alloc_returns_none_for_none_sub_id(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_alloc(None)) + + +class TestHrStorageHandlerNoMemoryStats(TestCase): + """Test hrStorageHandler when MOUNT_POINTS exist but no MEMORY_STATS.""" + + def _keys_side_effect(self, db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + if 'MEMORY_STATS' in pattern: + return [] + return None + + def _create_handler_no_memory_stats(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = self._keys_side_effect + mock_db.get_all.return_value = { + '1K-blocks': '52829740', 'Used': '0', + 'Filesystem': '/host' + } + handler = rfc2790.hrStorageHandler() + return handler, mock_db + + def test_init_hr_storage_no_memory_stats_logs_debug(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = self._keys_side_effect + mock_db.get_all.return_value = { + '1K-blocks': '52829740', 'Used': '0', + 'Filesystem': '/host' + } + + with patch.object(mibs.logger, 'info') as mock_info: + handler = rfc2790.hrStorageHandler() + mock_info.assert_any_call( + 'init_hr_storage - No memory stats found in STATE_DB MEMORY_STATS table' + ) + + def test_hr_storage_entries_has_mount_points_when_no_memory_stats(self): + handler, _ = self._create_handler_no_memory_stats() + self.assertEqual(handler.hr_storage_entries, ['MOUNT_POINTS|/dev']) + + def test_get_next_none_returns_first_when_no_memory_stats(self): + handler, mock_db = self._create_handler_no_memory_stats() + mock_db.keys.side_effect = self._keys_side_effect + mock_db.get_all.return_value = { + '1K-blocks': '52829740', 'Used': '0', + 'Filesystem': '/host' + } + result = handler.get_next(None) + self.assertEqual(result, (1,)) + + def test_get_next_with_sub_id_returns_none_past_end_when_no_memory_stats(self): + handler, _ = self._create_handler_no_memory_stats() + result = handler.get_next((len(handler.hr_storage_entries),)) + self.assertIsNone(result) + + +class TestHrStorageHandlerNeitherMountPointsNorMemoryStats(TestCase): + """Test hrStorageHandler when neither MOUNT_POINTS nor MEMORY_STATS exist.""" + + def test_init_hr_storage_logs_mount_points_missing_and_continues(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = [] + + with patch.object(mibs.logger, 'info') as mock_info: + handler = rfc2790.hrStorageHandler() + mock_info.assert_any_call( + 'init_hr_storage - No mount point found in STATE_DB MOUNT_POINTS table' + ) + mock_info.assert_any_call( + 'init_hr_storage - No memory stats found in STATE_DB MEMORY_STATS table' + ) + + self.assertEqual(handler.hr_storage_entries, []) + + +# =================================================================== +# Tests for stale data clearing after Redis is emptied +# =================================================================== + +def _populated_keys_side_effect(db, pattern): + """Return realistic MOUNT_POINTS and MEMORY_STATS keys.""" + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev', 'MOUNT_POINTS|/dev2'] + if 'MEMORY_STATS' in pattern: + return ['MEMORY_STATS|Physical'] + return None + + +def _populated_get_all_side_effect(db, key): + """Return realistic field data for each key.""" + data = { + 'MOUNT_POINTS|/dev': { + 'Type': 'ext4', 'MountPoint': '/dev', + '1K-blocks': '52829740', 'Used': '0', 'Filesystem': '/host', + }, + 'MOUNT_POINTS|/dev2': { + 'Type': 'ext3', 'MountPoint': '/dev2', + '1K-blocks': '12345', 'Used': '10', 'Filesystem': '/host2', + }, + 'MEMORY_STATS|Physical': { + '1K-blocks': '100000', 'Used': '1000', + }, + } + return data.get(key, {}) + + +class TestFsHandlerStaleDataClearing(TestCase): + """Verify that fsHandler clears stale cached entries when Redis is emptied.""" + + def _create_populated_handler(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.fsHandler() + return handler, mock_db + + def test_stale_fs_entries_cleared_when_mount_points_emptied(self): + handler, mock_db = self._create_populated_handler() + self.assertGreater(len(handler.fs_entries), 0, + "Precondition: handler should start with entries") + + mock_db.keys.side_effect = None + mock_db.keys.return_value = None + result = handler.get_next(None) + + self.assertIsNone(result) + self.assertEqual(handler.fs_entries, []) + + def test_stale_fs_entries_updated_when_memory_stats_emptied(self): + handler, mock_db = self._create_populated_handler() + original_count = len(handler.fs_entries) + self.assertGreater(original_count, 0) + + def keys_no_memory(db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + if 'MEMORY_STATS' in pattern: + return None + return None + + mock_db.keys.side_effect = keys_no_memory + mock_db.get_all.side_effect = _populated_get_all_side_effect + result = handler.get_next(None) + + self.assertEqual(result, (1,)) + self.assertEqual(handler.fs_entries, ['MOUNT_POINTS|/dev']) + + def test_reinit_data_clears_stale_fs_entries(self): + handler, mock_db = self._create_populated_handler() + self.assertGreater(len(handler.fs_entries), 0) + + mock_db.keys.side_effect = None + mock_db.keys.return_value = None + handler.reinit_data() + + self.assertEqual(handler.fs_entries, []) + + def test_reinit_data_repopulates_fs_entries(self): + handler, mock_db = self._create_populated_handler() + mock_db.keys.side_effect = None + mock_db.keys.return_value = None + handler.reinit_data() + self.assertEqual(handler.fs_entries, []) + + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler.reinit_data() + self.assertGreater(len(handler.fs_entries), 0) + + +class TestHrStorageHandlerStaleDataClearing(TestCase): + """Verify that hrStorageHandler clears stale cached entries when Redis is emptied.""" + + def _create_populated_handler(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.hrStorageHandler() + return handler, mock_db + + def test_stale_hr_storage_entries_cleared_when_mount_points_emptied(self): + handler, mock_db = self._create_populated_handler() + self.assertGreater(len(handler.hr_storage_entries), 0, + "Precondition: handler should start with entries") + + mock_db.keys.side_effect = None + mock_db.keys.return_value = None + result = handler.get_next(None) + + self.assertIsNone(result) + self.assertEqual(handler.hr_storage_entries, []) + + def test_stale_hr_storage_entries_updated_when_memory_stats_emptied(self): + handler, mock_db = self._create_populated_handler() + self.assertGreater(len(handler.hr_storage_entries), 0) + + def keys_no_memory(db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + if 'MEMORY_STATS' in pattern: + return None + return None + + mock_db.keys.side_effect = keys_no_memory + mock_db.get_all.side_effect = _populated_get_all_side_effect + result = handler.get_next(None) + + self.assertEqual(result, (1,)) + self.assertEqual(handler.hr_storage_entries, ['MOUNT_POINTS|/dev']) + + def test_reinit_data_clears_stale_hr_storage_entries(self): + handler, mock_db = self._create_populated_handler() + self.assertGreater(len(handler.hr_storage_entries), 0) + + mock_db.keys.side_effect = None + mock_db.keys.return_value = None + handler.reinit_data() + + self.assertEqual(handler.hr_storage_entries, []) + + def test_reinit_data_repopulates_hr_storage_entries(self): + handler, mock_db = self._create_populated_handler() + mock_db.keys.side_effect = None + mock_db.keys.return_value = None + handler.reinit_data() + self.assertEqual(handler.hr_storage_entries, []) + + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler.reinit_data() + self.assertGreater(len(handler.hr_storage_entries), 0) + + +# =================================================================== +# Tests for out-of-bounds and empty-list getter safety +# =================================================================== + +class TestFsHandlerGetterBoundsChecking(TestCase): + """Verify fsHandler getters return None for out-of-bounds or empty entries.""" + + def _create_handler_with_empty_keys(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = None + handler = rfc2790.fsHandler() + return handler, mock_db + + def _create_populated_handler(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.fsHandler() + return handler, mock_db + + def test_get_fs_type_returns_none_on_empty_entries(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_fs_type((1,))) + + def test_get_fs_mount_returns_none_on_empty_entries(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_fs_mount((1,))) + + def test_get_fs_type_returns_none_for_out_of_bounds(self): + handler, _ = self._create_populated_handler() + oob_index = len(handler.fs_entries) + 1 + self.assertIsNone(handler.get_fs_type((oob_index,))) + + def test_get_fs_mount_returns_none_for_out_of_bounds(self): + handler, _ = self._create_populated_handler() + oob_index = len(handler.fs_entries) + 1 + self.assertIsNone(handler.get_fs_mount((oob_index,))) + + def test_get_next_past_last_entry_returns_none(self): + handler, _ = self._create_populated_handler() + last_index = len(handler.fs_entries) + self.assertIsNone(handler.get_next((last_index,))) + + +class TestHrStorageHandlerGetterBoundsChecking(TestCase): + """Verify hrStorageHandler getters return None for out-of-bounds or empty entries.""" + + def _create_handler_with_empty_keys(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.return_value = None + handler = rfc2790.hrStorageHandler() + return handler, mock_db + + def _create_populated_handler(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.hrStorageHandler() + return handler, mock_db + + def test_get_hrstorage_used_returns_none_on_empty_entries(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_used((1,))) + + def test_get_hrstorage_size_returns_none_on_empty_entries(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_size((1,))) + + def test_get_hrstorage_descr_returns_none_on_empty_entries(self): + handler, _ = self._create_handler_with_empty_keys() + self.assertIsNone(handler.get_hrstorage_descr((1,))) + + def test_get_hrstorage_used_returns_none_for_out_of_bounds(self): + handler, _ = self._create_populated_handler() + oob_index = len(handler.hr_storage_entries) + 1 + self.assertIsNone(handler.get_hrstorage_used((oob_index,))) + + def test_get_hrstorage_size_returns_none_for_out_of_bounds(self): + handler, _ = self._create_populated_handler() + oob_index = len(handler.hr_storage_entries) + 1 + self.assertIsNone(handler.get_hrstorage_size((oob_index,))) + + def test_get_hrstorage_descr_returns_none_for_out_of_bounds(self): + handler, _ = self._create_populated_handler() + oob_index = len(handler.hr_storage_entries) + 1 + self.assertIsNone(handler.get_hrstorage_descr((oob_index,))) + + def test_get_next_past_last_entry_returns_none(self): + handler, _ = self._create_populated_handler() + last_index = len(handler.hr_storage_entries) + self.assertIsNone(handler.get_next((last_index,))) + + +# =================================================================== +# Tests for _db_call helper and STATE_DB access failure paths +# =================================================================== + +class TestDbCallHelper(TestCase): + """Verify _db_call logs errors and returns None on exception, + and returns results on success.""" + + def test_db_call_returns_result_on_success(self): + mock_func = MagicMock(return_value=['key1', 'key2']) + result = rfc2790._db_call(mock_func, 'test_context', 'arg1', 'arg2') + self.assertEqual(result, ['key1', 'key2']) + mock_func.assert_called_once_with('arg1', 'arg2') + + def test_db_call_returns_none_on_success(self): + mock_func = MagicMock(return_value=None) + result = rfc2790._db_call(mock_func, 'test_context', 'arg1') + self.assertIsNone(result) + + def test_db_call_logs_and_returns_none_on_exception(self): + mock_func = MagicMock(side_effect=ConnectionError("Redis unavailable")) + with patch.object(mibs.logger, 'error') as mock_error: + result = rfc2790._db_call(mock_func, 'MY_TABLE', 'arg1') + self.assertIsNone(result) + mock_error.assert_called_once() + self.assertIn('MY_TABLE', mock_error.call_args[0][0]) + self.assertIn('Redis unavailable', mock_error.call_args[0][0]) + + +class TestFsHandlerDbAccessFailure(TestCase): + """Verify fsHandler degrades gracefully when STATE_DB operations throw.""" + + def test_connect_failure_degrades_gracefully(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.connect.side_effect = ConnectionError("Redis down") + mock_db.keys.return_value = None + handler = rfc2790.fsHandler() + self.assertEqual(handler.fs_entries, []) + + def test_connect_failure_logs_error(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.connect.side_effect = ConnectionError("Redis down") + mock_db.keys.return_value = None + with patch.object(mibs.logger, 'error') as mock_error: + handler = rfc2790.fsHandler() + mock_error.assert_called() + error_messages = [str(c) for c in mock_error.call_args_list] + self.assertTrue( + any('Redis down' in msg for msg in error_messages)) + + def test_init_fs_keys_failure_degrades_gracefully(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = ConnectionError("Redis gone") + handler = rfc2790.fsHandler() + self.assertEqual(handler.fs_entries, []) + + def test_init_fs_get_all_failure_degrades_gracefully(self): + def keys_side_effect(db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + return None + + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = keys_side_effect + mock_db.get_all.side_effect = ConnectionError("Redis timeout") + handler = rfc2790.fsHandler() + self.assertEqual(handler.fs_entries, []) + + def test_init_fs_memory_stats_keys_failure_keeps_mount_points(self): + def keys_side_effect(db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + if 'MEMORY_STATS' in pattern: + raise ConnectionError("Redis lost") + return None + + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = keys_side_effect + mock_db.get_all.return_value = { + 'Type': 'ext4', 'MountPoint': '/dev' + } + handler = rfc2790.fsHandler() + self.assertEqual(handler.fs_entries, ['MOUNT_POINTS|/dev']) + + def test_get_fs_type_db_failure_returns_none(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.fsHandler() + + mock_db.get_all.side_effect = ConnectionError("Redis error") + result = handler.get_fs_type((1,)) + self.assertIsNone(result) + + def test_get_fs_mount_db_failure_returns_none(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.fsHandler() + + mock_db.get_all.side_effect = ConnectionError("Redis error") + result = handler.get_fs_mount((1,)) + self.assertIsNone(result) + + +class TestHrStorageHandlerDbAccessFailure(TestCase): + """Verify hrStorageHandler degrades gracefully when STATE_DB operations throw.""" + + def test_connect_failure_degrades_gracefully(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.connect.side_effect = ConnectionError("Redis down") + mock_db.keys.return_value = None + handler = rfc2790.hrStorageHandler() + self.assertEqual(handler.hr_storage_entries, []) + + def test_connect_failure_logs_error(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.connect.side_effect = ConnectionError("Redis down") + mock_db.keys.return_value = None + with patch.object(mibs.logger, 'error') as mock_error: + handler = rfc2790.hrStorageHandler() + mock_error.assert_called() + error_messages = [str(c) for c in mock_error.call_args_list] + self.assertTrue( + any('Redis down' in msg for msg in error_messages)) + + def test_init_hr_storage_keys_failure_degrades_gracefully(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = ConnectionError("Redis gone") + handler = rfc2790.hrStorageHandler() + self.assertEqual(handler.hr_storage_entries, []) + + def test_init_hr_storage_get_all_failure_degrades_gracefully(self): + def keys_side_effect(db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + return None + + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = keys_side_effect + mock_db.get_all.side_effect = ConnectionError("Redis timeout") + handler = rfc2790.hrStorageHandler() + self.assertEqual(handler.hr_storage_entries, []) + + def test_init_hr_storage_memory_stats_keys_failure_keeps_mount_points(self): + def keys_side_effect(db, pattern): + if 'MOUNT_POINTS' in pattern: + return ['MOUNT_POINTS|/dev'] + if 'MEMORY_STATS' in pattern: + raise ConnectionError("Redis lost") + return None + + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = keys_side_effect + mock_db.get_all.return_value = { + '1K-blocks': '52829740', 'Used': '0', 'Filesystem': '/host' + } + handler = rfc2790.hrStorageHandler() + self.assertEqual(handler.hr_storage_entries, ['MOUNT_POINTS|/dev']) + + def test_get_hrstorage_used_db_failure_returns_none(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.hrStorageHandler() + + mock_db.get_all.side_effect = ConnectionError("Redis error") + result = handler.get_hrstorage_used((1,)) + self.assertIsNone(result) + + def test_get_hrstorage_size_db_failure_returns_none(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.hrStorageHandler() + + mock_db.get_all.side_effect = ConnectionError("Redis error") + result = handler.get_hrstorage_size((1,)) + self.assertIsNone(result) + + def test_get_hrstorage_descr_db_failure_returns_none(self): + with patch.object(rfc2790.mibs, 'init_db') as mock_init_db: + mock_db = MagicMock() + mock_init_db.return_value = mock_db + mock_db.STATE_DB = 'STATE_DB' + mock_db.keys.side_effect = _populated_keys_side_effect + mock_db.get_all.side_effect = _populated_get_all_side_effect + handler = rfc2790.hrStorageHandler() + + mock_db.get_all.side_effect = ConnectionError("Redis error") + result = handler.get_hrstorage_descr((1,)) + self.assertIsNone(result) + diff --git a/tests/test_rfc2790_fs.py b/tests/test_rfc2790_fs.py new file mode 100644 index 000000000..c6550829d --- /dev/null +++ b/tests/test_rfc2790_fs.py @@ -0,0 +1,356 @@ +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") + + def test_getNextFSType3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 4)) + 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, "") + + def test_getFSType3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 4)) + 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, "") + + def test_getNextFSType4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 8)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 9)) + 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, "") + + def test_getFSType4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 4, 9)) + 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, "") + + # ======= 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") + + def test_getNextMountFS3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 4)) + 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, "") + + def test_getMountFS3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 4)) + 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, "") + + def test_getNextMountFS4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 8)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 9)) + 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, "") + + def test_getMountFS4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 3, 8, 1, 2, 9)) + 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, "") + diff --git a/tests/test_rfc2790_hrstorage.py b/tests/test_rfc2790_hrstorage.py new file mode 100644 index 000000000..27eb51428 --- /dev/null +++ b/tests/test_rfc2790_hrstorage.py @@ -0,0 +1,1038 @@ +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) + + def test_getNextHrStorageUsed3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 4)) + 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_getHrStorageUsed3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 4)) + 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) + + def test_getNextHrStorageUsed4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 4)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 5)) + 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_getHrStorageUsed4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 5)) + 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_getNextHrStorageUsed5(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 5)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 6)) + 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_getHrStorageUsed5(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 6)) + 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_getNextHrStorageUsed6(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 6)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 7)) + 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_getHrStorageUsed6(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 7)) + 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_getNextHrStorageUsed7(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 7)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 8)) + 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, 200) + + def test_getHrStorageUsed7(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 8)) + 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, 200) + + def test_getNextHrStorageUsed8(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 8)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 9)) + 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, 300) + + def test_getHrStorageUsed8(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 6, 9)) + 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, 300) + + # ====== 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) + + def test_getNextHrStorageSize3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 4)) + 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, 100000) + + def test_getHrStorageSize3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 4)) + 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, 100000) + + def test_getNextHrStorageSize4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 4)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 5)) + 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, 100000) + + def test_getHrStorageSize4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 5)) + 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, 100000) + + def test_getNextHrStorageSize5(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 5)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 6)) + 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_getHrStorageSize5(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 6)) + 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_getNextHrStorageSize6(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 6)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 7)) + 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_getHrStorageSize6(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 7)) + 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_getNextHrStorageSize7(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 7)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 8)) + 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, 200) + + def test_getHrStorageSize7(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 8)) + 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, 200) + + def test_getNextHrStorageSize8(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 8)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 9)) + 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, 678) + + def test_getHrStorageSize8(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 5, 9)) + 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, 678) + + # ====== 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") + + def test_getNextHrStorageDescr3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 4)) + 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, "Physical Memory") + + def test_getHrStorageDescr3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 4)) + 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, "Physical Memory") + + def test_getNextHrStorageDescr4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 4)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 5)) + 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, "Virtual Memory") + + def test_getHrStorageDescr4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 5)) + 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, "Virtual Memory") + + def test_getNextHrStorageDescr5(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 5)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 6)) + 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, "Swap Memory") + + def test_getHrStorageDescr5(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 6)) + 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, "Swap Memory") + + def test_getNextHrStorageDescr6(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 6)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 7)) + 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, "Cached Memory") + + def test_getHrStorageDescr6(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 7)) + 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, "Cached Memory") + + def test_getNextHrStorageDescr7(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 7)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 8)) + 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, "Shared Memory") + + def test_getHrStorageDescr7(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 8)) + 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, "Shared Memory") + + def test_getNextHrStorageDescr8(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 8)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 9)) + 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, "Buffer Memory") + + def test_getHrStorageDescr8(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 3, 9)) + 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, "Buffer Memory") + + # ====== 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_getHrStorageAlloc2(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) + + def test_getNextHrStorageAlloc3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 3)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 4)) + 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_getHrStorageAlloc3(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 4)) + 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_getNextHrStorageAlloc4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 8)) + expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 9)) + 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_getHrStorageAlloc4(self): + oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 2, 1, 25, 2, 3, 1, 4, 9)) + 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) +