saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #52401 from rsmekala/2019.2.1

Browse source 
Port Junos-related bug fixes from develop to 2019.2
This commit is contained in:
Daniel Wozniak 2019-07-06 14:50:50 -07:00 • committed by GitHub
commit 1a76e00e14
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 438 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

175
salt/modules/junos.py
View file

@ -18,6 +18,7 @@ Refer to :mod:`junos <salt.proxy.junos>` for information on connecting to junos
from __future__ import absolute_import, print_function, unicode_literals
import logging
import os
from functools import wraps
try:
from lxml import etree
@ -69,6 +70,29 @@ def __virtual__():
'junos-eznc or jxmlease or proxy could not be loaded.')
def timeoutDecorator(function):
@wraps(function)
def wrapper(*args, **kwargs):
if 'dev_timeout' in kwargs:
conn = __proxy__['junos.conn']()
restore_timeout = conn.timeout
conn.timeout = kwargs.pop('dev_timeout', None)
try:
result = function(*args, **kwargs)
conn.timeout = restore_timeout
return result
except Exception:
conn.timeout = restore_timeout
raise
else:
try:
return function(*args, **kwargs)
except Exception:
raise
return wrapper
def facts_refresh():
'''
Reload the facts dictionary from the device. Usually only needed if,
@ -82,7 +106,7 @@ def facts_refresh():
salt 'device_name' junos.facts_refresh
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
try:
conn.facts_refresh()
@ -111,7 +135,7 @@ def facts():
salt 'device_name' junos.facts
'''
ret = dict()
ret = {}
try:
ret['facts'] = __proxy__['junos.get_serialized_facts']()
ret['out'] = True
@ -122,6 +146,7 @@ def facts():
return ret
@timeoutDecorator
def rpc(cmd=None, dest=None, **kwargs):
'''
This function executes the RPC provided as arguments on the junos device.
@ -160,7 +185,7 @@ def rpc(cmd=None, dest=None, **kwargs):
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
if cmd is None:
@ -183,7 +208,6 @@ def rpc(cmd=None, dest=None, **kwargs):
op[key] = value
else:
op.update(kwargs)
op['dev_timeout'] = six.text_type(op.pop('timeout', conn.timeout))
if cmd in ['get-config', 'get_config']:
filter_reply = None
@ -204,7 +228,6 @@ def rpc(cmd=None, dest=None, **kwargs):
ret['out'] = False
return ret
else:
op['dev_timeout'] = int(op['dev_timeout'])
if 'filter' in op:
log.warning(
'Filter ignored as it is only used with "get-config" rpc')
@ -242,6 +265,7 @@ def rpc(cmd=None, dest=None, **kwargs):
return ret
@timeoutDecorator
def set_hostname(hostname=None, **kwargs):
'''
Set the device's hostname
@ -267,7 +291,7 @@ def set_hostname(hostname=None, **kwargs):
salt 'device_name' junos.set_hostname salt-device
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
if hostname is None:
ret['message'] = 'Please provide the hostname.'
ret['out'] = False
@ -318,6 +342,7 @@ def set_hostname(hostname=None, **kwargs):
return ret
@timeoutDecorator
def commit(**kwargs):
'''
To commit the changes loaded in the candidate configuration.
@ -403,6 +428,7 @@ def commit(**kwargs):
return ret
@timeoutDecorator
def rollback(**kwargs):
'''
Roll back the last committed configuration changes and commit
@ -435,7 +461,7 @@ def rollback(**kwargs):
'''
id_ = kwargs.pop('id', 0)
ret = dict()
ret = {}
conn = __proxy__['junos.conn']()
op = dict()
@ -512,7 +538,7 @@ def diff(**kwargs):
salt.utils.args.invalid_kwargs(kwargs)
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
try:
ret['message'] = conn.cu.diff(rb_id=id_)
@ -524,6 +550,7 @@ def diff(**kwargs):
return ret
@timeoutDecorator
def ping(dest_ip=None, **kwargs):
'''
Send a ping RPC to a device
@ -558,7 +585,7 @@ def ping(dest_ip=None, **kwargs):
salt 'device_name' junos.ping '8.8.8.8' ttl=1 rapid=True
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
if dest_ip is None:
ret['message'] = 'Please specify the destination ip to ping.'
@ -586,6 +613,7 @@ def ping(dest_ip=None, **kwargs):
return ret
@timeoutDecorator
def cli(command=None, **kwargs):
'''
Executes the CLI commands and returns the output in specified format. \
@ -619,7 +647,7 @@ def cli(command=None, **kwargs):
if not format_:
format_ = 'text'
ret = dict()
ret = {}
if command is None:
ret['message'] = 'Please provide the CLI command to be executed.'
ret['out'] = False
@ -688,10 +716,10 @@ def shutdown(**kwargs):
salt 'device_name' junos.shutdown shutdown=True
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
sw = SW(conn)
op = dict()
op = {}
if '__pub_arg' in kwargs:
if kwargs['__pub_arg']:
if isinstance(kwargs['__pub_arg'][-1], dict):
@ -729,6 +757,7 @@ def shutdown(**kwargs):
return ret
@timeoutDecorator
def install_config(path=None, **kwargs):
'''
Installs the given configuration file into the candidate configuration.
@ -803,7 +832,7 @@ def install_config(path=None, **kwargs):
salt 'device_name' junos.install_config 'salt://syslog_template.conf' template_vars='{"syslog_host": "10.180.222.7"}'
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
if path is None:
@ -812,7 +841,7 @@ def install_config(path=None, **kwargs):
ret['out'] = False
return ret
op = dict()
op = {}
if '__pub_arg' in kwargs:
if kwargs['__pub_arg']:
if isinstance(kwargs['__pub_arg'][-1], dict):
@ -820,7 +849,8 @@ def install_config(path=None, **kwargs):
else:
op.update(kwargs)
template_vars = dict()
test = op.pop('test', False)
template_vars = {}
if "template_vars" in op:
template_vars = op["template_vars"]
@ -874,7 +904,7 @@ def install_config(path=None, **kwargs):
except Exception as exception:
ret['message'] = 'Could not load configuration due to : "{0}"'.format(
exception)
ret['format'] = template_format
ret['format'] = op['format']
ret['out'] = False
return ret
@ -903,7 +933,7 @@ def install_config(path=None, **kwargs):
ret['out'] = False
return ret
if check:
if check and not test:
try:
cu.commit(**commit_params)
ret['message'] = 'Successfully loaded and committed!'
@ -913,10 +943,14 @@ def install_config(path=None, **kwargs):
.format(exception)
ret['out'] = False
return ret
else:
ret['message'] = 'Loaded configuration but commit check failed.'
ret['out'] = False
elif not check:
cu.rollback()
ret['message'] = 'Loaded configuration but commit check failed, hence rolling back configuration.'
ret['out'] = False
else:
cu.rollback()
ret['message'] = 'Commit check passed, but skipping commit for dry-run and rolling back configuration.'
ret['out'] = True
try:
if write_diff and config_diff is not None:
@ -941,7 +975,7 @@ def zeroize():
salt 'device_name' junos.zeroize
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
try:
conn.cli('request system zeroize')
@ -953,6 +987,7 @@ def zeroize():
return ret
@timeoutDecorator
def install_os(path=None, **kwargs):
'''
Installs the given image on the device. After the installation is complete\
@ -962,8 +997,19 @@ def install_os(path=None, **kwargs):
path (required)
Path where the image file is present on the proxy minion
remote_path :
If the value of path is a file path on the local
(Salt host's) filesystem, then the image is copied from the local
filesystem to the :remote_path: directory on the target Junos
device. The default is ``/var/tmp``. If the value of :path: or
is a URL, then the value of :remote_path: is unused.
dev_timeout : 30
The NETCONF RPC timeout (in seconds)
The NETCONF RPC timeout (in seconds). This argument was added since most of
the time the "package add" RPC takes a significant amount of time. The default
RPC timeout is 30 seconds. So this :timeout: value will be
used in the context of the SW installation process. Defaults to
30 minutes (30*60=1800)
reboot : False
Whether to reboot after installation
@ -971,6 +1017,23 @@ def install_os(path=None, **kwargs):
no_copy : False
If ``True`` the software package will not be SCP’d to the device
bool validate:
When ``True`` this method will perform a config validation against
the new image
bool issu:
When ``True`` allows unified in-service software upgrade
(ISSU) feature enables you to upgrade between two different Junos OS
releases with no disruption on the control plane and with minimal
disruption of traffic.
bool nssu:
When ``True`` allows nonstop software upgrade (NSSU)
enables you to upgrade the software running on a Juniper Networks
EX Series Virtual Chassis or a Juniper Networks EX Series Ethernet
Switch with redundant Routing Engines with a single command and
minimal disruption to network traffic.
CLI Examples:
.. code-block:: bash
@ -979,30 +1042,10 @@ def install_os(path=None, **kwargs):
salt 'device_name' junos.install_os 'salt://junos_16_1.tgz' dev_timeout=300
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
if path is None:
ret['message'] = \
'Please provide the salt path where the junos image is present.'
ret['out'] = False
return ret
image_cached_path = salt.utils.files.mkstemp()
__salt__['cp.get_file'](path, image_cached_path)
if not os.path.isfile(image_cached_path):
ret['message'] = 'Invalid image path.'
ret['out'] = False
return ret
if os.path.getsize(image_cached_path) == 0:
ret['message'] = 'Failed to copy image'
ret['out'] = False
return ret
path = image_cached_path
op = dict()
op = {}
if '__pub_arg' in kwargs:
if kwargs['__pub_arg']:
if isinstance(kwargs['__pub_arg'][-1], dict):
@ -1010,15 +1053,39 @@ def install_os(path=None, **kwargs):
else:
op.update(kwargs)
no_copy_ = op.get('no_copy', False)
if path is None:
ret['message'] = \
'Please provide the salt path where the junos image is present.'
ret['out'] = False
return ret
if not no_copy_:
image_cached_path = salt.utils.files.mkstemp()
__salt__['cp.get_file'](path, image_cached_path)
if not os.path.isfile(image_cached_path):
ret['message'] = 'Invalid image path.'
ret['out'] = False
return ret
if os.path.getsize(image_cached_path) == 0:
ret['message'] = 'Failed to copy image'
ret['out'] = False
return ret
path = image_cached_path
try:
conn.sw.install(path, progress=True)
conn.sw.install(path, progress=True, **op)
ret['message'] = 'Installed the os.'
except Exception as exception:
ret['message'] = 'Installation failed due to: "{0}"'.format(exception)
ret['out'] = False
return ret
finally:
salt.utils.files.safe_rm(image_cached_path)
if not no_copy_:
salt.utils.files.safe_rm(image_cached_path)
if 'reboot' in op and op['reboot'] is True:
try:
@ -1050,7 +1117,7 @@ def file_copy(src=None, dest=None):
salt 'device_name' junos.file_copy /home/m2/info.txt info_copy.txt
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
if src is None:
@ -1098,7 +1165,7 @@ def lock():
salt 'device_name' junos.lock
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
try:
conn.cu.lock()
@ -1121,7 +1188,7 @@ def unlock():
salt 'device_name' junos.unlock
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
try:
conn.cu.unlock()
@ -1188,7 +1255,7 @@ def load(path=None, **kwargs):
salt 'device_name' junos.load 'salt://syslog_template.conf' template_vars='{"syslog_host": "10.180.222.7"}'
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
if path is None:
@ -1197,7 +1264,7 @@ def load(path=None, **kwargs):
ret['out'] = False
return ret
op = dict()
op = {}
if '__pub_arg' in kwargs:
if kwargs['__pub_arg']:
if isinstance(kwargs['__pub_arg'][-1], dict):
@ -1205,7 +1272,7 @@ def load(path=None, **kwargs):
else:
op.update(kwargs)
template_vars = dict()
template_vars = {}
if "template_vars" in op:
template_vars = op["template_vars"]
@ -1252,7 +1319,7 @@ def load(path=None, **kwargs):
except Exception as exception:
ret['message'] = 'Could not load configuration due to : "{0}"'.format(
exception)
ret['format'] = template_format
ret['format'] = op['format']
ret['out'] = False
return ret
finally:
@ -1272,7 +1339,7 @@ def commit_check():
salt 'device_name' junos.commit_check
'''
conn = __proxy__['junos.conn']()
ret = dict()
ret = {}
ret['out'] = True
try:
conn.cu.commit_check()

