mirror of
https://github.com/ansible-collections/community.general.git
synced 2025-10-25 05:23:58 -07:00
8f8a0e1d7c
Some checks are pending
EOL CI / EOL Sanity (Ⓐ2.17) (push) Waiting to run
EOL CI / EOL Units (Ⓐ2.17+py3.10) (push) Waiting to run
EOL CI / EOL Units (Ⓐ2.17+py3.12) (push) Waiting to run
EOL CI / EOL Units (Ⓐ2.17+py3.7) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+alpine319+py:azp/posix/1/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+alpine319+py:azp/posix/2/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+alpine319+py:azp/posix/3/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+fedora39+py:azp/posix/1/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+fedora39+py:azp/posix/2/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+fedora39+py:azp/posix/3/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+ubuntu2004+py:azp/posix/1/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+ubuntu2004+py:azp/posix/2/) (push) Waiting to run
EOL CI / EOL I (Ⓐ2.17+ubuntu2004+py:azp/posix/3/) (push) Waiting to run
nox / Run extra sanity tests (push) Waiting to run
* Adjust all __future__ imports: for i in $(grep -REl "__future__.*absolute_import" plugins/ tests/); do sed -e 's/from __future__ import .*/from __future__ import annotations/g' -i $i; done * Remove all UTF-8 encoding specifications for Python source files: for i in $(grep -REl '[-][*]- coding: utf-8 -[*]-' plugins/ tests/); do sed -e '/^# -\*- coding: utf-8 -\*-/d' -i $i; done * Remove __metaclass__ = type: for i in $(grep -REl '__metaclass__ = type' plugins/ tests/); do sed -e '/^__metaclass__ = type/d' -i $i; done
208 lines
7.6 KiB
Python
208 lines
7.6 KiB
Python
#
|
|
# Copyright (c) 2021, Andreas Botzner <andreas at botzner dot com>
|
|
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import pytest
|
|
import json
|
|
import redis
|
|
from redis import __version__
|
|
|
|
from ansible_collections.community.general.plugins.modules import redis_data_incr
|
|
from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import set_module_args
|
|
|
|
|
|
HAS_REDIS_USERNAME_OPTION = True
|
|
if tuple(map(int, __version__.split('.'))) < (3, 4, 0):
|
|
HAS_REDIS_USERNAME_OPTION = False
|
|
if HAS_REDIS_USERNAME_OPTION:
|
|
from redis.exceptions import NoPermissionError
|
|
|
|
|
|
def test_redis_data_incr_without_arguments(capfd):
|
|
with set_module_args({}):
|
|
with pytest.raises(SystemExit) as results:
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
assert not err
|
|
assert json.loads(out)['failed']
|
|
|
|
|
|
@pytest.mark.skipif(not HAS_REDIS_USERNAME_OPTION, reason="Redis version < 3.4.0")
|
|
def test_redis_data_incr(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_user': 'root',
|
|
'login_password': 'secret',
|
|
'key': 'foo', }):
|
|
mocker.patch('redis.Redis.incr', return_value=57)
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['value'] == 57.0
|
|
assert json.loads(
|
|
out)['msg'] == 'Incremented key: foo to 57'
|
|
assert json.loads(out)['changed']
|
|
|
|
|
|
@pytest.mark.skipif(not HAS_REDIS_USERNAME_OPTION, reason="Redis version < 3.4.0")
|
|
def test_redis_data_incr_int(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_user': 'root',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'increment_int': 10}):
|
|
mocker.patch('redis.Redis.incrby', return_value=57)
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['value'] == 57.0
|
|
assert json.loads(
|
|
out)['msg'] == 'Incremented key: foo by 10 to 57'
|
|
assert json.loads(out)['changed']
|
|
|
|
|
|
@pytest.mark.skipif(not HAS_REDIS_USERNAME_OPTION, reason="Redis version < 3.4.0")
|
|
def test_redis_data_inc_float(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_user': 'root',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'increment_float': '5.5'}):
|
|
mocker.patch('redis.Redis.incrbyfloat', return_value=57.45)
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['value'] == 57.45
|
|
assert json.loads(
|
|
out)['msg'] == 'Incremented key: foo by 5.5 to 57.45'
|
|
assert json.loads(out)['changed']
|
|
|
|
|
|
@pytest.mark.skipif(not HAS_REDIS_USERNAME_OPTION, reason="Redis version < 3.4.0")
|
|
def test_redis_data_incr_float_wrong_value(capfd):
|
|
with set_module_args({
|
|
'login_host': 'localhost',
|
|
'login_user': 'root',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'increment_float': 'not_a_number'
|
|
}):
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['failed']
|
|
|
|
|
|
@pytest.mark.skipif(HAS_REDIS_USERNAME_OPTION, reason="Redis version > 3.4.0")
|
|
def test_redis_data_incr_fail_username(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_user': 'root',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'_ansible_check_mode': False}):
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['failed']
|
|
assert json.loads(
|
|
out)['msg'] == 'The option `username` in only supported with redis >= 3.4.0.'
|
|
|
|
|
|
def test_redis_data_incr_no_username(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_password': 'secret',
|
|
'key': 'foo', }):
|
|
mocker.patch('redis.Redis.incr', return_value=57)
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['value'] == 57.0
|
|
assert json.loads(
|
|
out)['msg'] == 'Incremented key: foo to 57'
|
|
assert json.loads(out)['changed']
|
|
|
|
|
|
def test_redis_data_incr_float_no_username(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'increment_float': '5.5'}):
|
|
mocker.patch('redis.Redis.incrbyfloat', return_value=57.45)
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['value'] == 57.45
|
|
assert json.loads(
|
|
out)['msg'] == 'Incremented key: foo by 5.5 to 57.45'
|
|
assert json.loads(out)['changed']
|
|
|
|
|
|
def test_redis_data_incr_check_mode(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'_ansible_check_mode': True}):
|
|
mocker.patch('redis.Redis.get', return_value=10)
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['value'] == 11.0
|
|
assert json.loads(out)['msg'] == 'Incremented key: foo by 1 to 11.0'
|
|
assert not json.loads(out)['changed']
|
|
|
|
|
|
def test_redis_data_incr_check_mode_not_incrementable(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'_ansible_check_mode': True}):
|
|
mocker.patch('redis.Redis.get', return_value='bar')
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['failed']
|
|
assert json.loads(out)[
|
|
'msg'] == "Value: bar of key: foo is not incrementable(int or float)"
|
|
assert 'value' not in json.loads(out)
|
|
assert not json.loads(out)['changed']
|
|
|
|
|
|
@pytest.mark.skipif(not HAS_REDIS_USERNAME_OPTION, reason="Redis version < 3.4.0")
|
|
def test_redis_data_incr_check_mode_permissions(capfd, mocker):
|
|
with set_module_args({'login_host': 'localhost',
|
|
'login_password': 'secret',
|
|
'key': 'foo',
|
|
'_ansible_check_mode': True}):
|
|
redis.Redis.get = mocker.Mock(side_effect=NoPermissionError(
|
|
"this user has no permissions to run the 'get' command or its subcommand"))
|
|
with pytest.raises(SystemExit):
|
|
redis_data_incr.main()
|
|
out, err = capfd.readouterr()
|
|
print(out)
|
|
assert not err
|
|
assert json.loads(out)['failed']
|
|
assert json.loads(out)['msg'].startswith(
|
|
'Failed to get value of key: foo with exception:')
|
|
assert 'value' not in json.loads(out)
|
|
assert not json.loads(out)['changed']
|