ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Proxmox tasks module (#3226) (#3508)

Browse source 
* Started creating Unit tests for Proxmox Tasks module

* Tried really hard to get the mock to work

* unit tests  for the module

* Fixed symslink and permissions

* Suggested changes and more unit tests

* Fixed isFalse

* Apply suggestions from code review

* Update plugins/modules/cloud/misc/proxmox_tasks_info.py

Co-authored-by: Felix Fontein <felix@fontein.de>
(cherry picked from commit e95f63b067)

Co-authored-by: Andreas Botzner <paginabianca66@gmail.com>
This commit is contained in:
patchback[bot] 2021-10-03 22:37:41 +02:00 committed by GitHub
parent d77e256088
commit 773df88a41
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 397 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

186
plugins/modules/cloud/misc/proxmox_tasks_info.py Normal file
View file

@ -0,0 +1,186 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2021, Andreas Botzner (@paginabianca) <andreas at botzner dot com>
# 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
DOCUMENTATION = r'''
---
module: proxmox_tasks_info
short_description: Retrieve information about one or more Proxmox VE tasks
version_added: 3.8.0
description:
- Retrieve information about one or more Proxmox VE tasks.
author: 'Andreas Botzner (@paginabianca) <andreas at botzner dot com>'
options:
node:
description:
- Node where to get tasks.
required: true
type: str
task:
description:
- Return specific task.
aliases: ['upid', 'name']
type: str
extends_documentation_fragment:
- community.general.proxmox.documentation
'''
EXAMPLES = '''
- name: List tasks on node01
community.general.proxmox_task_info:
api_host: proxmoxhost
api_user: root@pam
api_password: '{{ password | default(omit) }}'
api_token_id: '{{ token_id | default(omit) }}'
api_token_secret: '{{ token_secret | default(omit) }}'
node: node01
register: result
- name: Retrieve information about specific tasks on node01
community.general.proxmox_task_info:
api_host: proxmoxhost
api_user: root@pam
api_password: '{{ password | default(omit) }}'
api_token_id: '{{ token_id | default(omit) }}'
api_token_secret: '{{ token_secret | default(omit) }}'
task: 'UPID:node01:00003263:16167ACE:621EE230:srvreload:networking:root@pam:'
node: node01
register: proxmox_tasks
'''
RETURN = '''
proxmox_tasks:
description: List of tasks.
returned: on success
type: list
elements: dict
contains:
id:
description: ID of the task.
returned: on success
type: str
node:
description: Node name.
returned: on success
type: str
pid:
description: PID of the task.
returned: on success
type: int
pstart:
description: pastart of the task.
returned: on success
type: int
starttime:
description: Starting time of the task.
returned: on success
type: int
type:
description: Type of the task.
returned: on success
type: str
upid:
description: UPID of the task.
returned: on success
type: str
user:
description: User that owns the task.
returned: on success
type: str
endtime:
description: Endtime of the task.
returned: on success, can be absent
type: int
status:
description: Status of the task.
returned: on success, can be absent
type: str
failed:
description: If the task failed.
returned: when status is defined
type: bool
msg:
description: Short message.
returned: on failure
type: str
sample: 'Task: UPID:xyz:xyz does not exist on node: proxmoxnode'
'''
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
from ansible_collections.community.general.plugins.module_utils.proxmox import (
proxmox_auth_argument_spec, ProxmoxAnsible, HAS_PROXMOXER, PROXMOXER_IMP_ERR)
class ProxmoxTaskInfoAnsible(ProxmoxAnsible):
def get_task(self, upid, node):
tasks = self.get_tasks(node)
for task in tasks:
if task.info['upid'] == upid:
return [task]
def get_tasks(self, node):
tasks = self.proxmox_api.nodes(node).tasks.get()
return [ProxmoxTask(task) for task in tasks]
class ProxmoxTask:
def __init__(self, task):
self.info = dict()
for k, v in task.items():
if k == 'status' and isinstance(v, str):
self.info[k] = v
if v != 'OK':
self.info['failed'] = True
else:
self.info[k] = v
def proxmox_task_info_argument_spec():
return dict(
task=dict(type='str', aliases=['upid', 'name'], required=False),
node=dict(type='str', required=True),
)
def main():
module_args = proxmox_auth_argument_spec()
task_info_args = proxmox_task_info_argument_spec()
module_args.update(task_info_args)
module = AnsibleModule(
argument_spec=module_args,
required_together=[('api_token_id', 'api_token_secret'),
('api_user', 'api_password')],
required_one_of=[('api_password', 'api_token_id')],
supports_check_mode=True)
result = dict(changed=False)
if not HAS_PROXMOXER:
module.fail_json(msg=missing_required_lib(
'proxmoxer'), exception=PROXMOXER_IMP_ERR)
proxmox = ProxmoxTaskInfoAnsible(module)
upid = module.params['task']
node = module.params['node']
if upid:
tasks = proxmox.get_task(upid=upid, node=node)
else:
tasks = proxmox.get_tasks(node=node)
if tasks is not None:
result['proxmox_tasks'] = [task.info for task in tasks]
module.exit_json(**result)
else:
result['msg'] = 'Task: {0} does not exist on node: {1}.'.format(
upid, node)
module.fail_json(**result)
if __name__ == '__main__':
main()

1
plugins/modules/proxmox_tasks_info.py Symbolic link
View file

@ -0,0 +1 @@
cloud/misc/proxmox_tasks_info.py

210
tests/unit/plugins/modules/cloud/misc/test_proxmox_tasks_info.py Normal file
View file

@ -0,0 +1,210 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2021, Andreas Botzner (@paginabianca) <andreas at botzner dot com>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
#
# Proxmox Tasks module unit tests.
# The API responses used in these tests were recorded from PVE version 6.4-8
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import pytest
import json
from ansible_collections.community.general.plugins.modules.cloud.misc import proxmox_tasks_info
from ansible_collections.community.general.plugins.module_utils.proxmox import ProxmoxAnsible
from ansible_collections.community.general.tests.unit.compat.mock import MagicMock, patch
from ansible_collections.community.general.tests.unit.plugins.modules.utils import (
AnsibleExitJson, AnsibleFailJson, ModuleTestCase, set_module_args
)
from ansible_collections.community.general.tests.unit.plugins.modules.utils import set_module_args
from ansible_collections.community.general.plugins.module_utils import proxmox
NODE = 'node01'
TASK_UPID = 'UPID:iaclab-01-01:000029DD:1599528B:6108F068:srvreload:networking:root@pam:'
TASKS = [
{
"endtime": 1629092710,
"id": "networking",
"node": "iaclab-01-01",
"pid": 3539,
"pstart": 474062216,
"starttime": 1629092709,
"status": "OK",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:00000DD3:1C419D88:6119FB65:srvreload:networking:root@pam:",
"user": "root@pam"
},
{
"endtime": 1627975785,
"id": "networking",
"node": "iaclab-01-01",
"pid": 10717,
"pstart": 362369675,
"starttime": 1627975784,
"status": "command 'ifreload -a' failed: exit code 1",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:000029DD:1599528B:6108F068:srvreload:networking:root@pam:",
"user": "root@pam"
},
{
"endtime": 1627975503,
"id": "networking",
"node": "iaclab-01-01",
"pid": 6778,
"pstart": 362341540,
"starttime": 1627975503,
"status": "OK",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:00001A7A:1598E4A4:6108EF4F:srvreload:networking:root@pam:",
"user": "root@pam"
}
]
EXPECTED_TASKS = [
{
"endtime": 1629092710,
"id": "networking",
"node": "iaclab-01-01",
"pid": 3539,
"pstart": 474062216,
"starttime": 1629092709,
"status": "OK",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:00000DD3:1C419D88:6119FB65:srvreload:networking:root@pam:",
"user": "root@pam",
"failed": False
},
{
"endtime": 1627975785,
"id": "networking",
"node": "iaclab-01-01",
"pid": 10717,
"pstart": 362369675,
"starttime": 1627975784,
"status": "command 'ifreload -a' failed: exit code 1",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:000029DD:1599528B:6108F068:srvreload:networking:root@pam:",
"user": "root@pam",
"failed": True
},
{
"endtime": 1627975503,
"id": "networking",
"node": "iaclab-01-01",
"pid": 6778,
"pstart": 362341540,
"starttime": 1627975503,
"status": "OK",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:00001A7A:1598E4A4:6108EF4F:srvreload:networking:root@pam:",
"user": "root@pam",
"failed": False
}
]
EXPECTED_SINGLE_TASK = [
{
"endtime": 1627975785,
"id": "networking",
"node": "iaclab-01-01",
"pid": 10717,
"pstart": 362369675,
"starttime": 1627975784,
"status": "command 'ifreload -a' failed: exit code 1",
"type": "srvreload",
"upid": "UPID:iaclab-01-01:000029DD:1599528B:6108F068:srvreload:networking:root@pam:",
"user": "root@pam",
"failed": True
},
]
@patch('ansible_collections.community.general.plugins.module_utils.proxmox.ProxmoxAnsible._connect')
def test_without_required_parameters(connect_mock, capfd, mocker):
set_module_args({})
with pytest.raises(SystemExit):
proxmox_tasks_info.main()
out, err = capfd.readouterr()
assert not err
assert json.loads(out)['failed']
@patch('ansible_collections.community.general.plugins.module_utils.proxmox.ProxmoxAnsible._connect')
def test_get_tasks(connect_mock, capfd, mocker):
set_module_args({'api_host': 'proxmoxhost',
'api_user': 'root@pam',
'api_password': 'supersecret',
'node': NODE})
def f():
m = mocker.MagicMock()
g = mocker.MagicMock()
m.nodes = mocker.MagicMock(return_value=g)
g.tasks.get = mocker.MagicMock(return_value=TASKS)
return m
connect_mock.side_effect = f
proxmox_tasks_info.HAS_PROXMOXER = True
with pytest.raises(SystemExit):
proxmox_tasks_info.main()
out, err = capfd.readouterr()
assert not err
assert len(json.loads(out)['proxmox_tasks']) != 0
assert not json.loads(out)['changed']
@patch('ansible_collections.community.general.plugins.module_utils.proxmox.ProxmoxAnsible._connect')
def test_get_single_task(connect_mock, capfd, mocker):
set_module_args({'api_host': 'proxmoxhost',
'api_user': 'root@pam',
'api_password': 'supersecret',
'node': NODE,
'task': TASK_UPID})
def f():
m = mocker.MagicMock()
g = mocker.MagicMock()
m.nodes = mocker.MagicMock(return_value=g)
g.tasks.get = mocker.MagicMock(return_value=TASKS)
return m
connect_mock.side_effect = f
proxmox_tasks_info.HAS_PROXMOXER = True
with pytest.raises(SystemExit):
proxmox_tasks_info.main()
out, err = capfd.readouterr()
assert not err
assert len(json.loads(out)['proxmox_tasks']) == 1
assert json.loads(out)
assert not json.loads(out)['changed']
@patch('ansible_collections.community.general.plugins.module_utils.proxmox.ProxmoxAnsible._connect')
def test_get_non_existent_task(connect_mock, capfd, mocker):
set_module_args({'api_host': 'proxmoxhost',
'api_user': 'root@pam',
'api_password': 'supersecret',
'node': NODE,
'task': 'UPID:nonexistent'})
def f():
m = mocker.MagicMock()
g = mocker.MagicMock()
m.nodes = mocker.MagicMock(return_value=g)
g.tasks.get = mocker.MagicMock(return_value=TASKS)
return m
connect_mock.side_effect = f
proxmox_tasks_info.HAS_PROXMOXER = True
with pytest.raises(SystemExit):
proxmox_tasks_info.main()
out, err = capfd.readouterr()
assert not err
assert json.loads(out)['failed']
assert 'proxmox_tasks' not in json.loads(out)
assert not json.loads(out)['changed']
assert json.loads(
out)['msg'] == 'Task: UPID:nonexistent does not exist on node: node01.'