diff --git a/networkapi/plugins/Juniper/JUNOS/plugin.py b/networkapi/plugins/Juniper/JUNOS/plugin.py index b38d891a0..ec8a29b44 100644 --- a/networkapi/plugins/Juniper/JUNOS/plugin.py +++ b/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. 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..79add3b3a --- /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', 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)) diff --git a/networkapi/plugins/Juniper/JUNOS/tests.py b/networkapi/plugins/Juniper/JUNOS/tests.py index 2b2e3e9a7..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,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')