97
salt/proxy/junos.py
View file

@ -46,10 +46,9 @@ try:
import jnpr.junos.utils
import jnpr.junos.utils.config
import jnpr.junos.utils.sw
from jnpr.junos.exception import RpcTimeoutError
from jnpr.junos.exception import ConnectClosedError
from jnpr.junos.exception import RpcError
from jnpr.junos.exception import ConnectError
from jnpr.junos.exception import RpcTimeoutError, ConnectClosedError,\
RpcError, ConnectError, ProbeError, ConnectAuthError,\
ConnectRefusedError, ConnectTimeoutError
from ncclient.operations.errors import TimeoutExpiredError
except ImportError:
HAS_JUNOS = False
@ -106,9 +105,32 @@ def init(opts):
args[arg] = opts['proxy'][arg]
thisproxy['conn'] = jnpr.junos.Device(**args)
thisproxy['conn'].open()
thisproxy['conn'].bind(cu=jnpr.junos.utils.config.Config)
thisproxy['conn'].bind(sw=jnpr.junos.utils.sw.SW)
try:
thisproxy['conn'].open()
except (ProbeError, ConnectAuthError, ConnectRefusedError, ConnectTimeoutError,
ConnectError) as ex:
log.error("{} : not able to initiate connection to the device".format(str(ex)))
thisproxy['initialized'] = False
return
if 'timeout' in proxy_keys:
timeout = int(opts['proxy']['timeout'])
try:
thisproxy['conn'].timeout = timeout
except Exception as ex:
log.error('Not able to set timeout due to: %s', str(ex))
else:
log.debug('RPC timeout set to %d seconds', timeout)
try:
thisproxy['conn'].bind(cu=jnpr.junos.utils.config.Config)
except Exception as ex:
log.error('Bind failed with Config class due to: {}'.format(str(ex)))
try:
thisproxy['conn'].bind(sw=jnpr.junos.utils.sw.SW)
except Exception as ex:
log.error('Bind failed with SW class due to: {}'.format(str(ex)))
thisproxy['initialized'] = True
@ -129,6 +151,20 @@ def alive(opts):
dev = conn()
thisproxy['conn'].connected = ping()
if not dev.connected:
__salt__['event.fire_master']({}, 'junos/proxy/{}/stop'.format(
opts['proxy']['host']))
return dev.connected
def ping():
'''
Ping? Pong!
'''
dev = conn()
# Check that the underlying netconf connection still exists.
if dev._conn is None:
return False
@ -137,16 +173,35 @@ def alive(opts):
# rpc call is going on.
if hasattr(dev._conn, '_session'):
if dev._conn._session._transport.is_active():
# there is no on going rpc call.
if dev._conn._session._q.empty():
thisproxy['conn'].connected = ping()
# there is no on going rpc call. buffer tell can be 1 as it stores
# remaining char after "]]>]]>" which can be a new line char
if dev._conn._session._buffer.tell() <= 1 and \
dev._conn._session._q.empty():
return _rpc_file_list(dev)
else:
log.debug('skipped ping() call as proxy already getting data')
return True
else:
# ssh connection is lost
dev.connected = False
return False
else:
# other connection modes, like telnet
thisproxy['conn'].connected = ping()
return dev.connected
return _rpc_file_list(dev)
def _rpc_file_list(dev):
try:
dev.rpc.file_list(path='/dev/null', dev_timeout=5)
return True
except (RpcTimeoutError, ConnectClosedError):
try:
dev.close()
return False
except (RpcError, ConnectError, TimeoutExpiredError):
return False
except AttributeError as ex:
if "'NoneType' object has no attribute 'timeout'" in ex:
return False
def proxytype():
@ -170,22 +225,6 @@ def get_serialized_facts():
return facts
def ping():
'''
Ping? Pong!
'''
dev = conn()
try:
dev.rpc.file_list(path='/dev/null', dev_timeout=2)
return True
except (RpcTimeoutError, ConnectClosedError):
try:
dev.close()
except (RpcError, ConnectError, TimeoutExpiredError):
return False
def shutdown(opts):
'''
This is called when the proxy-minion is exiting to make sure the

26
salt/states/junos.py
View file

@ -16,10 +16,22 @@ Refer to :mod:`junos <salt.proxy.junos>` for information on connecting to junos
# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals
import logging
from functools import wraps
log = logging.getLogger()
def resultdecorator(function):
@wraps(function)
def wrapper(*args, **kwargs):
ret = function(*args, **kwargs)
ret['result'] = ret['changes']['out']
return ret
return wrapper
@resultdecorator
def rpc(name, dest=None, format='xml', args=None, **kwargs):
'''
Executes the given rpc. The returned data can be stored in a file
@ -71,6 +83,7 @@ def rpc(name, dest=None, format='xml', args=None, **kwargs):
return ret
@resultdecorator
def set_hostname(name, **kwargs):
'''
Changes the hostname of the device.
@ -104,6 +117,7 @@ def set_hostname(name, **kwargs):
return ret
@resultdecorator
def commit(name, **kwargs):
'''
Commits the changes loaded into the candidate configuration.
@ -147,6 +161,7 @@ def commit(name, **kwargs):
return ret
@resultdecorator
def rollback(name, id, **kwargs):
'''
Rollbacks the committed changes.
@ -181,6 +196,7 @@ def rollback(name, id, **kwargs):
return ret
@resultdecorator
def diff(name, d_id):
'''
Gets the difference between the candidate and the current configuration.
@ -202,6 +218,7 @@ def diff(name, d_id):
return ret
@resultdecorator
def cli(name, **kwargs):
'''
Executes the CLI commands and reuturns the text output.
@ -234,6 +251,7 @@ def cli(name, **kwargs):
return ret
@resultdecorator
def shutdown(name, **kwargs):
'''
Shuts down the device.
@ -260,6 +278,7 @@ def shutdown(name, **kwargs):
return ret
@resultdecorator
def install_config(name, **kwargs):
'''
Loads and commits the configuration provided.
@ -331,6 +350,7 @@ def install_config(name, **kwargs):
return ret
@resultdecorator
def zeroize(name):
'''
Resets the device to default factory settings.
@ -347,6 +367,7 @@ def zeroize(name):
return ret
@resultdecorator
def install_os(name, **kwargs):
'''
Installs the given image on the device. After the installation is complete
@ -382,6 +403,7 @@ def install_os(name, **kwargs):
return ret
@resultdecorator
def file_copy(name, dest=None, **kwargs):
'''
Copies the file from the local device to the junos device.
@ -405,6 +427,7 @@ def file_copy(name, dest=None, **kwargs):
return ret
@resultdecorator
def lock(name):
'''
Attempts an exclusive lock on the candidate configuration. This
@ -426,6 +449,7 @@ def lock(name):
return ret
@resultdecorator
def unlock(name):
'''
Unlocks the candidate configuration.
@ -441,6 +465,7 @@ def unlock(name):
return ret
@resultdecorator
def load(name, **kwargs):
'''
Loads the configuration provided onto the junos device.
@ -502,6 +527,7 @@ def load(name, **kwargs):
return ret
@resultdecorator
def commit_check(name):
'''

