From 844f0aa1a6cb419d5b023079ae42e84211f15653 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 5 Sep 2023 13:17:47 +0200 Subject: [PATCH 01/25] Add more exceptions --- python/nav/portadmin/handlers.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/python/nav/portadmin/handlers.py b/python/nav/portadmin/handlers.py index 250342272d..ec63baca11 100644 --- a/python/nav/portadmin/handlers.py +++ b/python/nav/portadmin/handlers.py @@ -338,3 +338,15 @@ class ProtocolError(ManagementError): """Raised when some non-categorized error in the underlying protocol occurred during communication """ + + +class POENotSupportedError(ManagementError): + """Raised when an interface that does not support PoE is used in a context where PoE support is expected""" + + +class POEStateNotSupportedError(ManagementError): + """Raised when a PoE state is detected in a context where it is not supported""" + + +class XMLParseError(ManagementError): + """Raised when failing to parse XML""" From 416fe8d4748b0734bb8f24d7f2461ecf8770d052 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 5 Sep 2023 13:59:31 +0200 Subject: [PATCH 02/25] Add juniper commandline scripts --- .../portadmin/napalm/templates/portadmin/junos-disable-poe.djt | 1 + .../portadmin/napalm/templates/portadmin/junos-enable-poe.djt | 1 + 2 files changed, 2 insertions(+) create mode 100644 python/nav/portadmin/napalm/templates/portadmin/junos-disable-poe.djt create mode 100644 python/nav/portadmin/napalm/templates/portadmin/junos-enable-poe.djt diff --git a/python/nav/portadmin/napalm/templates/portadmin/junos-disable-poe.djt b/python/nav/portadmin/napalm/templates/portadmin/junos-disable-poe.djt new file mode 100644 index 0000000000..da2379be33 --- /dev/null +++ b/python/nav/portadmin/napalm/templates/portadmin/junos-disable-poe.djt @@ -0,0 +1 @@ +set poe interface {{ ifname }} disable diff --git a/python/nav/portadmin/napalm/templates/portadmin/junos-enable-poe.djt b/python/nav/portadmin/napalm/templates/portadmin/junos-enable-poe.djt new file mode 100644 index 0000000000..20a8a9e08c --- /dev/null +++ b/python/nav/portadmin/napalm/templates/portadmin/junos-enable-poe.djt @@ -0,0 +1 @@ +delete poe interface {{ ifname }} disable From 7ff3818c184eab553d7adb7936a9dcf3b40f6573 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:43:14 +0100 Subject: [PATCH 03/25] Add poestates for juniper --- python/nav/portadmin/napalm/juniper.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 455e510cba..beefda5733 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -44,6 +44,7 @@ AuthenticationError, NoResponseError, ProtocolError, + PoeState, ) from nav.junos.nav_views import ( EthernetSwitchingInterfaceTable, @@ -102,6 +103,8 @@ class Juniper(ManagementHandler): VENDOR = VENDOR_ID_JUNIPER_NETWORKS_INC PROTOCOL = manage.ManagementProfile.PROTOCOL_NAPALM + POE_ENABLED = PoeState(state=1, name="ENABLED") + POE_DISABLED = PoeState(state=2, name="DISABLED") def __init__(self, netbox: manage.Netbox, **kwargs): super().__init__(netbox, **kwargs) From c8a8c8893d9006df864a926601dbf4e6411739e4 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:44:12 +0100 Subject: [PATCH 04/25] Add get poe state options for juniper --- python/nav/portadmin/napalm/juniper.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index beefda5733..958fd805b2 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -444,6 +444,10 @@ def raise_if_not_configurable(self): if not self.profile: raise DeviceNotConfigurableError("Device has no NAPALM profile") + def get_poe_state_options(self) -> Sequence[PoeState]: + """Returns the available options for enabling/disabling PoE on this netbox""" + return [self.POE_ENABLED, self.POE_DISABLED] + # FIXME Implement dot1x fetcher methods # dot1x authentication configuration fetchers aren't implemented yet, for lack # of configured devices to test on From 7a8729148009c88b8d7f61180288cb571ea56bc0 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:46:15 +0100 Subject: [PATCH 05/25] Add set_poe_state for juniper --- python/nav/portadmin/napalm/juniper.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 958fd805b2..27f15cbd5e 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -45,6 +45,7 @@ NoResponseError, ProtocolError, PoeState, + POEStateNotSupportedError, ) from nav.junos.nav_views import ( EthernetSwitchingInterfaceTable, @@ -448,6 +449,23 @@ def get_poe_state_options(self) -> Sequence[PoeState]: """Returns the available options for enabling/disabling PoE on this netbox""" return [self.POE_ENABLED, self.POE_DISABLED] + @wrap_unhandled_rpc_errors + def set_poe_state(self, interface: manage.Interface, state: PoeState): + """Set state for enabling/disabling PoE on this interface. + Available options should be retrieved using `get_poe_state_options` + """ + if not isinstance(state, PoeState): + raise TypeError("state must be a PoeState object") + if state == self.POE_ENABLED: + template = get_template("portadmin/junos-enable-poe.djt") + elif state == self.POE_DISABLED: + template = get_template("portadmin/junos-disable-poe.djt") + else: + raise POEStateNotSupportedError(f"state {state} is not a valid state") + master, _ = split_master_unit(interface.ifname) + config = template.render({"ifname": master}) + self.device.load_merge_candidate(config=config) + # FIXME Implement dot1x fetcher methods # dot1x authentication configuration fetchers aren't implemented yet, for lack # of configured devices to test on From 1d8fa2bbd076855d0c4f6a5c8f590ff5415ef984 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:47:03 +0100 Subject: [PATCH 06/25] Add funcs for getting poe state via napalm --- python/nav/portadmin/napalm/juniper.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 27f15cbd5e..9664a23cca 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -34,6 +34,7 @@ from napalm.base.exceptions import ConnectAuthError, ConnectionException from jnpr.junos.op.vlan import VlanTable from jnpr.junos.exception import RpcError +from lxml.etree import ElementTree from nav.napalm import connect as napalm_connect from nav.enterprise.ids import VENDOR_ID_JUNIPER_NETWORKS_INC @@ -466,6 +467,14 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): config = template.render({"ifname": master}) self.device.load_merge_candidate(config=config) + @wrap_unhandled_rpc_errors + def _get_all_poe_interface_information(self) -> ElementTree: + return self.device.device.rpc.get_poe_interface_information() + + @wrap_unhandled_rpc_errors + def _get_poe_interface_information(self, ifname: str) -> ElementTree: + return self.device.device.rpc.get_poe_interface_information(ifname=ifname) + # FIXME Implement dot1x fetcher methods # dot1x authentication configuration fetchers aren't implemented yet, for lack # of configured devices to test on From 18bccea538e332341e23aa2bc8b3961ec7d548b5 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:48:02 +0100 Subject: [PATCH 07/25] Add func for parsing poe state --- python/nav/portadmin/napalm/juniper.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 9664a23cca..740842ec56 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -475,6 +475,18 @@ def _get_all_poe_interface_information(self) -> ElementTree: def _get_poe_interface_information(self, ifname: str) -> ElementTree: return self.device.device.rpc.get_poe_interface_information(ifname=ifname) + def _poe_string_to_state(self, state_str: str) -> PoeState: + """Converts from internal juniper state names to + corresponding PoeState objects + """ + state_cleaned = state_str.strip().lower() + if state_cleaned == "enabled": + return self.POE_ENABLED + elif state_cleaned == "disabled": + return self.POE_DISABLED + else: + raise POEStateNotSupportedError(f"Unknown PoE state {state_str}") + # FIXME Implement dot1x fetcher methods # dot1x authentication configuration fetchers aren't implemented yet, for lack # of configured devices to test on From d44ec52a3f4753a04a233ae0c46228d1396de854 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:49:44 +0100 Subject: [PATCH 08/25] Add func for setting state for one interface --- python/nav/portadmin/napalm/juniper.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 740842ec56..7cee214738 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -47,6 +47,8 @@ ProtocolError, PoeState, POEStateNotSupportedError, + POENotSupportedError, + XMLParseError, ) from nav.junos.nav_views import ( EthernetSwitchingInterfaceTable, @@ -467,6 +469,25 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): config = template.render({"ifname": master}) self.device.load_merge_candidate(config=config) + def _get_poe_state_for_single_interface( + self, interface: manage.Interface + ) -> PoeState: + tree = self._get_poe_interface_information(ifname=interface.ifname) + matching_elements = tree.xpath( + "//poe/interface-information-detail/interface-enabled-detail" + ) + # Interfaces that do not support PoE will not have this element + if not matching_elements: + raise POENotSupportedError( + f"Interface {interface.ifname} does not support PoE" + ) + if len(matching_elements) != 1: + raise XMLParseError( + f"Expected 1 matching element in xml response, {len(matching_elements)} found" + ) + ifenabled = matching_elements[0].text.lower() + return self._poe_string_to_state(ifenabled) + @wrap_unhandled_rpc_errors def _get_all_poe_interface_information(self) -> ElementTree: return self.device.device.rpc.get_poe_interface_information() From c50d46322519cc2e14e3c45cba5e1f42ae15a97f Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:50:35 +0100 Subject: [PATCH 09/25] Add func for getting state for multiple interfaces --- python/nav/portadmin/napalm/juniper.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 7cee214738..ea32842161 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -28,7 +28,7 @@ """ from __future__ import annotations from operator import attrgetter -from typing import List, Any, Dict, Tuple, Sequence +from typing import List, Any, Dict, Tuple, Sequence, Optional from django.template.loader import get_template from napalm.base.exceptions import ConnectAuthError, ConnectionException @@ -488,6 +488,22 @@ def _get_poe_state_for_single_interface( ifenabled = matching_elements[0].text.lower() return self._poe_string_to_state(ifenabled) + def _get_poe_state_for_multiple_interfaces( + self, interfaces: Sequence[manage.Interface] + ) -> Dict[int, Optional[PoeState]]: + tree = self._get_all_poe_interface_information() + interface_information_elements = tree.findall(".//interface-information") + ifname_to_state_dict = {} + for element in interface_information_elements: + ifname = element.findall(".//interface-name")[0].text.strip().lower() + ifenabled = element.findall(".//interface-enabled")[0].text.strip().lower() + ifname_to_state_dict[ifname] = self._poe_string_to_state(ifenabled) + ifindex_to_state_dict = { + interface.ifindex: ifname_to_state_dict.get(interface.ifname.lower()) + for interface in interfaces + } + return ifindex_to_state_dict + @wrap_unhandled_rpc_errors def _get_all_poe_interface_information(self) -> ElementTree: return self.device.device.rpc.get_poe_interface_information() From 8a0459b484352f7e767c046a27cffc19cbdd84e4 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 14:55:08 +0100 Subject: [PATCH 10/25] Add get_poe_states for juniper --- python/nav/portadmin/napalm/juniper.py | 30 ++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index ea32842161..209483318d 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -469,6 +469,36 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): config = template.render({"ifname": master}) self.device.load_merge_candidate(config=config) + def get_poe_states( + self, interfaces: Sequence[manage.Interface] = None + ) -> Dict[int, Optional[PoeState]]: + """Retrieves current PoE state for interfaces on this device. + + :param interfaces: Optional sequence of interfaces to filter for, as fetching + data for all interfaces may be a waste of time if only a + single interface is needed. If this parameter is omitted, + the default behavior is to filter on all Interface objects + registered for this device. + :returns: A dict mapping interfaces to their discovered PoE state. + The key matches the `ifindex` attribute for the related + Interface object. + The value will be None if the interface does not support PoE. + """ + if not interfaces: + if self.netbox.interfaces: + interfaces = self.netbox.interfaces + else: + return {} + if len(interfaces) == 1: + interface = interfaces[0] + try: + state = self._get_poe_state_for_single_interface(interface) + except POENotSupportedError: + state = None + return {interface.ifindex: state} + else: + return self._get_poe_state_for_multiple_interfaces(interfaces) + def _get_poe_state_for_single_interface( self, interface: manage.Interface ) -> PoeState: From 2bc25a234ec31b3e81012a8f4787698074fa11b1 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 15:09:03 +0100 Subject: [PATCH 11/25] Add new fixtures --- .../portadmin/napalm/juniper_test.py | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/tests/unittests/portadmin/napalm/juniper_test.py b/tests/unittests/portadmin/napalm/juniper_test.py index 4f3fa2a9c5..83d860416d 100644 --- a/tests/unittests/portadmin/napalm/juniper_test.py +++ b/tests/unittests/portadmin/napalm/juniper_test.py @@ -19,6 +19,7 @@ from unittest.mock import Mock, patch from jnpr.junos.exception import RpcError +from lxml import etree from nav.enterprise.ids import VENDOR_ID_RESERVED, VENDOR_ID_JUNIPER_NETWORKS_INC from nav.models import manage @@ -45,6 +46,62 @@ def profile_mock(): yield profile +@pytest.fixture() +def handler_mock(netbox_mock, profile_mock): + """Create management handler mock object""" + juniper = Juniper(netbox=netbox_mock) + juniper._profile = profile_mock + yield juniper + + +@pytest.fixture() +def xml(interface1_mock): + """Creates a ElementTree containing poe information for one interface""" + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def xml_bulk(interface1_mock, interface2_mock): + """Creates a ElementTree containing poe information for two interfaces""" + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + + {interface2_mock.ifname} + Disabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def interface1_mock(): + interface = Mock() + interface.ifname = "ge-0/0/1" + interface.ifindex = 1 + yield interface + + +@pytest.fixture() +def interface2_mock(): + interface = Mock() + interface.ifname = "ge-0/0/2" + interface.ifindex = 2 + yield interface + + class TestWrapUnhandledRpcErrors: def test_rpcerrors_should_become_protocolerrors(self): @wrap_unhandled_rpc_errors From 49ba479210316b429bc4a746a89da2eddfe849e0 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 15:09:11 +0100 Subject: [PATCH 12/25] Replace old fixture with new --- tests/unittests/portadmin/napalm/juniper_test.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/tests/unittests/portadmin/napalm/juniper_test.py b/tests/unittests/portadmin/napalm/juniper_test.py index 83d860416d..5b93bdc2d3 100644 --- a/tests/unittests/portadmin/napalm/juniper_test.py +++ b/tests/unittests/portadmin/napalm/juniper_test.py @@ -121,7 +121,7 @@ def wrapped_function(): class TestJuniper: - def test_juniper_device_returns_device_connection(self, netbox_mock, profile_mock): + def test_juniper_device_returns_device_connection(self, handler_mock): driver = napalm.get_network_driver('mock') device = driver( hostname='foo', @@ -130,10 +130,7 @@ def test_juniper_device_returns_device_connection(self, netbox_mock, profile_moc optional_args={}, ) device.open() - juniper = Juniper(netbox=netbox_mock) - juniper._profile = profile_mock - - assert juniper.device + assert handler_mock.device def test_juniper_device_raises_error_if_vendor_not_juniper( self, netbox_mock, profile_mock From 009242f283b584d584b1ba16ae1473f0d0856035 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Tue, 31 Oct 2023 15:14:14 +0100 Subject: [PATCH 13/25] Add juniper PoE tests --- .../portadmin/napalm/juniper_test.py | 48 ++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/tests/unittests/portadmin/napalm/juniper_test.py b/tests/unittests/portadmin/napalm/juniper_test.py index 5b93bdc2d3..bba95697ca 100644 --- a/tests/unittests/portadmin/napalm/juniper_test.py +++ b/tests/unittests/portadmin/napalm/juniper_test.py @@ -23,7 +23,11 @@ from nav.enterprise.ids import VENDOR_ID_RESERVED, VENDOR_ID_JUNIPER_NETWORKS_INC from nav.models import manage -from nav.portadmin.handlers import DeviceNotConfigurableError, ProtocolError +from nav.portadmin.handlers import ( + DeviceNotConfigurableError, + ProtocolError, + POEStateNotSupportedError, +) from nav.portadmin.napalm.juniper import wrap_unhandled_rpc_errors, Juniper @@ -161,3 +165,45 @@ def vlans(self): m = MockedJuniperHandler(Mock()) assert len(m.get_netbox_vlans()) == 1 + + +class TestJuniperPoe: + def test_returns_correct_state_options(self, handler_mock): + state_options = handler_mock.get_poe_state_options() + assert Juniper.POE_ENABLED in state_options + assert Juniper.POE_DISABLED in state_options + + def test_state_converter_returns_correct_states(self, handler_mock): + assert handler_mock._poe_string_to_state("enabled") == Juniper.POE_ENABLED + assert handler_mock._poe_string_to_state("disabled") == Juniper.POE_DISABLED + + def test_state_converter_raises_error_for_invalid_states(self, handler_mock): + with pytest.raises(POEStateNotSupportedError): + handler_mock._poe_string_to_state("invalid_state") + + def test_get_poe_state_for_single_interface_returns_correct_state( + self, handler_mock, xml, interface1_mock + ): + handler_mock._get_poe_interface_information = Mock(return_value=xml) + state = handler_mock._get_poe_state_for_single_interface(interface1_mock) + assert state == Juniper.POE_ENABLED + + def test_get_poe_state_for_multiple_interfaces_returns_correct_states( + self, handler_mock, xml_bulk, interface1_mock, interface2_mock + ): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + states = handler_mock._get_poe_state_for_multiple_interfaces( + [interface1_mock, interface2_mock] + ) + assert states[interface1_mock.ifindex] == Juniper.POE_ENABLED + assert states[interface2_mock.ifindex] == Juniper.POE_DISABLED + + def test_get_poe_states_for_multiple_interfaces_maps_interface_to_none_if_poe_not_supported( + self, handler_mock, xml_bulk + ): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + if_mock = Mock() + if_mock.ifname == "random_if" + if_mock.ifindex = 0 + states = handler_mock._get_poe_state_for_multiple_interfaces([if_mock]) + assert states[if_mock.ifindex] is None From b8c4341463af06920fa47ff832d964221127aef3 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Thu, 2 Nov 2023 08:59:26 +0100 Subject: [PATCH 14/25] Move fixtures to bottom of file --- .../portadmin/napalm/juniper_test.py | 165 ++++++++++-------- 1 file changed, 90 insertions(+), 75 deletions(-) diff --git a/tests/unittests/portadmin/napalm/juniper_test.py b/tests/unittests/portadmin/napalm/juniper_test.py index bba95697ca..4c4d510449 100644 --- a/tests/unittests/portadmin/napalm/juniper_test.py +++ b/tests/unittests/portadmin/napalm/juniper_test.py @@ -31,81 +31,6 @@ from nav.portadmin.napalm.juniper import wrap_unhandled_rpc_errors, Juniper -@pytest.fixture() -def netbox_mock(): - """Create netbox model mock object""" - netbox = Mock() - netbox.ip = '10.0.0.1' - netbox.type.get_enterprise_id.return_value = VENDOR_ID_JUNIPER_NETWORKS_INC - yield netbox - - -@pytest.fixture() -def profile_mock(): - """Create management profile model mock object""" - profile = Mock() - profile.protocol = manage.ManagementProfile.PROTOCOL_NAPALM - profile.PROTOCOL_NAPALM = manage.ManagementProfile.PROTOCOL_NAPALM - profile.configuration = {"driver": "mock"} - yield profile - - -@pytest.fixture() -def handler_mock(netbox_mock, profile_mock): - """Create management handler mock object""" - juniper = Juniper(netbox=netbox_mock) - juniper._profile = profile_mock - yield juniper - - -@pytest.fixture() -def xml(interface1_mock): - """Creates a ElementTree containing poe information for one interface""" - tree_string = f""" - - - {interface1_mock.ifname} - Enabled - - """ - tree = etree.fromstring(tree_string) - yield tree - - -@pytest.fixture() -def xml_bulk(interface1_mock, interface2_mock): - """Creates a ElementTree containing poe information for two interfaces""" - tree_string = f""" - - - {interface1_mock.ifname} - Enabled - - - {interface2_mock.ifname} - Disabled - - """ - tree = etree.fromstring(tree_string) - yield tree - - -@pytest.fixture() -def interface1_mock(): - interface = Mock() - interface.ifname = "ge-0/0/1" - interface.ifindex = 1 - yield interface - - -@pytest.fixture() -def interface2_mock(): - interface = Mock() - interface.ifname = "ge-0/0/2" - interface.ifindex = 2 - yield interface - - class TestWrapUnhandledRpcErrors: def test_rpcerrors_should_become_protocolerrors(self): @wrap_unhandled_rpc_errors @@ -207,3 +132,93 @@ def test_get_poe_states_for_multiple_interfaces_maps_interface_to_none_if_poe_no if_mock.ifindex = 0 states = handler_mock._get_poe_state_for_multiple_interfaces([if_mock]) assert states[if_mock.ifindex] is None + + def test_get_poe_state_uses_interfaces_from_db_if_input_is_none(self): + pass + + def test_get_poe_state_uses_interfaces_from_db_if_input_is_empty(self): + pass + + def test_get_poe_state_raises_exception_if_no_interfaces_in_xml(self): + pass + + def test_get_poe_state_raises_exception_if_multiple_interfaces_in_xml(self): + pass + + def test_get_poe_states_bulk_returns_empty_dict_if_no_interfaces_in_xml(self): + pass + + +@pytest.fixture() +def netbox_mock(): + """Create netbox model mock object""" + netbox = Mock() + netbox.ip = '10.0.0.1' + netbox.type.get_enterprise_id.return_value = VENDOR_ID_JUNIPER_NETWORKS_INC + yield netbox + + +@pytest.fixture() +def profile_mock(): + """Create management profile model mock object""" + profile = Mock() + profile.protocol = manage.ManagementProfile.PROTOCOL_NAPALM + profile.PROTOCOL_NAPALM = manage.ManagementProfile.PROTOCOL_NAPALM + profile.configuration = {"driver": "mock"} + yield profile + + +@pytest.fixture() +def handler_mock(netbox_mock, profile_mock): + """Create management handler mock object""" + juniper = Juniper(netbox=netbox_mock) + juniper._profile = profile_mock + yield juniper + + +@pytest.fixture() +def xml(interface1_mock): + """Creates a ElementTree containing poe information for one interface""" + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def xml_bulk(interface1_mock, interface2_mock): + """Creates a ElementTree containing poe information for two interfaces""" + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + + {interface2_mock.ifname} + Disabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def interface1_mock(): + interface = Mock() + interface.ifname = "ge-0/0/1" + interface.ifindex = 1 + yield interface + + +@pytest.fixture() +def interface2_mock(): + interface = Mock() + interface.ifname = "ge-0/0/2" + interface.ifindex = 2 + yield interface From 4e561b39a9c4a298be556bbb2a2831864581e533 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Thu, 2 Nov 2023 11:02:05 +0100 Subject: [PATCH 15/25] Rename functions --- python/nav/portadmin/napalm/juniper.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 209483318d..d982c4d9da 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -492,16 +492,14 @@ def get_poe_states( if len(interfaces) == 1: interface = interfaces[0] try: - state = self._get_poe_state_for_single_interface(interface) + state = self._get_single_poe_state(interface) except POENotSupportedError: state = None return {interface.ifindex: state} else: - return self._get_poe_state_for_multiple_interfaces(interfaces) + return self._get_poe_states_bulk(interfaces) - def _get_poe_state_for_single_interface( - self, interface: manage.Interface - ) -> PoeState: + def _get_single_poe_state(self, interface: manage.Interface) -> PoeState: tree = self._get_poe_interface_information(ifname=interface.ifname) matching_elements = tree.xpath( "//poe/interface-information-detail/interface-enabled-detail" @@ -518,7 +516,7 @@ def _get_poe_state_for_single_interface( ifenabled = matching_elements[0].text.lower() return self._poe_string_to_state(ifenabled) - def _get_poe_state_for_multiple_interfaces( + def _get_poe_states_bulk( self, interfaces: Sequence[manage.Interface] ) -> Dict[int, Optional[PoeState]]: tree = self._get_all_poe_interface_information() From 684e1330a14930b18577cb05715bbfaae7964fb1 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Thu, 2 Nov 2023 11:11:19 +0100 Subject: [PATCH 16/25] Update tests --- .../portadmin/napalm/juniper_test.py | 112 ++++++++++++++---- 1 file changed, 87 insertions(+), 25 deletions(-) diff --git a/tests/unittests/portadmin/napalm/juniper_test.py b/tests/unittests/portadmin/napalm/juniper_test.py index 4c4d510449..4c7d4720f2 100644 --- a/tests/unittests/portadmin/napalm/juniper_test.py +++ b/tests/unittests/portadmin/napalm/juniper_test.py @@ -27,6 +27,8 @@ DeviceNotConfigurableError, ProtocolError, POEStateNotSupportedError, + POENotSupportedError, + XMLParseError, ) from nav.portadmin.napalm.juniper import wrap_unhandled_rpc_errors, Juniper @@ -106,55 +108,85 @@ def test_state_converter_raises_error_for_invalid_states(self, handler_mock): with pytest.raises(POEStateNotSupportedError): handler_mock._poe_string_to_state("invalid_state") - def test_get_poe_state_for_single_interface_returns_correct_state( + def test_get_poe_states_uses_interfaces_from_db_if_input_is_none( + self, handler_mock, xml_bulk + ): + expected_interfaces = handler_mock.netbox.interfaces + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + return_dict = handler_mock.get_poe_states() + for interface in expected_interfaces: + assert interface.ifindex in return_dict + + def test_get_poe_states_uses_interfaces_from_db_if_input_is_empty( + self, handler_mock, xml_bulk + ): + expected_interfaces = handler_mock.netbox.interfaces + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + return_dict = handler_mock.get_poe_states([]) + for interface in expected_interfaces: + assert interface.ifindex in return_dict + + +class TestGetSinglePoeState: + def test_returns_correct_state_for_interface_that_exists_in_xml_response( self, handler_mock, xml, interface1_mock ): handler_mock._get_poe_interface_information = Mock(return_value=xml) - state = handler_mock._get_poe_state_for_single_interface(interface1_mock) + state = handler_mock._get_single_poe_state(interface1_mock) assert state == Juniper.POE_ENABLED - def test_get_poe_state_for_multiple_interfaces_returns_correct_states( + def test_raises_exception_if_no_interfaces_in_xml( + self, handler_mock, interface1_mock, xml_empty + ): + handler_mock._get_poe_interface_information = Mock(return_value=xml_empty) + with pytest.raises(POENotSupportedError): + handler_mock._get_single_poe_state(interface1_mock) + + def test_raises_exception_if_multiple_interfaces_in_xml( + self, handler_mock, interface1_mock, xml_bulk_wrong_format + ): + handler_mock._get_poe_interface_information = Mock( + return_value=xml_bulk_wrong_format + ) + with pytest.raises(XMLParseError): + handler_mock._get_single_poe_state(interface1_mock) + + +class TestGetPoeStatesBulk: + def test_returns_correct_states( self, handler_mock, xml_bulk, interface1_mock, interface2_mock ): handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) - states = handler_mock._get_poe_state_for_multiple_interfaces( - [interface1_mock, interface2_mock] - ) + states = handler_mock._get_poe_states_bulk([interface1_mock, interface2_mock]) assert states[interface1_mock.ifindex] == Juniper.POE_ENABLED assert states[interface2_mock.ifindex] == Juniper.POE_DISABLED - def test_get_poe_states_for_multiple_interfaces_maps_interface_to_none_if_poe_not_supported( - self, handler_mock, xml_bulk - ): + def test_maps_interface_to_none_if_poe_not_supported(self, handler_mock, xml_bulk): handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) if_mock = Mock() if_mock.ifname == "random_if" if_mock.ifindex = 0 - states = handler_mock._get_poe_state_for_multiple_interfaces([if_mock]) + states = handler_mock._get_poe_states_bulk([if_mock]) assert states[if_mock.ifindex] is None - def test_get_poe_state_uses_interfaces_from_db_if_input_is_none(self): - pass - - def test_get_poe_state_uses_interfaces_from_db_if_input_is_empty(self): - pass - - def test_get_poe_state_raises_exception_if_no_interfaces_in_xml(self): - pass - - def test_get_poe_state_raises_exception_if_multiple_interfaces_in_xml(self): - pass - - def test_get_poe_states_bulk_returns_empty_dict_if_no_interfaces_in_xml(self): - pass + def test_returns_none_values_if_no_interfaces_in_xml( + self, handler_mock, interface1_mock, interface2_mock, xml_empty + ): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_empty) + return_dict = handler_mock._get_poe_states_bulk( + [interface1_mock, interface2_mock] + ) + assert return_dict[interface1_mock.ifindex] is None + assert return_dict[interface2_mock.ifindex] is None @pytest.fixture() -def netbox_mock(): +def netbox_mock(interface1_mock, interface2_mock): """Create netbox model mock object""" netbox = Mock() netbox.ip = '10.0.0.1' netbox.type.get_enterprise_id.return_value = VENDOR_ID_JUNIPER_NETWORKS_INC + netbox.interfaces = [interface1_mock, interface2_mock] yield netbox @@ -190,6 +222,26 @@ def xml(interface1_mock): yield tree +@pytest.fixture() +def xml_bulk_wrong_format(interface1_mock, interface2_mock): + """Creates a ElementTree with the format meant for a single interface in the response, + but it contains poe information for two interfaces + """ + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + + {interface2_mock.ifname} + Enabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + @pytest.fixture() def xml_bulk(interface1_mock, interface2_mock): """Creates a ElementTree containing poe information for two interfaces""" @@ -208,6 +260,16 @@ def xml_bulk(interface1_mock, interface2_mock): yield tree +@pytest.fixture() +def xml_empty(): + """Creates a ElementTree containing no poe state for any interface""" + tree_string = f""" + + """ + tree = etree.fromstring(tree_string) + yield tree + + @pytest.fixture() def interface1_mock(): interface = Mock() From e9cf952198bbe0eea2fb7ca12b33c9025c223f51 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Thu, 2 Nov 2023 11:17:26 +0100 Subject: [PATCH 17/25] Move tests and fixtures into separate files --- tests/unittests/portadmin/napalm/conftest.py | 50 +++++ .../portadmin/napalm/juniper_test.py | 199 +----------------- .../portadmin/napalm/juniper_test_poe.py | 158 ++++++++++++++ 3 files changed, 209 insertions(+), 198 deletions(-) create mode 100644 tests/unittests/portadmin/napalm/conftest.py create mode 100644 tests/unittests/portadmin/napalm/juniper_test_poe.py diff --git a/tests/unittests/portadmin/napalm/conftest.py b/tests/unittests/portadmin/napalm/conftest.py new file mode 100644 index 0000000000..fb0d7394e7 --- /dev/null +++ b/tests/unittests/portadmin/napalm/conftest.py @@ -0,0 +1,50 @@ +import pytest +from unittest.mock import Mock + +from nav.enterprise.ids import VENDOR_ID_JUNIPER_NETWORKS_INC +from nav.models import manage +from nav.portadmin.napalm.juniper import Juniper + + +@pytest.fixture() +def netbox_mock(interface1_mock, interface2_mock): + """Create netbox model mock object""" + netbox = Mock() + netbox.ip = '10.0.0.1' + netbox.type.get_enterprise_id.return_value = VENDOR_ID_JUNIPER_NETWORKS_INC + netbox.interfaces = [interface1_mock, interface2_mock] + yield netbox + + +@pytest.fixture() +def profile_mock(): + """Create management profile model mock object""" + profile = Mock() + profile.protocol = manage.ManagementProfile.PROTOCOL_NAPALM + profile.PROTOCOL_NAPALM = manage.ManagementProfile.PROTOCOL_NAPALM + profile.configuration = {"driver": "mock"} + yield profile + + +@pytest.fixture() +def handler_mock(netbox_mock, profile_mock): + """Create management handler mock object""" + juniper = Juniper(netbox=netbox_mock) + juniper._profile = profile_mock + yield juniper + + +@pytest.fixture() +def interface1_mock(): + interface = Mock() + interface.ifname = "ge-0/0/1" + interface.ifindex = 1 + yield interface + + +@pytest.fixture() +def interface2_mock(): + interface = Mock() + interface.ifname = "ge-0/0/2" + interface.ifindex = 2 + yield interface diff --git a/tests/unittests/portadmin/napalm/juniper_test.py b/tests/unittests/portadmin/napalm/juniper_test.py index 4c7d4720f2..58c088edae 100644 --- a/tests/unittests/portadmin/napalm/juniper_test.py +++ b/tests/unittests/portadmin/napalm/juniper_test.py @@ -19,16 +19,11 @@ from unittest.mock import Mock, patch from jnpr.junos.exception import RpcError -from lxml import etree -from nav.enterprise.ids import VENDOR_ID_RESERVED, VENDOR_ID_JUNIPER_NETWORKS_INC -from nav.models import manage +from nav.enterprise.ids import VENDOR_ID_RESERVED from nav.portadmin.handlers import ( DeviceNotConfigurableError, ProtocolError, - POEStateNotSupportedError, - POENotSupportedError, - XMLParseError, ) from nav.portadmin.napalm.juniper import wrap_unhandled_rpc_errors, Juniper @@ -92,195 +87,3 @@ def vlans(self): m = MockedJuniperHandler(Mock()) assert len(m.get_netbox_vlans()) == 1 - - -class TestJuniperPoe: - def test_returns_correct_state_options(self, handler_mock): - state_options = handler_mock.get_poe_state_options() - assert Juniper.POE_ENABLED in state_options - assert Juniper.POE_DISABLED in state_options - - def test_state_converter_returns_correct_states(self, handler_mock): - assert handler_mock._poe_string_to_state("enabled") == Juniper.POE_ENABLED - assert handler_mock._poe_string_to_state("disabled") == Juniper.POE_DISABLED - - def test_state_converter_raises_error_for_invalid_states(self, handler_mock): - with pytest.raises(POEStateNotSupportedError): - handler_mock._poe_string_to_state("invalid_state") - - def test_get_poe_states_uses_interfaces_from_db_if_input_is_none( - self, handler_mock, xml_bulk - ): - expected_interfaces = handler_mock.netbox.interfaces - handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) - return_dict = handler_mock.get_poe_states() - for interface in expected_interfaces: - assert interface.ifindex in return_dict - - def test_get_poe_states_uses_interfaces_from_db_if_input_is_empty( - self, handler_mock, xml_bulk - ): - expected_interfaces = handler_mock.netbox.interfaces - handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) - return_dict = handler_mock.get_poe_states([]) - for interface in expected_interfaces: - assert interface.ifindex in return_dict - - -class TestGetSinglePoeState: - def test_returns_correct_state_for_interface_that_exists_in_xml_response( - self, handler_mock, xml, interface1_mock - ): - handler_mock._get_poe_interface_information = Mock(return_value=xml) - state = handler_mock._get_single_poe_state(interface1_mock) - assert state == Juniper.POE_ENABLED - - def test_raises_exception_if_no_interfaces_in_xml( - self, handler_mock, interface1_mock, xml_empty - ): - handler_mock._get_poe_interface_information = Mock(return_value=xml_empty) - with pytest.raises(POENotSupportedError): - handler_mock._get_single_poe_state(interface1_mock) - - def test_raises_exception_if_multiple_interfaces_in_xml( - self, handler_mock, interface1_mock, xml_bulk_wrong_format - ): - handler_mock._get_poe_interface_information = Mock( - return_value=xml_bulk_wrong_format - ) - with pytest.raises(XMLParseError): - handler_mock._get_single_poe_state(interface1_mock) - - -class TestGetPoeStatesBulk: - def test_returns_correct_states( - self, handler_mock, xml_bulk, interface1_mock, interface2_mock - ): - handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) - states = handler_mock._get_poe_states_bulk([interface1_mock, interface2_mock]) - assert states[interface1_mock.ifindex] == Juniper.POE_ENABLED - assert states[interface2_mock.ifindex] == Juniper.POE_DISABLED - - def test_maps_interface_to_none_if_poe_not_supported(self, handler_mock, xml_bulk): - handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) - if_mock = Mock() - if_mock.ifname == "random_if" - if_mock.ifindex = 0 - states = handler_mock._get_poe_states_bulk([if_mock]) - assert states[if_mock.ifindex] is None - - def test_returns_none_values_if_no_interfaces_in_xml( - self, handler_mock, interface1_mock, interface2_mock, xml_empty - ): - handler_mock._get_all_poe_interface_information = Mock(return_value=xml_empty) - return_dict = handler_mock._get_poe_states_bulk( - [interface1_mock, interface2_mock] - ) - assert return_dict[interface1_mock.ifindex] is None - assert return_dict[interface2_mock.ifindex] is None - - -@pytest.fixture() -def netbox_mock(interface1_mock, interface2_mock): - """Create netbox model mock object""" - netbox = Mock() - netbox.ip = '10.0.0.1' - netbox.type.get_enterprise_id.return_value = VENDOR_ID_JUNIPER_NETWORKS_INC - netbox.interfaces = [interface1_mock, interface2_mock] - yield netbox - - -@pytest.fixture() -def profile_mock(): - """Create management profile model mock object""" - profile = Mock() - profile.protocol = manage.ManagementProfile.PROTOCOL_NAPALM - profile.PROTOCOL_NAPALM = manage.ManagementProfile.PROTOCOL_NAPALM - profile.configuration = {"driver": "mock"} - yield profile - - -@pytest.fixture() -def handler_mock(netbox_mock, profile_mock): - """Create management handler mock object""" - juniper = Juniper(netbox=netbox_mock) - juniper._profile = profile_mock - yield juniper - - -@pytest.fixture() -def xml(interface1_mock): - """Creates a ElementTree containing poe information for one interface""" - tree_string = f""" - - - {interface1_mock.ifname} - Enabled - - """ - tree = etree.fromstring(tree_string) - yield tree - - -@pytest.fixture() -def xml_bulk_wrong_format(interface1_mock, interface2_mock): - """Creates a ElementTree with the format meant for a single interface in the response, - but it contains poe information for two interfaces - """ - tree_string = f""" - - - {interface1_mock.ifname} - Enabled - - - {interface2_mock.ifname} - Enabled - - """ - tree = etree.fromstring(tree_string) - yield tree - - -@pytest.fixture() -def xml_bulk(interface1_mock, interface2_mock): - """Creates a ElementTree containing poe information for two interfaces""" - tree_string = f""" - - - {interface1_mock.ifname} - Enabled - - - {interface2_mock.ifname} - Disabled - - """ - tree = etree.fromstring(tree_string) - yield tree - - -@pytest.fixture() -def xml_empty(): - """Creates a ElementTree containing no poe state for any interface""" - tree_string = f""" - - """ - tree = etree.fromstring(tree_string) - yield tree - - -@pytest.fixture() -def interface1_mock(): - interface = Mock() - interface.ifname = "ge-0/0/1" - interface.ifindex = 1 - yield interface - - -@pytest.fixture() -def interface2_mock(): - interface = Mock() - interface.ifname = "ge-0/0/2" - interface.ifindex = 2 - yield interface diff --git a/tests/unittests/portadmin/napalm/juniper_test_poe.py b/tests/unittests/portadmin/napalm/juniper_test_poe.py new file mode 100644 index 0000000000..1efb546628 --- /dev/null +++ b/tests/unittests/portadmin/napalm/juniper_test_poe.py @@ -0,0 +1,158 @@ +import pytest +from unittest.mock import Mock + +from lxml import etree + +from nav.portadmin.handlers import ( + POEStateNotSupportedError, + POENotSupportedError, + XMLParseError, +) +from nav.portadmin.napalm.juniper import Juniper + + +def test_returns_correct_state_options(handler_mock): + state_options = handler_mock.get_poe_state_options() + assert Juniper.POE_ENABLED in state_options + assert Juniper.POE_DISABLED in state_options + + +def test_state_converter_returns_correct_states(handler_mock): + assert handler_mock._poe_string_to_state("enabled") == Juniper.POE_ENABLED + assert handler_mock._poe_string_to_state("disabled") == Juniper.POE_DISABLED + + +def test_state_converter_raises_error_for_invalid_states(handler_mock): + with pytest.raises(POEStateNotSupportedError): + handler_mock._poe_string_to_state("invalid_state") + + +class TestGetPoeStates: + def test_interfaces_from_db_is_used_if_input_is_none(self, handler_mock, xml_bulk): + expected_interfaces = handler_mock.netbox.interfaces + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + return_dict = handler_mock.get_poe_states() + for interface in expected_interfaces: + assert interface.ifindex in return_dict + + def test_interfaces_from_db_is_used_if_input_is_empty(self, handler_mock, xml_bulk): + expected_interfaces = handler_mock.netbox.interfaces + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + return_dict = handler_mock.get_poe_states([]) + for interface in expected_interfaces: + assert interface.ifindex in return_dict + + +class TestGetSinglePoeState: + def test_returns_correct_state_for_interface_that_exists_in_xml_response( + self, handler_mock, xml, interface1_mock + ): + handler_mock._get_poe_interface_information = Mock(return_value=xml) + state = handler_mock._get_single_poe_state(interface1_mock) + assert state == Juniper.POE_ENABLED + + def test_raises_exception_if_no_interfaces_in_xml( + self, handler_mock, interface1_mock, xml_empty + ): + handler_mock._get_poe_interface_information = Mock(return_value=xml_empty) + with pytest.raises(POENotSupportedError): + handler_mock._get_single_poe_state(interface1_mock) + + def test_raises_exception_if_multiple_interfaces_in_xml( + self, handler_mock, interface1_mock, xml_bulk_wrong_format + ): + handler_mock._get_poe_interface_information = Mock( + return_value=xml_bulk_wrong_format + ) + with pytest.raises(XMLParseError): + handler_mock._get_single_poe_state(interface1_mock) + + +class TestGetPoeStatesBulk: + def test_returns_correct_states( + self, handler_mock, xml_bulk, interface1_mock, interface2_mock + ): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + states = handler_mock._get_poe_states_bulk([interface1_mock, interface2_mock]) + assert states[interface1_mock.ifindex] == Juniper.POE_ENABLED + assert states[interface2_mock.ifindex] == Juniper.POE_DISABLED + + def test_maps_interface_to_none_if_poe_not_supported(self, handler_mock, xml_bulk): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + if_mock = Mock() + if_mock.ifname == "random_if" + if_mock.ifindex = 0 + states = handler_mock._get_poe_states_bulk([if_mock]) + assert states[if_mock.ifindex] is None + + def test_returns_none_values_if_no_interfaces_in_xml( + self, handler_mock, interface1_mock, interface2_mock, xml_empty + ): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_empty) + return_dict = handler_mock._get_poe_states_bulk( + [interface1_mock, interface2_mock] + ) + assert return_dict[interface1_mock.ifindex] is None + assert return_dict[interface2_mock.ifindex] is None + + +@pytest.fixture() +def xml(interface1_mock): + """Creates a ElementTree containing poe information for one interface""" + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def xml_bulk_wrong_format(interface1_mock, interface2_mock): + """Creates a ElementTree with the format meant for a single interface in the response, + but it contains poe information for two interfaces + """ + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + + {interface2_mock.ifname} + Enabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def xml_bulk(interface1_mock, interface2_mock): + """Creates a ElementTree containing poe information for two interfaces""" + tree_string = f""" + + + {interface1_mock.ifname} + Enabled + + + {interface2_mock.ifname} + Disabled + + """ + tree = etree.fromstring(tree_string) + yield tree + + +@pytest.fixture() +def xml_empty(): + """Creates a ElementTree containing no poe state for any interface""" + tree_string = f""" + + """ + tree = etree.fromstring(tree_string) + yield tree From c6b4bd2e1dbd086ee8e1ef87181b4672de8dbdd9 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Thu, 2 Nov 2023 14:33:29 +0100 Subject: [PATCH 18/25] Fix linelength --- python/nav/portadmin/handlers.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/python/nav/portadmin/handlers.py b/python/nav/portadmin/handlers.py index ec63baca11..3a63c8437d 100644 --- a/python/nav/portadmin/handlers.py +++ b/python/nav/portadmin/handlers.py @@ -341,7 +341,9 @@ class ProtocolError(ManagementError): class POENotSupportedError(ManagementError): - """Raised when an interface that does not support PoE is used in a context where PoE support is expected""" + """Raised when an interface that does not support PoE is used in a context + where PoE support is expected + """ class POEStateNotSupportedError(ManagementError): From 87b25aed8cba95a89d28447b45b0a78dcc98b4d2 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 14:56:54 +0100 Subject: [PATCH 19/25] Make options sequence static --- python/nav/portadmin/napalm/juniper.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index d982c4d9da..f31654f770 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -109,6 +109,7 @@ class Juniper(ManagementHandler): PROTOCOL = manage.ManagementProfile.PROTOCOL_NAPALM POE_ENABLED = PoeState(state=1, name="ENABLED") POE_DISABLED = PoeState(state=2, name="DISABLED") + POE_OPTIONS = [POE_ENABLED, POE_DISABLED] def __init__(self, netbox: manage.Netbox, **kwargs): super().__init__(netbox, **kwargs) @@ -450,7 +451,7 @@ def raise_if_not_configurable(self): def get_poe_state_options(self) -> Sequence[PoeState]: """Returns the available options for enabling/disabling PoE on this netbox""" - return [self.POE_ENABLED, self.POE_DISABLED] + return self.POE_OPTIONS @wrap_unhandled_rpc_errors def set_poe_state(self, interface: manage.Interface, state: PoeState): From 7d8940b83632be28df453db450dbf747c27bd685 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 14:57:51 +0100 Subject: [PATCH 20/25] Make interfaces optional --- python/nav/portadmin/handlers.py | 2 +- python/nav/portadmin/napalm/juniper.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/python/nav/portadmin/handlers.py b/python/nav/portadmin/handlers.py index 3a63c8437d..bd3138af4e 100644 --- a/python/nav/portadmin/handlers.py +++ b/python/nav/portadmin/handlers.py @@ -301,7 +301,7 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): raise NotImplementedError def get_poe_states( - self, interfaces: Sequence[manage.Interface] = None + self, interfaces: Optional[Sequence[manage.Interface]] = None ) -> Dict[int, Optional[PoeState]]: """Retrieves current PoE state for interfaces on this device. diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index f31654f770..d2a2a12142 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -471,7 +471,7 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): self.device.load_merge_candidate(config=config) def get_poe_states( - self, interfaces: Sequence[manage.Interface] = None + self, interfaces: Optional[Sequence[manage.Interface]] = None ) -> Dict[int, Optional[PoeState]]: """Retrieves current PoE state for interfaces on this device. From 5fd54ebc556ed05aace7f8115f88fcc73771d40a Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 15:02:32 +0100 Subject: [PATCH 21/25] Use ifname as key --- python/nav/portadmin/handlers.py | 4 ++-- python/nav/portadmin/napalm/juniper.py | 10 +++++----- .../unittests/portadmin/napalm/juniper_test_poe.py | 14 +++++++------- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/python/nav/portadmin/handlers.py b/python/nav/portadmin/handlers.py index bd3138af4e..7c74e559df 100644 --- a/python/nav/portadmin/handlers.py +++ b/python/nav/portadmin/handlers.py @@ -302,7 +302,7 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): def get_poe_states( self, interfaces: Optional[Sequence[manage.Interface]] = None - ) -> Dict[int, Optional[PoeState]]: + ) -> Dict[str, Optional[PoeState]]: """Retrieves current PoE state for interfaces on this device. :param interfaces: Optional sequence of interfaces to filter for, as fetching @@ -311,7 +311,7 @@ def get_poe_states( the default behavior is to filter on all Interface objects registered for this device. :returns: A dict mapping interfaces to their discovered PoE state. - The key matches the `ifindex` attribute for the related + The key matches the `ifname` attribute for the related Interface object. The value will be None if the interface does not support PoE. """ diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index d2a2a12142..59a5f1ea67 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -472,7 +472,7 @@ def set_poe_state(self, interface: manage.Interface, state: PoeState): def get_poe_states( self, interfaces: Optional[Sequence[manage.Interface]] = None - ) -> Dict[int, Optional[PoeState]]: + ) -> Dict[str, Optional[PoeState]]: """Retrieves current PoE state for interfaces on this device. :param interfaces: Optional sequence of interfaces to filter for, as fetching @@ -481,7 +481,7 @@ def get_poe_states( the default behavior is to filter on all Interface objects registered for this device. :returns: A dict mapping interfaces to their discovered PoE state. - The key matches the `ifindex` attribute for the related + The key matches the `ifname` attribute for the related Interface object. The value will be None if the interface does not support PoE. """ @@ -496,7 +496,7 @@ def get_poe_states( state = self._get_single_poe_state(interface) except POENotSupportedError: state = None - return {interface.ifindex: state} + return {interface.ifname: state} else: return self._get_poe_states_bulk(interfaces) @@ -519,7 +519,7 @@ def _get_single_poe_state(self, interface: manage.Interface) -> PoeState: def _get_poe_states_bulk( self, interfaces: Sequence[manage.Interface] - ) -> Dict[int, Optional[PoeState]]: + ) -> Dict[str, Optional[PoeState]]: tree = self._get_all_poe_interface_information() interface_information_elements = tree.findall(".//interface-information") ifname_to_state_dict = {} @@ -528,7 +528,7 @@ def _get_poe_states_bulk( ifenabled = element.findall(".//interface-enabled")[0].text.strip().lower() ifname_to_state_dict[ifname] = self._poe_string_to_state(ifenabled) ifindex_to_state_dict = { - interface.ifindex: ifname_to_state_dict.get(interface.ifname.lower()) + interface.ifname: ifname_to_state_dict.get(interface.ifname.lower()) for interface in interfaces } return ifindex_to_state_dict diff --git a/tests/unittests/portadmin/napalm/juniper_test_poe.py b/tests/unittests/portadmin/napalm/juniper_test_poe.py index 1efb546628..f0d06865b7 100644 --- a/tests/unittests/portadmin/napalm/juniper_test_poe.py +++ b/tests/unittests/portadmin/napalm/juniper_test_poe.py @@ -33,14 +33,14 @@ def test_interfaces_from_db_is_used_if_input_is_none(self, handler_mock, xml_bul handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) return_dict = handler_mock.get_poe_states() for interface in expected_interfaces: - assert interface.ifindex in return_dict + assert interface.ifname in return_dict def test_interfaces_from_db_is_used_if_input_is_empty(self, handler_mock, xml_bulk): expected_interfaces = handler_mock.netbox.interfaces handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) return_dict = handler_mock.get_poe_states([]) for interface in expected_interfaces: - assert interface.ifindex in return_dict + assert interface.ifname in return_dict class TestGetSinglePoeState: @@ -74,8 +74,8 @@ def test_returns_correct_states( ): handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) states = handler_mock._get_poe_states_bulk([interface1_mock, interface2_mock]) - assert states[interface1_mock.ifindex] == Juniper.POE_ENABLED - assert states[interface2_mock.ifindex] == Juniper.POE_DISABLED + assert states[interface1_mock.ifname] == Juniper.POE_ENABLED + assert states[interface2_mock.ifname] == Juniper.POE_DISABLED def test_maps_interface_to_none_if_poe_not_supported(self, handler_mock, xml_bulk): handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) @@ -83,7 +83,7 @@ def test_maps_interface_to_none_if_poe_not_supported(self, handler_mock, xml_bul if_mock.ifname == "random_if" if_mock.ifindex = 0 states = handler_mock._get_poe_states_bulk([if_mock]) - assert states[if_mock.ifindex] is None + assert states[if_mock.ifname] is None def test_returns_none_values_if_no_interfaces_in_xml( self, handler_mock, interface1_mock, interface2_mock, xml_empty @@ -92,8 +92,8 @@ def test_returns_none_values_if_no_interfaces_in_xml( return_dict = handler_mock._get_poe_states_bulk( [interface1_mock, interface2_mock] ) - assert return_dict[interface1_mock.ifindex] is None - assert return_dict[interface2_mock.ifindex] is None + assert return_dict[interface1_mock.ifname] is None + assert return_dict[interface2_mock.ifname] is None @pytest.fixture() From 4ee4ddc9b3df8d1a268c0769339d59e5339027fa Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 15:04:12 +0100 Subject: [PATCH 22/25] Fix linelength --- python/nav/portadmin/napalm/juniper.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/nav/portadmin/napalm/juniper.py b/python/nav/portadmin/napalm/juniper.py index 59a5f1ea67..bdf3648560 100644 --- a/python/nav/portadmin/napalm/juniper.py +++ b/python/nav/portadmin/napalm/juniper.py @@ -512,7 +512,8 @@ def _get_single_poe_state(self, interface: manage.Interface) -> PoeState: ) if len(matching_elements) != 1: raise XMLParseError( - f"Expected 1 matching element in xml response, {len(matching_elements)} found" + f"Expected 1 matching element in xml response, " + f"{len(matching_elements)} found" ) ifenabled = matching_elements[0].text.lower() return self._poe_string_to_state(ifenabled) From 1f78f3e87551011e9b6eebb9d4d8d5bb5e8be2bb Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 15:13:50 +0100 Subject: [PATCH 23/25] Rename test file test has to be at the end ... --- .../portadmin/napalm/{juniper_test_poe.py => juniper_poe_test.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename tests/unittests/portadmin/napalm/{juniper_test_poe.py => juniper_poe_test.py} (100%) diff --git a/tests/unittests/portadmin/napalm/juniper_test_poe.py b/tests/unittests/portadmin/napalm/juniper_poe_test.py similarity index 100% rename from tests/unittests/portadmin/napalm/juniper_test_poe.py rename to tests/unittests/portadmin/napalm/juniper_poe_test.py From d51b62e15113fd6014433caddb0c8c93cb4d0e24 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 15:30:22 +0100 Subject: [PATCH 24/25] Add more tests --- .../portadmin/napalm/juniper_poe_test.py | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/unittests/portadmin/napalm/juniper_poe_test.py b/tests/unittests/portadmin/napalm/juniper_poe_test.py index f0d06865b7..f0ecf50d5c 100644 --- a/tests/unittests/portadmin/napalm/juniper_poe_test.py +++ b/tests/unittests/portadmin/napalm/juniper_poe_test.py @@ -42,6 +42,28 @@ def test_interfaces_from_db_is_used_if_input_is_empty(self, handler_mock, xml_bu for interface in expected_interfaces: assert interface.ifname in return_dict + def test_returns_empty_dict_if_no_input_and_no_interfaces_in_db( + self, handler_mock, xml_bulk + ): + handler_mock.netbox.interfaces = [] + return_dict = handler_mock.get_poe_states() + assert return_dict == {} + + def test_returns_correct_state_if_input_has_one_interface( + self, handler_mock, xml, interface1_mock + ): + handler_mock._get_poe_interface_information = Mock(return_value=xml) + return_dict = handler_mock.get_poe_states([interface1_mock]) + assert return_dict[interface1_mock.ifname] == handler_mock.POE_ENABLED + + def test_returns_correct_states_if_input_has_multiple_interfaces( + self, handler_mock, xml_bulk, interface1_mock, interface2_mock + ): + handler_mock._get_all_poe_interface_information = Mock(return_value=xml_bulk) + return_dict = handler_mock.get_poe_states([interface1_mock, interface2_mock]) + assert return_dict[interface1_mock.ifname] == Juniper.POE_ENABLED + assert return_dict[interface2_mock.ifname] == Juniper.POE_DISABLED + class TestGetSinglePoeState: def test_returns_correct_state_for_interface_that_exists_in_xml_response( From 9659a7673e973bb00dd15ba0f3783f97c474bd38 Mon Sep 17 00:00:00 2001 From: Simon Oliver Tveit Date: Mon, 6 Nov 2023 16:44:28 +0100 Subject: [PATCH 25/25] Add more tests --- .../portadmin/napalm/juniper_poe_test.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/unittests/portadmin/napalm/juniper_poe_test.py b/tests/unittests/portadmin/napalm/juniper_poe_test.py index f0ecf50d5c..5c38b7ea4c 100644 --- a/tests/unittests/portadmin/napalm/juniper_poe_test.py +++ b/tests/unittests/portadmin/napalm/juniper_poe_test.py @@ -64,6 +64,22 @@ def test_returns_correct_states_if_input_has_multiple_interfaces( assert return_dict[interface1_mock.ifname] == Juniper.POE_ENABLED assert return_dict[interface2_mock.ifname] == Juniper.POE_DISABLED + def test_returns_none_for_single_interface_that_does_not_support_poe( + self, handler_mock, interface1_mock + ): + handler_mock._get_single_poe_state = Mock(side_effect=POENotSupportedError) + return_dict = handler_mock.get_poe_states([interface1_mock]) + assert return_dict[interface1_mock.ifname] is None + + def test_returns_none_for_multiple_interfaces_that_does_not_support_poe( + self, handler_mock, interface1_mock, interface2_mock + ): + bulk_return_dict = {interface1_mock.ifname: None, interface2_mock.ifname: None} + handler_mock._get_poe_states_bulk = Mock(return_value=bulk_return_dict) + return_dict = handler_mock.get_poe_states([interface1_mock, interface2_mock]) + assert return_dict[interface1_mock.ifname] is None + assert return_dict[interface2_mock.ifname] is None + class TestGetSinglePoeState: def test_returns_correct_state_for_interface_that_exists_in_xml_response(