Merge pull request #292 from Malinoski/develop

Junos plugin updates

networkapi/plugins/Juniper/JUNOS/plugin.py

@@ -40,11 +40,13 @@
 
 
 class JUNOS(BasePlugin):
+
     configuration = None
     quantity_of_times_to_try_lock = 3
     seconds_to_wait_to_try_lock = 10
     alternative_variable_base_path_list = ['path_to_tftpboot']
     alternative_static_base_path_list = ['/mnt/scripts/tftpboot/']
+    ignore_warning_list = ['statement not found']
 
     def __init__(self, **kwargs):
         super(JUNOS, self).__init__(connect_port=830, **kwargs)
@@ -89,7 +91,7 @@ def connect(self):
 
         except ConnectError as e:
             log.error("Could not connect to Juniper host {}: {}".format(self.equipment_access.fqdn, e))
-            raise ConnectError
+            raise ConnectError(e)
 
         except Exception, e:
             log.error("Unknown error while connecting to host {}: {}".format(self.equipment_access.fqdn, e))
@@ -119,7 +121,7 @@ def close(self):
             log.error("Unknown error while closing connection on host {}: {}".format(self.equipment_access.fqdn, e))
             raise Exception
 
-    def copyScriptFileToConfig(self, filename, use_vrf='', destination=''):
+    def copyScriptFileToConfig(self, filename):
 
         """
         Receives the file path (usually in /mnt/scripts/tftpboot/networkapi/generated_config/interface/)
@@ -156,7 +158,7 @@ def copyScriptFileToConfig(self, filename, use_vrf='', destination=''):
             self.close()
             raise Exception
 
-    def exec_command(self, command, success_regex='', invalid_regex=None, error_regex=None):
+    def exec_command(self, command):
 
         """
         Execute a junos command 'set' in the equipment.
