junos scp module (#31950)

* junos_scp module

This module transfers files via SCP from or to remote devices running Junos.

* fix version

* add return documentation

* updated return documentation

* docu, renamed args and exceptions
+ update docu
+ rename arg download to remote_src (simitlar to copy module)
+ exception handling for transfer errors

* add tests

* add test_junos_scp_all

* update to reorganized module utils

* fix unit tests
This commit is contained in:
Christian Giese 2018-01-19 17:07:47 +01:00 committed by ansibot
parent cea681a5c0
commit 2697c0e237
2 changed files with 279 additions and 0 deletions

View file

@ -0,0 +1,186 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2018, Ansible by Red Hat, inc
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'network'}
DOCUMENTATION = """
---
module: junos_scp
version_added: "2.5"
author: "Christian Giese (@GIC-de)"
short_description: Transfer files from or to remote devices running Junos
description:
- This module transfers files via SCP from or to remote devices
running Junos.
extends_documentation_fragment: junos
options:
src:
description:
- The C(src) argument takes a single path, or a list of paths to be
transfered. The argument C(recursive) must be C(true) to transfer
directories.
required: true
default: null
dest:
description:
- The C(dest) argument specifies the path in which to receive the files.
required: false
default: '.'
recursive:
description:
- The C(recursive) argument enables recursive transfer of files and
directories.
required: false
default: false
choices: ['true', 'false']
remote_src:
description:
- The C(remote_src) argument enables the download of files (I(scp get)) from
the remote device. The default behavior is to upload files (I(scp put))
to the remote device.
required: false
default: false
choices: ['true', 'false']
requirements:
- junos-eznc
- ncclient (>=v0.5.2)
notes:
- This module requires the netconf system service be enabled on
the remote device being managed.
- Tested against vMX JUNOS version 17.3R1.10.
"""
EXAMPLES = """
# the required set of connection arguments have been purposely left off
# the examples for brevity
- name: upload local file to home directory on remote device
junos_scp:
src: test.tgz
- name: upload local file to tmp directory on remote device
junos_scp:
src: test.tgz
dest: /tmp/
- name: download file from remote device
junos_scp:
src: test.tgz
remote_src: true
"""
RETURN = """
changed:
description: always true
returned: always
type: bool
"""
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.network.junos.junos import junos_argument_spec, get_param
from ansible.module_utils.pycompat24 import get_exception
try:
from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP
from jnpr.junos.exception import ConnectError
HAS_PYEZ = True
except ImportError:
HAS_PYEZ = False
def connect(module):
host = get_param(module, 'host')
kwargs = {
'port': get_param(module, 'port') or 830,
'user': get_param(module, 'username')
}
if get_param(module, 'password'):
kwargs['passwd'] = get_param(module, 'password')
if get_param(module, 'ssh_keyfile'):
kwargs['ssh_private_key_file'] = get_param(module, 'ssh_keyfile')
kwargs['gather_facts'] = False
try:
device = Device(host, **kwargs)
device.open()
device.timeout = get_param(module, 'timeout') or 10
except ConnectError:
exc = get_exception()
module.fail_json('unable to connect to %s: %s' % (host, str(exc)))
return device
def transfer_files(module, device):
dest = module.params['dest']
recursive = module.params['recursive']
with SCP(device) as scp:
for src in module.params['src']:
if module.params['remote_src']:
scp.get(src.strip(), local_path=dest, recursive=recursive)
else:
scp.put(src.strip(), remote_path=dest, recursive=recursive)
def main():
""" Main entry point for Ansible module execution
"""
argument_spec = dict(
src=dict(type='list', required=True),
dest=dict(type='path', required=False, default="."),
recursive=dict(type='bool', default=False),
remote_src=dict(type='bool', default=False),
transport=dict(default='netconf', choices=['netconf'])
)
argument_spec.update(junos_argument_spec)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
if module.params['provider'] is None:
module.params['provider'] = {}
if not HAS_PYEZ:
module.fail_json(
msg='junos-eznc is required but does not appear to be installed. '
'It can be installed using `pip install junos-eznc`'
)
result = dict(changed=True)
if not module.check_mode:
# open pyez connection and transfer files via SCP
try:
device = connect(module)
transfer_files(module, device)
except Exception as ex:
module.fail_json(
msg=str(ex)
)
finally:
try:
# close pyez connection and ignore exceptions
device.close()
except Exception:
pass
module.exit_json(**result)
if __name__ == '__main__':
main()

View file

@ -0,0 +1,93 @@
# (c) 2018 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.compat.tests.mock import patch, MagicMock
from units.modules.utils import set_module_args
from .junos_module import TestJunosModule
jnpr_mock = MagicMock()
scp_mock = MagicMock()
modules = {
'jnpr': jnpr_mock,
'jnpr.junos': jnpr_mock.junos,
'jnpr.junos.utils': jnpr_mock.junos.utils,
'jnpr.junos.utils.scp': jnpr_mock.junos.utils.scp,
'jnpr.junos.exception': jnpr_mock.junos.execption
}
module_patcher = patch.dict('sys.modules', modules)
module_patcher.start()
jnpr_mock.junos.utils.scp.SCP().__enter__.return_value = scp_mock
from ansible.modules.network.junos import junos_scp
class TestJunosCommandModule(TestJunosModule):
module = junos_scp
def setUp(self):
super(TestJunosCommandModule, self).setUp()
def tearDown(self):
super(TestJunosCommandModule, self).tearDown()
def test_junos_scp_src(self):
set_module_args(dict(src='test.txt'))
result = self.execute_module(changed=True)
args, kwargs = scp_mock.put.call_args
self.assertEqual(args[0], 'test.txt')
self.assertEqual(result['changed'], True)
def test_junos_scp_src_fail(self):
scp_mock.put.side_effect = OSError("[Errno 2] No such file or directory: 'text.txt'")
set_module_args(dict(src='test.txt'))
result = self.execute_module(changed=True, failed=True)
self.assertEqual(result['msg'], "[Errno 2] No such file or directory: 'text.txt'")
def test_junos_scp_remote_src(self):
set_module_args(dict(src='test.txt', remote_src=True))
result = self.execute_module(changed=True)
args, kwargs = scp_mock.get.call_args
self.assertEqual(args[0], 'test.txt')
self.assertEqual(result['changed'], True)
def test_junos_scp_all(self):
set_module_args(dict(src='test', remote_src=True, dest="tmp", recursive=True))
result = self.execute_module(changed=True)
args, kwargs = scp_mock.get.call_args
self.assertEqual(args[0], 'test')
self.assertEqual(kwargs['local_path'], 'tmp')
self.assertEqual(kwargs['recursive'], True)
self.assertEqual(result['changed'], True)
def test_junos_scp_device_param(self):
set_module_args(dict(src='test.txt',
provider={'username': 'unit', 'host': 'test', 'ssh_keyfile': 'path',
'password': 'test', 'port': 234}))
self.execute_module(changed=True)
args, kwargs = jnpr_mock.junos.Device.call_args
self.assertEqual(args[0], 'test')
self.assertEqual(kwargs['passwd'], 'test')
self.assertEqual(kwargs['ssh_private_key_file'], 'path')
self.assertEqual(kwargs['port'], 234)
self.assertEqual(kwargs['user'], 'unit')