Merge pull request #2363 from lunkwill42/bugfix/prettier-rpc-errors

Wrap RpcErrors in PortAdmin ProtocolErrors, for more user-friendly error reporting

python/nav/portadmin/napalm/juniper.py

@@ -33,6 +33,7 @@
 from django.template.loader import get_template
 from napalm.base.exceptions import ConnectAuthError, ConnectionException
 from jnpr.junos.op.vlan import VlanTable
+from jnpr.junos.exception import RpcError
 
 from nav.napalm import connect as napalm_connect
 from nav.enterprise.ids import VENDOR_ID_JUNIPER_NETWORKS_INC
@@ -42,6 +43,7 @@
     DeviceNotConfigurableError,
     AuthenticationError,
     NoResponseError,
+    ProtocolError,
 )
 from nav.junos.nav_views import (
     EthernetSwitchingInterfaceTable,
@@ -61,6 +63,21 @@
 SNMP_STATUS_MAP = {"up": 1, "down": 2, True: 1, False: 2}
 
 
+def wrap_unhandled_rpc_errors(func):
+    """Decorates RPC-enabled handler function to ensure unhandled RpcErrors are
+    translated into ProtocolErrors, which can be reported nicely to the end user by
+    the PortAdmin framework
+    """
+
+    def wrap_rpc_errors(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except RpcError as error:
+            raise ProtocolError(f"Device raised RpcError: {error.message}") from error
+
+    return wrap_rpc_errors
+
+
 class Juniper(ManagementHandler):
     """Juniper specific version of a Napalm PortAdmin handler.
 
@@ -231,6 +248,7 @@ def get_native_and_trunked_vlans(self, interface) -> Tuple[int, List[int]]:
         untagged = first_true(vlans, pred=lambda vlan: not vlan.tagged)
         return (untagged.tag if untagged else None), tagged
 
+    @wrap_unhandled_rpc_errors
     def set_interface_description(self, interface: manage.Interface, description: str):
         # never set description on units but on master interface
         master, _ = split_master_unit(interface.ifname)
@@ -243,9 +261,11 @@ def set_interface_description(self, interface: manage.Interface, description: st
         config = template.render(context)
         self.device.load_merge_candidate(config=config)
 
+    @wrap_unhandled_rpc_errors
     def set_vlan(self, interface: manage.Interface, vlan: int):
         self.set_access(interface, vlan)
 
+    @wrap_unhandled_rpc_errors
     def set_access(self, interface: manage.Interface, access_vlan: int):
         master, unit = split_master_unit(interface.ifname)
         current = InterfaceConfigTable(self.device.device).get(master)[master]
@@ -273,6 +293,7 @@ def _save_access_interface(interface: manage.Interface, access_vlan: int):
             pass
         interface.save()
 
+    @wrap_unhandled_rpc_errors
     def set_trunk(
         self, interface: manage.Interface, native_vlan: int, trunk_vlans: Sequence[int]
     ):
@@ -323,6 +344,7 @@ def cycle_interfaces(
         # and that operation will likely delay at least as much as the wait would have
         return super().cycle_interfaces(interfaces=interfaces, wait=0, commit=True)
 
+    @wrap_unhandled_rpc_errors
     def set_interface_down(self, interface: manage.Interface):
         # does not set oper on logical units, only on physical masters
         master, _unit = split_master_unit(interface.ifname)
@@ -332,6 +354,7 @@ def set_interface_down(self, interface: manage.Interface):
 
         self._save_interface_oper(interface, interface.OPER_DOWN)
 
+    @wrap_unhandled_rpc_errors
     def set_interface_up(self, interface: manage.Interface):
         # does not set oper on logical units, only on physical masters
         master, _unit = split_master_unit(interface.ifname)
@@ -351,6 +374,7 @@ def _save_interface_oper(interface: manage.Interface, ifoperstatus: int):
             )
             master_interface.update(ifoperstatus=ifoperstatus)
 
+    @wrap_unhandled_rpc_errors
     def commit_configuration(self):
         # Only take our sweet time to commit if there are pending changes
         if self.device.compare_config():

tests/unittests/portadmin/napalm/juniper_test.py

@@ -0,0 +1,39 @@
+#
+# Copyright (C) 2022 Sikt AS
+#
+# This file is part of Network Administration Visualized (NAV).
+#
+# NAV is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License version 3 as published by
+# the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+# more details.  You should have received a copy of the GNU General Public
+# License along with NAV. If not, see <http://www.gnu.org/licenses/>.
+#
+import pytest
+
+from jnpr.junos.exception import RpcError
+
+from nav.portadmin.handlers import ProtocolError
+from nav.portadmin.napalm.juniper import wrap_unhandled_rpc_errors
+
+
+class TestWrapUnhandledRpcErrors:
+    def test_rpcerrors_should_become_protocolerrors(self):
+        @wrap_unhandled_rpc_errors
+        def wrapped_function():
+            raise RpcError("bogus")
+
+        with pytest.raises(ProtocolError):
+            wrapped_function()
+
+    def test_non_rpcerrors_should_pass_through(self):
+        @wrap_unhandled_rpc_errors
+        def wrapped_function():
+            raise TypeError("bogus")
+
+        with pytest.raises(TypeError):
+            wrapped_function()