Merge pull request #2460 from johannaengland/napalm

 Timeout values should be configurable in NAPALM profiles

python/nav/napalm.py

@@ -23,6 +23,7 @@
 
 from nav.models import manage
 
+DEFAULT_TIMEOUT_SECONDS = 60
 
 Host = TypeVar("Host", str, manage.Netbox)
 _logger = logging.getLogger(__name__)
@@ -33,14 +34,18 @@ def connect(host: Host, profile: manage.ManagementProfile) -> NetworkDriver:
     driver = get_driver(profile)
     config = profile.configuration
     hostname = host if not isinstance(host, manage.Netbox) else host.ip
-    optional_args = {"config_lock": True, "lock_disable": True}
+    optional_args = {
+        "config_lock": True,
+        "lock_disable": True,
+    }
     key_file = _write_key_to_temporary_file(config, optional_args)
 
     try:
         device = driver(
             hostname=hostname,
             username=config.get("username"),
             password=config.get("password"),
+            timeout=config.get("timeout", DEFAULT_TIMEOUT_SECONDS),
             optional_args=optional_args,
         )
         # Let temporary file live as long as the device connection exists

python/nav/portadmin/napalm/juniper.py

@@ -115,9 +115,7 @@ def __init__(self, netbox: manage.Netbox, **kwargs):
     def profile(self) -> manage.ManagementProfile:
         """Returns the selected NAPALM profile for this netbox"""
         if not self._profile:
-            profiles = self.netbox.profiles.filter(protocol=self.PROTOCOL)
-            if profiles:
-                self._profile = profiles[0]
+            self._profile = self.netbox.profiles.filter(protocol=self.PROTOCOL).first()
         return self._profile
 
     @property

python/nav/web/seeddb/page/management_profile/forms.py

@@ -107,6 +107,7 @@ class Meta(object):
             "private_key",
             "use_keys",
             "alternate_port",
+            "timeout",
         ]
         fields = []
 
@@ -135,6 +136,12 @@ class Meta(object):
         min_value=1,
         max_value=65535,
     )
+    timeout = forms.IntegerField(
+        required=False,
+        help_text="Timeout value in seconds",
+        min_value=1,
+        max_value=600,
+    )
 
 
 FORM_MAPPING = {

tests/unittests/napalm_test.py

@@ -0,0 +1,66 @@
+#
+# 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 unittest.mock import Mock
+
+from nav.models import manage
+from nav.napalm import connect, NapalmError
+
+
+@pytest.fixture()
+def netbox_mock():
+    """Create netbox model mock object"""
+    netbox = Mock()
+    netbox.ip = '10.0.0.1'
+    return 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"}
+    return profile
+
+
+class TestNapalm:
+    def test_napalm_connect_runs_without_errors_for_correct_netbox_and_profile(
+        self, netbox_mock, profile_mock
+    ):
+        device = connect(
+            host=netbox_mock,
+            profile=profile_mock,
+        )
+
+        assert device
+
+    def test_napalm_connect_throws_exception_for_snmp_profile(
+        self, netbox_mock, profile_mock
+    ):
+        profile_mock.protocol = manage.ManagementProfile.PROTOCOL_SNMP
+
+        with pytest.raises(NapalmError):
+            connect(host=netbox_mock, profile=profile_mock)
+
+    def test_napalm_connect_throws_exception_for_profile_without_driver(
+        self, netbox_mock, profile_mock
+    ):
+        profile_mock.configuration = {}
+        with pytest.raises(NapalmError):
+            connect(host=netbox_mock, profile=profile_mock)

tests/unittests/portadmin/napalm/juniper_test.py

@@ -13,16 +13,38 @@
 # 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/>.
 #
-from unittest.mock import Mock, patch
 
+import napalm
 import pytest
+from unittest.mock import Mock, patch
 
 from jnpr.junos.exception import RpcError
 
-from nav.portadmin.handlers import ProtocolError
+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.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
+
+
 class TestWrapUnhandledRpcErrors:
     def test_rpcerrors_should_become_protocolerrors(self):
         @wrap_unhandled_rpc_errors
@@ -42,6 +64,37 @@ def wrapped_function():
 
 
 class TestJuniper:
+    def test_juniper_device_returns_device_connection(self, netbox_mock, profile_mock):
+        driver = napalm.get_network_driver('mock')
+        device = driver(
+            hostname='foo',
+            username='user',
+            password='pass',
+            optional_args={},
+        )
+        device.open()
+        juniper = Juniper(netbox=netbox_mock)
+        juniper._profile = profile_mock
+
+        assert juniper.device
+
+    def test_juniper_device_raises_error_if_vendor_not_juniper(
+        self, netbox_mock, profile_mock
+    ):
+        netbox_mock.type.get_enterprise_id.return_value = VENDOR_ID_RESERVED
+        juniper = Juniper(netbox=netbox_mock)
+        juniper._profile = profile_mock
+
+        with pytest.raises(DeviceNotConfigurableError):
+            juniper.device
+
+    def test_juniper_device_raises_error_if_no_connected_profile(self, netbox_mock):
+        juniper = Juniper(netbox=netbox_mock)
+        netbox_mock.profiles.filter.return_value.first.return_value = None
+
+        with pytest.raises(DeviceNotConfigurableError):
+            juniper.device
+
     @patch('nav.models.manage.Vlan.objects', Mock(return_value=[]))
     def test_get_netbox_vlans_should_ignore_vlans_with_non_integer_tags(self):
         """Regression test for #2452"""