ceph
/
ceph-ansible
mirror of https://github.com/ceph/ceph-ansible.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

tests: use ca_test_common with ceph_volume_simple 

To avoid duplicate code.

Signed-off-by: Dimitri Savineau <dsavinea@redhat.com>
pull/6091/head
Dimitri Savineau 2020-11-27 15:21:13 -05:00 committed by Guillaume Abrioux
parent d76fbb38d5
commit c610a1b411
2 changed files with 68 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
tests/library/test_ceph_volume_simple_activate.py
View File

@ -1,9 +1,7 @@
from mock.mock import patch from mock.mock import patch
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
import json
import os import os
import pytest import pytest
import ca_test_common
import ceph_volume_simple_activate import ceph_volume_simple_activate
fake_cluster = 'ceph' fake_cluster = 'ceph'
@ -14,39 +12,18 @@ fake_uuid = '0c4a7eca-0c2a-4c12-beff-08a80f064c52'
fake_path = '/etc/ceph/osd/{}-{}.json'.format(fake_id, fake_uuid) fake_path = '/etc/ceph/osd/{}-{}.json'.format(fake_id, fake_uuid)
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class AnsibleExitJson(Exception):
pass
class AnsibleFailJson(Exception):
pass
def exit_json(*args, **kwargs):
raise AnsibleExitJson(kwargs)
def fail_json(*args, **kwargs):
raise AnsibleFailJson(kwargs)
class TestCephVolumeSimpleActivateModule(object): class TestCephVolumeSimpleActivateModule(object):
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
def test_with_check_mode(self, m_exit_json): def test_with_check_mode(self, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
'osd_id': fake_id, 'osd_id': fake_id,
'osd_fsid': fake_uuid, 'osd_fsid': fake_uuid,
'_ansible_check_mode': True '_ansible_check_mode': True
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]
@ -56,20 +33,20 @@ class TestCephVolumeSimpleActivateModule(object):
assert not result['stdout'] assert not result['stdout']
assert not result['stderr'] assert not result['stderr']
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_with_failure(self, m_run_command, m_exit_json): def test_with_failure(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
'osd_id': fake_id, 'osd_id': fake_id,
'osd_fsid': fake_uuid 'osd_fsid': fake_uuid
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = 'error' stderr = 'error'
rc = 2 rc = 2
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]
@ -78,19 +55,19 @@ class TestCephVolumeSimpleActivateModule(object):
assert result['rc'] == rc assert result['rc'] == rc
assert result['stderr'] == stderr assert result['stderr'] == stderr
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_activate_all_osds(self, m_run_command, m_exit_json): def test_activate_all_osds(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
'osd_all': True 'osd_all': True
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]
@ -101,19 +78,19 @@ class TestCephVolumeSimpleActivateModule(object):
assert result['stdout'] == stdout assert result['stdout'] == stdout
@patch.object(os.path, 'exists', return_value=True) @patch.object(os.path, 'exists', return_value=True)
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_activate_path_exists(self, m_run_command, m_exit_json, m_os_path): def test_activate_path_exists(self, m_run_command, m_exit_json, m_os_path):
set_module_args({ ca_test_common.set_module_args({
'path': fake_path 'path': fake_path
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]
@ -124,35 +101,35 @@ class TestCephVolumeSimpleActivateModule(object):
assert result['stdout'] == stdout assert result['stdout'] == stdout
@patch.object(os.path, 'exists', return_value=False) @patch.object(os.path, 'exists', return_value=False)
@patch.object(basic.AnsibleModule, 'fail_json') @patch('ansible.module_utils.basic.AnsibleModule.fail_json')
def test_activate_path_not_exists(self, m_fail_json, m_os_path): def test_activate_path_not_exists(self, m_fail_json, m_os_path):
set_module_args({ ca_test_common.set_module_args({
'path': fake_path 'path': fake_path
}) })
m_fail_json.side_effect = fail_json m_fail_json.side_effect = ca_test_common.fail_json
with pytest.raises(AnsibleFailJson) as result: with pytest.raises(ca_test_common.AnsibleFailJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]
assert result['msg'] == '{} does not exist'.format(fake_path) assert result['msg'] == '{} does not exist'.format(fake_path)
assert result['rc'] == 1 assert result['rc'] == 1
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_activate_without_systemd(self, m_run_command, m_exit_json): def test_activate_without_systemd(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
'osd_id': fake_id, 'osd_id': fake_id,
'osd_fsid': fake_uuid, 'osd_fsid': fake_uuid,
'systemd': False 'systemd': False
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]
@ -164,20 +141,20 @@ class TestCephVolumeSimpleActivateModule(object):
@patch.dict(os.environ, {'CEPH_CONTAINER_BINARY': fake_container_binary}) @patch.dict(os.environ, {'CEPH_CONTAINER_BINARY': fake_container_binary})
@patch.dict(os.environ, {'CEPH_CONTAINER_IMAGE': fake_container_image}) @patch.dict(os.environ, {'CEPH_CONTAINER_IMAGE': fake_container_image})
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_activate_with_container(self, m_run_command, m_exit_json): def test_activate_with_container(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
'osd_id': fake_id, 'osd_id': fake_id,
'osd_fsid': fake_uuid, 'osd_fsid': fake_uuid,
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_activate.main() ceph_volume_simple_activate.main()
result = result.value.args[0] result = result.value.args[0]

91
tests/library/test_ceph_volume_simple_scan.py
View File

@ -1,9 +1,7 @@
from mock.mock import patch from mock.mock import patch
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
import json
import os import os
import pytest import pytest
import ca_test_common
import ceph_volume_simple_scan import ceph_volume_simple_scan
fake_cluster = 'ceph' fake_cluster = 'ceph'
@ -12,37 +10,16 @@ fake_container_image = 'quay.ceph.io/ceph/daemon:latest'
fake_path = '/var/lib/ceph/osd/ceph-0' fake_path = '/var/lib/ceph/osd/ceph-0'
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class AnsibleExitJson(Exception):
pass
class AnsibleFailJson(Exception):
pass
def exit_json(*args, **kwargs):
raise AnsibleExitJson(kwargs)
def fail_json(*args, **kwargs):
raise AnsibleFailJson(kwargs)
class TestCephVolumeSimpleScanModule(object): class TestCephVolumeSimpleScanModule(object):
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
def test_with_check_mode(self, m_exit_json): def test_with_check_mode(self, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
'_ansible_check_mode': True '_ansible_check_mode': True
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]
@ -52,18 +29,18 @@ class TestCephVolumeSimpleScanModule(object):
assert not result['stdout'] assert not result['stdout']
assert not result['stderr'] assert not result['stderr']
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_with_failure(self, m_run_command, m_exit_json): def test_with_failure(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = 'error' stderr = 'error'
rc = 2 rc = 2
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]
@ -72,18 +49,18 @@ class TestCephVolumeSimpleScanModule(object):
assert result['rc'] == rc assert result['rc'] == rc
assert result['stderr'] == stderr assert result['stderr'] == stderr
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_scan_all_osds(self, m_run_command, m_exit_json): def test_scan_all_osds(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]
@ -94,19 +71,19 @@ class TestCephVolumeSimpleScanModule(object):
assert result['stdout'] == stdout assert result['stdout'] == stdout
@patch.object(os.path, 'exists', return_value=True) @patch.object(os.path, 'exists', return_value=True)
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_scan_path_exists(self, m_run_command, m_exit_json, m_os_path): def test_scan_path_exists(self, m_run_command, m_exit_json, m_os_path):
set_module_args({ ca_test_common.set_module_args({
'path': fake_path 'path': fake_path
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]
@ -117,14 +94,14 @@ class TestCephVolumeSimpleScanModule(object):
assert result['stdout'] == stdout assert result['stdout'] == stdout
@patch.object(os.path, 'exists', return_value=False) @patch.object(os.path, 'exists', return_value=False)
@patch.object(basic.AnsibleModule, 'fail_json') @patch('ansible.module_utils.basic.AnsibleModule.fail_json')
def test_scan_path_not_exists(self, m_fail_json, m_os_path): def test_scan_path_not_exists(self, m_fail_json, m_os_path):
set_module_args({ ca_test_common.set_module_args({
'path': fake_path 'path': fake_path
}) })
m_fail_json.side_effect = fail_json m_fail_json.side_effect = ca_test_common.fail_json
with pytest.raises(AnsibleFailJson) as result: with pytest.raises(ca_test_common.AnsibleFailJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]
@ -132,21 +109,21 @@ class TestCephVolumeSimpleScanModule(object):
assert result['rc'] == 1 assert result['rc'] == 1
@patch.object(os.path, 'exists', return_value=True) @patch.object(os.path, 'exists', return_value=True)
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_scan_path_stdout_force(self, m_run_command, m_exit_json, m_os_path): def test_scan_path_stdout_force(self, m_run_command, m_exit_json, m_os_path):
set_module_args({ ca_test_common.set_module_args({
'path': fake_path, 'path': fake_path,
'force': True, 'force': True,
'stdout': True 'stdout': True
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]
@ -158,18 +135,18 @@ class TestCephVolumeSimpleScanModule(object):
@patch.dict(os.environ, {'CEPH_CONTAINER_BINARY': fake_container_binary}) @patch.dict(os.environ, {'CEPH_CONTAINER_BINARY': fake_container_binary})
@patch.dict(os.environ, {'CEPH_CONTAINER_IMAGE': fake_container_image}) @patch.dict(os.environ, {'CEPH_CONTAINER_IMAGE': fake_container_image})
@patch.object(basic.AnsibleModule, 'exit_json') @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
@patch.object(basic.AnsibleModule, 'run_command') @patch('ansible.module_utils.basic.AnsibleModule.run_command')
def test_scan_with_container(self, m_run_command, m_exit_json): def test_scan_with_container(self, m_run_command, m_exit_json):
set_module_args({ ca_test_common.set_module_args({
}) })
m_exit_json.side_effect = exit_json m_exit_json.side_effect = ca_test_common.exit_json
stdout = '' stdout = ''
stderr = '' stderr = ''
rc = 0 rc = 0
m_run_command.return_value = rc, stdout, stderr m_run_command.return_value = rc, stdout, stderr
with pytest.raises(AnsibleExitJson) as result: with pytest.raises(ca_test_common.AnsibleExitJson) as result:
ceph_volume_simple_scan.main() ceph_volume_simple_scan.main()
result = result.value.args[0] result = result.value.args[0]