From 3d08a918e14485532c05f07771a49de7fdef1497 Mon Sep 17 00:00:00 2001 From: malinoski Date: Fri, 23 Oct 2020 13:51:34 -0300 Subject: [PATCH 1/5] Created a script python witch help to test Junos commands from terminal --- .../JUNOS/samples/sample_command_line.py | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py diff --git a/networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py b/networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py new file mode 100644 index 000000000..f411a7bd6 --- /dev/null +++ b/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') + 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)) From 6b70a357f9338b72b880ed7a65f8e51b6e61d84f Mon Sep 17 00:00:00 2001 From: malinoski Date: Mon, 26 Oct 2020 13:06:30 -0300 Subject: [PATCH 2/5] Included test cases to assert success or exception on ensure privilege method --- networkapi/plugins/Juniper/JUNOS/tests.py | 26 +++++++++++++++++++---- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/networkapi/plugins/Juniper/JUNOS/tests.py b/networkapi/plugins/Juniper/JUNOS/tests.py index 2b2e3e9a7..f4d2563f5 100644 --- a/networkapi/plugins/Juniper/JUNOS/tests.py +++ b/networkapi/plugins/Juniper/JUNOS/tests.py @@ -113,10 +113,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') From a1ac883eb8f07327184683d0dbc6fe62164d7c76 Mon Sep 17 00:00:00 2001 From: malinoski Date: Tue, 27 Oct 2020 11:28:16 -0300 Subject: [PATCH 3/5] Junos plugin was updated with: ignore warning commands, in a programaticaly way; removed unsused function's parameters (inherited ones); refactored class attributes in proper way. The test cases and sample codes was updated accordly --- networkapi/plugins/Juniper/JUNOS/plugin.py | 20 ++++++++++--------- .../JUNOS/samples/sample_command_line.py | 2 +- networkapi/plugins/Juniper/JUNOS/tests.py | 12 +++++------ 3 files changed, 18 insertions(+), 16 deletions(-) diff --git a/networkapi/plugins/Juniper/JUNOS/plugin.py b/networkapi/plugins/Juniper/JUNOS/plugin.py index b38d891a0..bbffabc99 100644 --- a/networkapi/plugins/Juniper/JUNOS/plugin.py +++ b/networkapi/plugins/Juniper/JUNOS/plugin.py @@ -40,11 +40,6 @@ 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/'] def __init__(self, **kwargs): super(JUNOS, self).__init__(connect_port=830, **kwargs) @@ -54,6 +49,13 @@ def __init__(self, **kwargs): if 'seconds_to_wait_to_try_lock' in kwargs: self.seconds_to_wait_to_try_lock = kwargs.get('seconds_to_wait_to_try_lock') + self.configuration = None + self.quantity_of_times_to_try_lock = 3 + self.seconds_to_wait_to_try_lock = 10 + self.alternative_variable_base_path_list = ['path_to_tftpboot'] + self.alternative_static_base_path_list = ['/mnt/scripts/tftpboot/'] + self.ignore_warning_list = ['statement not found'] + def connect(self): """ @@ -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. diff --git a/networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py b/networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py index f411a7bd6..79add3b3a 100644 --- a/networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py +++ b/networkapi/plugins/Juniper/JUNOS/samples/sample_command_line.py @@ -35,7 +35,7 @@ rollback_response = conf.rollback() print("Rollback config ... {}".format(rollback_response)) - load_result = conf.load(command, format='set') + 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)) diff --git a/networkapi/plugins/Juniper/JUNOS/tests.py b/networkapi/plugins/Juniper/JUNOS/tests.py index f4d2563f5..aac040b60 100644 --- a/networkapi/plugins/Juniper/JUNOS/tests.py +++ b/networkapi/plugins/Juniper/JUNOS/tests.py @@ -64,7 +64,7 @@ def test_connect_success(self, mock_device): self.assertIsNotNone(plugin.configuration) self.assertEqual(connection_response, True) - @patch('jnpr.junos.utils.config.Config', autospec=True) + @patch('jnpr.junos.utils.config.Config') def test_exec_command_success(self, mock_config): """ @@ -82,11 +82,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') From 5167356bcc8ebd6149d8f418a7f524900a2036e9 Mon Sep 17 00:00:00 2001 From: malinoski Date: Tue, 27 Oct 2020 15:01:19 -0300 Subject: [PATCH 4/5] Included junit test case for connect exception and plugin was updated accordingly --- networkapi/plugins/Juniper/JUNOS/plugin.py | 2 +- networkapi/plugins/Juniper/JUNOS/tests.py | 12 +++++++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/networkapi/plugins/Juniper/JUNOS/plugin.py b/networkapi/plugins/Juniper/JUNOS/plugin.py index bbffabc99..0e4d6125f 100644 --- a/networkapi/plugins/Juniper/JUNOS/plugin.py +++ b/networkapi/plugins/Juniper/JUNOS/plugin.py @@ -91,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)) diff --git a/networkapi/plugins/Juniper/JUNOS/tests.py b/networkapi/plugins/Juniper/JUNOS/tests.py index aac040b60..6dbf67910 100644 --- a/networkapi/plugins/Juniper/JUNOS/tests.py +++ b/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,6 +64,16 @@ def test_connect_success(self, mock_device): self.assertIsNotNone(plugin.configuration) self.assertEqual(connection_response, 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): From 6c6f86d32ea2c43fbe1d94b78c36b9d3b05fd419 Mon Sep 17 00:00:00 2001 From: malinoski Date: Tue, 27 Oct 2020 17:33:46 -0300 Subject: [PATCH 5/5] Junos plugin was reverted to use variablea like before, to follow the superclass architecture --- networkapi/plugins/Juniper/JUNOS/plugin.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/networkapi/plugins/Juniper/JUNOS/plugin.py b/networkapi/plugins/Juniper/JUNOS/plugin.py index 0e4d6125f..ec8a29b44 100644 --- a/networkapi/plugins/Juniper/JUNOS/plugin.py +++ b/networkapi/plugins/Juniper/JUNOS/plugin.py @@ -41,6 +41,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) if 'quantity_of_times_to_try_lock' in kwargs: @@ -49,13 +56,6 @@ def __init__(self, **kwargs): if 'seconds_to_wait_to_try_lock' in kwargs: self.seconds_to_wait_to_try_lock = kwargs.get('seconds_to_wait_to_try_lock') - self.configuration = None - self.quantity_of_times_to_try_lock = 3 - self.seconds_to_wait_to_try_lock = 10 - self.alternative_variable_base_path_list = ['path_to_tftpboot'] - self.alternative_static_base_path_list = ['/mnt/scripts/tftpboot/'] - self.ignore_warning_list = ['statement not found'] - def connect(self): """