From c9962e2821cb2ce4b0c8fe8c27e0b8161eddc9ba Mon Sep 17 00:00:00 2001 From: purhan Date: Sun, 13 Jun 2021 13:56:18 +0530 Subject: [PATCH] [chores/qa] Improvements to tests - Fix failing tests - Pass more realistic values in mocks - Remove redundant code --- netengine/backends/snmp/openwrt.py | 13 +++-- tests/static/test-openwrt-snmp-oid.json | 3 +- tests/test_snmp/test_airos.py | 66 +++++++------------------ tests/test_snmp/test_openwrt.py | 54 ++++++++------------ tests/utils.py | 44 ++++++++++++----- 5 files changed, 81 insertions(+), 99 deletions(-) diff --git a/netengine/backends/snmp/openwrt.py b/netengine/backends/snmp/openwrt.py index 8de89dd..7254125 100644 --- a/netengine/backends/snmp/openwrt.py +++ b/netengine/backends/snmp/openwrt.py @@ -507,10 +507,15 @@ def neighbors(self): result = [] for index, neighbor in enumerate(neighbors): - mac = EUI(int(neighbor[0][1].prettyPrint(), 16), dialect=mac_unix_expanded) - interface_num = neighbor[0][0].getOid()[10] - interface = self.get(f'1.3.6.1.2.1.31.1.1.1.1.{interface_num}')[3][0][1] - state = states_map[str(neighbor_states[index][0][1])] + try: + mac = EUI( + int(neighbor[0][1].prettyPrint(), 16), dialect=mac_unix_expanded + ) + interface_num = neighbor[0][0].getOid()[10] + interface = self.get(f'1.3.6.1.2.1.31.1.1.1.1.{interface_num}')[3][0][1] + state = states_map[str(neighbor_states[index][0][1])] + except (IndexError, TypeError): + continue result.append( self._dict( {'mac': str(mac), 'state': str(state), 'interface': str(interface)} diff --git a/tests/static/test-openwrt-snmp-oid.json b/tests/static/test-openwrt-snmp-oid.json index 480de5c..f82cd10 100644 --- a/tests/static/test-openwrt-snmp-oid.json +++ b/tests/static/test-openwrt-snmp-oid.json @@ -54,5 +54,6 @@ "1.3.6.1.2.1.25.1.2.0": { "type": "bytes", "value": "\\x07\\xe5\\x06\\x0b\\x06\\x00\r\\x00+\\x00\\x00" - } + }, + "1.3.6.1.2.1.31.1.1.1.1.5": "br-lan" } diff --git a/tests/test_snmp/test_airos.py b/tests/test_snmp/test_airos.py index ab19a75..a9b5315 100644 --- a/tests/test_snmp/test_airos.py +++ b/tests/test_snmp/test_airos.py @@ -29,7 +29,7 @@ def setUp(self): target=cmdgen.CommandGenerator, attribute='nextCmd', wrap_obj=self.device._command, - return_value=[0, 0, 0, [[[0, 1]]] * 5], + side_effect=self._get_mocked_nextcmd, ) self.getcmd_patcher = SpyMock._patch( target=cmdgen.CommandGenerator, @@ -40,6 +40,7 @@ def setUp(self): ), ) self.getcmd_patcher.start() + self.nextcmd_patcher.start() def test_get_value_error(self): self.getcmd_patcher.stop() @@ -81,53 +82,37 @@ def test_os(self): self.assertIsInstance(self.device.os, tuple) def test_get_interfaces(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.get_interfaces(), list) + self.assertIsInstance(self.device.get_interfaces(), list) def test_get_interfaces_mtu(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_mtu, list) + self.assertIsInstance(self.device.interfaces_mtu, list) def test_interfaces_state(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_state, list) + self.assertIsInstance(self.device.interfaces_state, list) def test_interfaces_speed(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_speed, list) + self.assertIsInstance(self.device.interfaces_speed, list) def test_interfaces_bytes(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_bytes, list) + self.assertIsInstance(self.device.interfaces_bytes, list) def test_interfaces_MAC(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_MAC, list) + self.assertIsInstance(self.device.interfaces_MAC, list) def test_interfaces_type(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_type, list) + self.assertIsInstance(self.device.interfaces_type, list) def test_interfaces_to_dict(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_to_dict, list) + self.assertIsInstance(self.device.interfaces_to_dict, list) def test_wireless_dbm(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.wireless_dbm, list) + self.assertIsInstance(self.device.wireless_dbm, list) def test_interfaces_number(self): self.assertIsInstance(self.device.interfaces_number, int) def test_wireless_to_dict(self): - with self.nextcmd_patcher as np: - SpyMock._update_patch( - np, - _mock_side_effect=lambda *args: self._get_mocked_wireless_links( - data=args - ), - ) - self.assertIsInstance(self.device.wireless_links, list) + self.assertIsInstance(self.device.wireless_links, list) def test_RAM_free(self): self.assertIsInstance(self.device.RAM_free, int) @@ -136,31 +121,16 @@ def test_RAM_total(self): self.assertIsInstance(self.device.RAM_total, int) def test_to_dict(self): - with self.nextcmd_patcher as np: - SpyMock._update_patch( - np, - _mock_side_effect=lambda *args: self._get_mocked_wireless_links( - data=args - ), - ) - self.assertTrue(isinstance(self.device.to_dict(), dict)) + self.assertTrue(isinstance(self.device.to_dict(), dict)) def test_netjson_compliance(self): - with self.nextcmd_patcher as np: - SpyMock._update_patch( - np, - _mock_side_effect=lambda *args: self._get_mocked_wireless_links( - data=args - ), - ) - device_dict = self.device.to_dict() - device_json = self.device.to_json() - validate(instance=device_dict, schema=schema) - validate(instance=json.loads(device_json), schema=schema) + device_dict = self.device.to_dict() + device_json = self.device.to_json() + validate(instance=device_dict, schema=schema) + validate(instance=json.loads(device_json), schema=schema) def test_manufacturer(self): - with self.nextcmd_patcher: - self.assertIsNotNone(self.device.manufacturer) + self.assertIsNotNone(self.device.manufacturer) def test_model(self): self.assertIsInstance(self.device.model, str) diff --git a/tests/test_snmp/test_openwrt.py b/tests/test_snmp/test_openwrt.py index 40c1725..906be6c 100644 --- a/tests/test_snmp/test_openwrt.py +++ b/tests/test_snmp/test_openwrt.py @@ -27,7 +27,7 @@ def setUp(self): target=cmdgen.CommandGenerator, attribute='nextCmd', wrap_obj=self.device._command, - return_value=[0, 0, 0, [[[0, 1]]] * 5], + side_effect=self._get_mocked_nextcmd, ) self.getcmd_patcher = SpyMock._patch( target=cmdgen.CommandGenerator, @@ -38,13 +38,13 @@ def setUp(self): ), ) self.getcmd_patcher.start() + self.nextcmd_patcher.start() def test_os(self): self.assertIsInstance(self.device.os, tuple) def test_manufacturer(self): - with self.nextcmd_patcher: - self.assertIsNotNone(self.device.manufacturer) + self.assertIsNotNone(self.device.manufacturer) def test_name(self): self.assertIsInstance(self.device.name, str) @@ -56,41 +56,31 @@ def test_uptime_tuple(self): self.assertIsInstance(self.device.uptime_tuple, tuple) def test_get_interfaces(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.get_interfaces(), list) + self.assertIsInstance(self.device.get_interfaces(), list) def test_interfaces_speed(self): self.assertIsInstance(self.device.interfaces_speed, list) def test_interfaces_bytes(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_bytes, list) + self.assertIsInstance(self.device.interfaces_bytes, list) def test_interfaces_MAC(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_MAC, list) + self.assertIsInstance(self.device.interfaces_MAC, list) def test_interfaces_type(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_type, list) + self.assertIsInstance(self.device.interfaces_type, list) def test_interfaces_mtu(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_mtu, list) + self.assertIsInstance(self.device.interfaces_mtu, list) def test_interfaces_state(self): - with self.nextcmd_patcher: - self.assertIsInstance(self.device.interfaces_up, list) + self.assertIsInstance(self.device.interfaces_up, list) def test_interfaces_to_dict(self): - with self.nextcmd_patcher as p: - p.return_value = (0, 0, 0, []) - self.assertIsInstance(self.device.interfaces_to_dict, list) + self.assertIsInstance(self.device.interfaces_to_dict, list) def test_interface_addr_and_mask(self): - with self.nextcmd_patcher as p: - p.return_value = (0, 0, 0, []) - self.assertIsInstance(self.device.interface_addr_and_mask, dict) + self.assertIsInstance(self.device.interface_addr_and_mask, dict) def test_RAM_total(self): self.assertIsInstance(self.device.RAM_total, int) @@ -120,21 +110,17 @@ def test_local_time(self): self.assertIsInstance(self.device.local_time, int) def test_to_dict(self): - with self.nextcmd_patcher as p: - SpyMock._update_patch(p, _mock_return_value=[0, 0, 0, []]) - device_dict = self.device.to_dict() - self.assertIsInstance(device_dict, dict) - self.assertEqual( - len(device_dict['interfaces']), len(self.device.get_interfaces()), - ) + device_dict = self.device.to_dict() + self.assertIsInstance(device_dict, dict) + self.assertEqual( + len(device_dict['interfaces']), len(self.device.get_interfaces()), + ) def test_netjson_compliance(self): - with self.nextcmd_patcher as p: - SpyMock._update_patch(p, _mock_return_value=[0, 0, 0, []]) - device_dict = self.device.to_dict() - device_json = self.device.to_json() - validate(instance=device_dict, schema=schema) - validate(instance=json.loads(device_json), schema=schema) + device_dict = self.device.to_dict() + device_json = self.device.to_json() + validate(instance=device_dict, schema=schema) + validate(instance=json.loads(device_json), schema=schema) def tearDown(self): patch.stopall() diff --git a/tests/utils.py b/tests/utils.py index e6fcac0..db1a7f2 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -3,9 +3,19 @@ import os from unittest import mock +from pysnmp.hlapi import OctetString + from .settings import settings +class MockOid: + def getOid(self): + return self.oid + + def __init__(self, oid): + self.oid = oid + + class SpyMock: @staticmethod def _patch(*args, **kwargs): @@ -43,17 +53,27 @@ def _get_mocked_getcmd(data, input): return [0, 0, 0, [[0, result]]] @staticmethod - def _get_mocked_wireless_links(data): - oid = data[2] - return_data = { - '1.3.6.1.4.1.14988.1.1.1.2.1': [0, 0, 0, [[[0, 0], 0]] * 28], - '1.3.6.1.4.1.14988.1.1.1.2.1.3': [0, 0, 0, [0, 0]], - '1.3.6.1.4.1.14988.1.1.1.2.1.3.0': [None, 0, 0, []], - '1.3.6.1.2.1.1.9.1.1': [ - 0, - 0, - 0, - [[[0, 1]], [[0, 2]], [[0, 3]], [[0, 4]], [[0, 5]]], + def _get_mocked_nextcmd(*args, **kwargs): + def _get_nextcmd_list(return_value): + # pass `None` as the data we don't use + return [None, None, None, return_value] + + res = { + '1.3.6.1.4.1.14988.1.1.1.2.1': [[[0, 0], 0]] * 28, + '1.3.6.1.4.1.14988.1.1.1.2.1.3': [0, 0], + '1.3.6.1.4.1.14988.1.1.1.2.1.3.0': [], + '1.3.6.1.2.1.1.9.1.1': [[[0, 1]], [[0, 2]], [[0, 3]], [[0, 4]], [[0, 5]]], + '1.3.6.1.2.1.4.35.1.4': [ + MockOid('1.3.6.1.2.5'), + OctetString('0x040e3cca555f'), ], + '1.3.6.1.2.1.4.35.1.7': [MockOid('1.3.6.1.2.5'), 1], + '1.3.6.1.2.1.2.2.1.6.': [[[0, 1]], [[0, 2]], [[0, 3]], [[0, 4]], [[0, 5]]], + '1.3.6.1.2.1.2.2.1.1': [[[0, 1]], [[0, 2]], [[0, 3]], [[0, 4]], [[0, 5]]], + '1.3.6.1.2.1.4.20.1.1': [[[0, OctetString('127.0.0.1')]]], + '1.3.6.1.2.1.4.20.1.2': [[[0, 1]]], + '1.3.6.1.2.1.25.3.3.1.2': [0, 2], + '1.3.6.1.2.1.4.20.1.3': [[[0, OctetString('192.168.0.1')]]], } - return return_data[oid] + oid = args[2] + return _get_nextcmd_list(res[oid])