250
tests/unit/modules/test_junos.py
View file

@ -7,7 +7,7 @@ from __future__ import absolute_import, print_function, unicode_literals
# Import test libs
from tests.support.mixins import LoaderModuleMockMixin, XMLEqualityMixin
from tests.support.mock import patch, mock_open
from tests.support.mock import patch, mock_open, PropertyMock, call, ANY
from tests.support.unit import skipIf, TestCase
# Import 3rd-party libs
@ -20,8 +20,8 @@ try:
from jnpr.junos.utils.config import Config
from jnpr.junos.utils.sw import SW
from jnpr.junos.device import Device
from jnpr.junos.device import Device
import jxmlease # pylint: disable=unused-import
from jnpr.junos.exception import LockError, UnlockError
HAS_JUNOS = True
except ImportError:
HAS_JUNOS = False
@ -50,7 +50,7 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
def make_connect(self):
with patch('ncclient.manager.connect') as mock_connect:
self.dev = self.dev = Device(
self.dev = Device(
host='1.1.1.1',
user='test',
password='test123',
@ -130,6 +130,18 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
'virtual': True}
return facts
def test_timeout_decorator(self):
with patch('jnpr.junos.Device.timeout',
new_callable=PropertyMock) as mock_timeout:
mock_timeout.return_value = 30
def function(x):
return x
decorator = junos.timeoutDecorator(function)
decorator("Test Mock", dev_timeout=10)
calls = [call(), call(10), call(30)]
mock_timeout.assert_has_calls(calls)
def test_facts_refresh(self):
with patch('salt.modules.saltutil.sync_grains') as mock_sync_grains:
ret = dict()
@ -514,10 +526,9 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
mock_commit_check.return_value = True
args = {'comment': 'Comitted via salt',
'__pub_user': 'root',
'dev_timeout': 40,
'__pub_arg': [2,
{'comment': 'Comitted via salt',
'timeout': 40,
'dev_timeout': 40,
'confirm': 1}],
'confirm': 1,
'__pub_fun': 'junos.rollback',
@ -528,7 +539,7 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
junos.rollback(id=2, **args)
mock_rollback.assert_called_with(2)
mock_commit.assert_called_with(
comment='Comitted via salt', confirm=1, timeout=40)
comment='Comitted via salt', confirm=1, dev_timeout=40)
def test_rollback_with_only_single_arg(self):
with patch('jnpr.junos.utils.config.Config.commit_check') as mock_commit_check, \
@ -1154,25 +1165,6 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
ret)
mock_commit.assert_called_with(comment='comitted via salt', confirm=3)
def test_install_config_commit_check_exception(self):
with patch('jnpr.junos.utils.config.Config.commit_check') as mock_commit_check, \
patch('jnpr.junos.utils.config.Config.diff') as mock_diff, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.safe_rm') as mock_safe_rm, \
patch('salt.utils.files.mkstemp') as mock_mkstemp, \
patch('os.path.isfile') as mock_isfile, \
patch('os.path.getsize') as mock_getsize:
mock_isfile.return_value = True
mock_getsize.return_value = 10
mock_mkstemp.return_value = 'test/path/config'
mock_diff.return_value = 'diff'
mock_commit_check.side_effect = self.raise_exception
ret = dict()
ret['message'] = 'Commit check threw the following exception: "Test exception"'
ret['out'] = False
self.assertEqual(junos.install_config('actual/path/config.xml'), ret)
def test_install_config_commit_check_fails(self):
with patch('jnpr.junos.utils.config.Config.commit_check') as mock_commit_check, \
patch('jnpr.junos.utils.config.Config.diff') as mock_diff, \
@ -1188,7 +1180,7 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
mock_commit_check.return_value = False
ret = dict()
ret['message'] = 'Loaded configuration but commit check failed.'
ret['message'] = 'Loaded configuration but commit check failed, hence rolling back configuration.'
ret['out'] = False
self.assertEqual(junos.install_config('actual/path/config.xml'), ret)
@ -1322,6 +1314,51 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
ret['out'] = False
self.assertEqual(junos.install_os('path', **args), ret)
def test_install_os_no_copy(self):
with patch('jnpr.junos.utils.sw.SW.install') as mock_install, \
patch('salt.utils.files.safe_rm') as mock_safe_rm, \
patch('salt.utils.files.mkstemp') as mock_mkstemp, \
patch('os.path.isfile') as mock_isfile, \
patch('os.path.getsize') as mock_getsize:
mock_getsize.return_value = 10
mock_isfile.return_value = True
ret = dict()
ret['out'] = True
ret['message'] = 'Installed the os.'
self.assertEqual(junos.install_os('path', no_copy=True), ret)
mock_install.assert_called_with(u'path', no_copy=True, progress=True)
mock_mkstemp.assert_not_called()
mock_safe_rm.assert_not_called()
def test_install_os_issu(self):
with patch('jnpr.junos.utils.sw.SW.install') as mock_install, \
patch('salt.utils.files.safe_rm') as mock_safe_rm, \
patch('salt.utils.files.mkstemp') as mock_mkstemp, \
patch('os.path.isfile') as mock_isfile, \
patch('os.path.getsize') as mock_getsize:
mock_getsize.return_value = 10
mock_isfile.return_value = True
ret = dict()
ret['out'] = True
ret['message'] = 'Installed the os.'
self.assertEqual(junos.install_os('path', issu=True), ret)
mock_install.assert_called_with(ANY, issu=True, progress=True)
def test_install_os_add_params(self):
with patch('jnpr.junos.utils.sw.SW.install') as mock_install, \
patch('salt.utils.files.safe_rm') as mock_safe_rm, \
patch('salt.utils.files.mkstemp') as mock_mkstemp, \
patch('os.path.isfile') as mock_isfile, \
patch('os.path.getsize') as mock_getsize:
mock_getsize.return_value = 10
mock_isfile.return_value = True
ret = dict()
ret['out'] = True
ret['message'] = 'Installed the os.'
remote_path = '/path/to/file'
self.assertEqual(junos.install_os('path', remote_path=remote_path, nssu=True, validate=True), ret)
mock_install.assert_called_with(ANY, nssu=True, remote_path=remote_path, progress=True, validate=True)
def test_file_copy_without_args(self):
ret = dict()
ret['message'] = \
@ -1425,7 +1462,6 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
args = mock_execute.call_args
expected_rpc = '<get-interface-information format="json"/>'
self.assertEqualXML(args[0][0], expected_rpc)
self.assertEqual(args[1], {'dev_timeout': 30})
def test_rpc_get_interface_information_with_kwargs(self):
with patch('jnpr.junos.device.Device.execute') as mock_execute:
@ -1496,3 +1532,163 @@ class Test_Junos_Module(TestCase, LoaderModuleMockMixin, XMLEqualityMixin):
junos.rpc('get-chassis-inventory', '/path/to/file')
writes = m_open.write_calls()
assert writes == ['xml rpc reply'], writes
def test_lock_success(self):
ret_exp = {'out': True, 'message': 'Successfully locked the configuration.'}
ret = junos.lock()
self.assertEqual(ret, ret_exp)
def test_lock_error(self):
ret_exp = {'out': False, 'message': 'Could not gain lock due to : "LockError"'}
with patch('jnpr.junos.utils.config.Config.lock') as mock_lock:
mock_lock.side_effect = LockError(None)
ret = junos.lock()
self.assertEqual(ret, ret_exp)
def test_unlock_success(self):
ret_exp = {'out': True, 'message': 'Successfully unlocked the configuration.'}
ret = junos.unlock()
self.assertEqual(ret, ret_exp)
def test_unlock_error(self):
ret_exp = {'out': False, 'message': 'Could not unlock configuration due to : "UnlockError"'}
with patch('jnpr.junos.utils.config.Config.unlock') as mock_unlock:
mock_unlock.side_effect = UnlockError(None)
ret = junos.unlock()
self.assertEqual(ret, ret_exp)
def test_load_none_path(self):
ret_exp = {'out': False,
'message': 'Please provide the salt path where the configuration is present'}
ret = junos.load()
self.assertEqual(ret, ret_exp)
def test_load_wrong_tmp_file(self):
ret_exp = {'out': False, 'message': 'Invalid file path.'}
with patch('salt.utils.files.mkstemp') as mock_mkstemp:
mock_mkstemp.return_value = '/pat/to/tmp/file'
ret = junos.load('/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_invalid_path(self):
ret_exp = {'out': False, 'message': 'Template failed to render'}
ret = junos.load('/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_no_extension(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file')
mock_load.assert_called_with(format='text', path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_xml_extension(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file.xml')
mock_load.assert_called_with(format='xml', path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_set_extension(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file.set')
mock_load.assert_called_with(format='set', path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_replace_true(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file', replace=True)
mock_load.assert_called_with(format='text', merge=False, path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_replace_false(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file', replace=False)
mock_load.assert_called_with(format='text', replace=False, path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_overwrite_true(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file', overwrite=True)
mock_load.assert_called_with(format='text', overwrite=True, path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_overwrite_false(self):
ret_exp = {'out': True, 'message': 'Successfully loaded the configuration.'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
ret = junos.load('/path/to/file', overwrite=False)
mock_load.assert_called_with(format='text', merge=True, path='/path/to/file')
self.assertEqual(ret, ret_exp)
def test_load_error(self):
ret_exp = {'out': False,
'format': 'text',
'message': 'Could not load configuration due to : "Test Error"'}
with patch('os.path.getsize') as mock_getsize, \
patch('jnpr.junos.utils.config.Config.load') as mock_load, \
patch('salt.utils.files.mkstemp') as mock_mkstmp, \
patch('os.path.isfile') as mock_isfile:
mock_getsize.return_value = 1000
mock_mkstmp.return_value = '/path/to/file'
mock_isfile.return_value = True
mock_load.side_effect = Exception('Test Error')
ret = junos.load('/path/to/file')
self.assertEqual(ret, ret_exp)
def test_commit_check_success(self):
ret_exp = {'out': True, 'message': 'Commit check succeeded.'}
ret = junos.commit_check()
self.assertEqual(ret, ret_exp)
def test_commit_check_error(self):
ret_exp = {'out': False, 'message': 'Commit check failed with '}
with patch('jnpr.junos.utils.config.Config.commit_check') as mock_check:
mock_check.side_effect = Exception
ret = junos.commit_check()
self.assertEqual(ret, ret_exp)