@@ -175,7 +177,7 @@ def exec_command(self, command, success_regex='', invalid_regex=None, error_rege
         try:
             self.__try_lock()
             self.configuration.rollback()
-            self.configuration.load(command, format='set')
+            self.configuration.load(command, format='set', ignore_warning=self.ignore_warning_list)
             self.configuration.commit_check()
             self.configuration.commit()
             self.configuration.unlock()
@@ -227,7 +229,7 @@ def exec_command(self, command, success_regex='', invalid_regex=None, error_rege
             self.close()
             raise Exception
 
-    def ensure_privilege_level(self, privilege_level=None):
+    def ensure_privilege_level(self):
 
         """
         Ensure privilege level verifying if the current user is super-user.

networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py

@@ -0,0 +1,60 @@
+"""
+How to use:
+python sample_command_line.py -device 'HOST' -user 'SSH_USER_NAME' -password 'SSH_USER_PASSWORD' -command 'COMMAND'
+"""
+
+from lxml import etree
+from jnpr.junos import Device
+from jnpr.junos.utils.config import Config
+import argparse
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-device', help='Input host', type=str)
+parser.add_argument('-user', help='Input user name', type=str)
+parser.add_argument('-password', help='Input password', type=str)
+parser.add_argument('-command', help='Input command', type=str)
+args, unknown = parser.parse_known_args()
+
+device = args.device
+user = args.user
+password = args.password
+command = args.command
+
+try:
+
+    dev = Device(host=device, user=user, password=password, gather_facts=False)
+    open_result = dev.open()
+    print("Open connection ... {}".format(open_result.connected))
+
+    conf = Config(dev)
+    print("Load config  ...")
+
+    lock_response = conf.lock()
+    print("Locking config ... {}".format(lock_response))
+
+    rollback_response = conf.rollback()
+    print("Rollback config ... {}".format(rollback_response))
+
+    load_result = conf.load(command, format='set', ignore_warning=['statement not found'])
+    load_result_tostring = etree.tostring(load_result, encoding='unicode', pretty_print=True)
+    print("Load command ... \n{}".format(load_result_tostring))
+
+    commit_check_result = conf.commit_check()
+    print("Check result ... {}".format(commit_check_result))
+
+    commit_result = conf.commit()
+    print("Commit  ... {}".format(commit_result))
+
+    unlock_response = conf.unlock()
+    print("Unlocking config ... {}".format(unlock_response))
+
+    close_response = dev.close()
+    print("Close connection ... {}".format("Success" if not open_result.connected else "Failed"))
+
+    print ("DONE")
+
+except Exception, e:
+
+    print(e)
+    close_response = dev.close()
+    print("Closed connection? {}".format(not open_result.connected))

networkapi/plugins/Juniper/JUNOS/tests.py

@@ -1,8 +1,8 @@
 from networkapi.test.test_case import NetworkApiTestCase
 from networkapi.plugins.base import BasePlugin
 from networkapi.plugins.Juniper.JUNOS.plugin import JUNOS
-import mock
 from mock import patch, MagicMock
+from jnpr.junos.exception import ConnectError, LockError
 
 
 class JunosPluginTest(NetworkApiTestCase):
@@ -64,7 +64,17 @@ def test_connect_success(self, mock_device):
         self.assertIsNotNone(plugin.configuration)
         self.assertEqual(connection_response, True)
 
-    @patch('jnpr.junos.utils.config.Config', autospec=True)
+    def test_connect_wrong_data_exception(self):
+
+        """
+        test_connect_wrong_data_exception
+        """
+
+        plugin = JUNOS(equipment_access=self.mock_equipment_access)
+        with self.assertRaises(ConnectError):
+            plugin.connect()
+
+    @patch('jnpr.junos.utils.config.Config')
     def test_exec_command_success(self, mock_config):
 
         """
@@ -82,11 +92,11 @@ def test_exec_command_success(self, mock_config):
         exec_command_response = plugin.exec_command("any command")
 
         # Assert
-        plugin.configuration.rollback.assert_called_once_with()
-        plugin.configuration.load.assert_called_once_with("any command", format='set')
-        plugin.configuration.commit_check.assert_called_once_with()
-        plugin.configuration.commit.assert_called_once_with()
-        plugin.configuration.unlock.assert_called_once_with()
+        plugin.configuration.rollback.assert_called_once()
+        plugin.configuration.load.assert_called_once()
+        plugin.configuration.commit_check.assert_called_once()
+        plugin.configuration.commit.assert_called_once()
+        plugin.configuration.unlock.assert_called_once()
         self.assertIsNotNone(exec_command_response)
 
     @patch('jnpr.junos.Device')
@@ -113,10 +123,28 @@ def test_call_copyScriptFileToConfig(self, mock_junos_plugin):
         mock_junos_plugin.copyScriptFileToConfig("any file path")
         mock_junos_plugin.copyScriptFileToConfig.assert_called_with("any file path")
 
-    @patch('networkapi.plugins.Juniper.JUNOS.plugin.JUNOS', autospec=True)
-    def test_call_ensure_privilege_level(self, mock_junos_plugin):
-        mock_junos_plugin.ensure_privilege_level()
-        mock_junos_plugin.ensure_privilege_level.assert_called_with()
+    @patch('networkapi.plugins.Juniper.JUNOS.plugin.StartShell')
+    def test_call_ensure_privilege_level_success(self, mock_start_shell):
+
+        """
+        test_call_ensure_privilege_level_success
+
+        Note: The shell run function expects an array as a return value,
+        and ensure_privilege_level() parse it to ensure the privilege.
+        """
+
+        mock_start_shell.return_value.run.return_value = [
+            False,
+            u'cli -c "show cli authorization"\r\r\nCurrent user: \'root        \' class \'super-user\'\r']
+
+        plugin = JUNOS(equipment_access=self.mock_equipment_access)
+        result = plugin.ensure_privilege_level()
+        self.assertTrue(result)
+
+    def test_call_ensure_privilege_level_fail(self):
+        plugin = JUNOS(equipment_access=self.mock_equipment_access)
+        with self.assertRaises(Exception):
+            plugin.ensure_privilege_level()
 
     @patch('os.path.isfile')
     @patch('networkapi.plugins.Juniper.JUNOS.plugin.get_value')