diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 16:04:21 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 16:04:21 +0000 |
commit | 8a754e0858d922e955e71b253c139e071ecec432 (patch) | |
tree | 527d16e74bfd1840c85efd675fdecad056c54107 /test/units/module_utils/facts/test_timeout.py | |
parent | Initial commit. (diff) | |
download | ansible-core-8a754e0858d922e955e71b253c139e071ecec432.tar.xz ansible-core-8a754e0858d922e955e71b253c139e071ecec432.zip |
Adding upstream version 2.14.3.upstream/2.14.3upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'test/units/module_utils/facts/test_timeout.py')
-rw-r--r-- | test/units/module_utils/facts/test_timeout.py | 171 |
1 files changed, 171 insertions, 0 deletions
diff --git a/test/units/module_utils/facts/test_timeout.py b/test/units/module_utils/facts/test_timeout.py new file mode 100644 index 0000000..2adbc4a --- /dev/null +++ b/test/units/module_utils/facts/test_timeout.py @@ -0,0 +1,171 @@ +# -*- coding: utf-8 -*- +# (c) 2017, Toshio Kuratomi <tkuratomi@ansible.com> +# +# 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 + +import sys +import time + +import pytest + +from ansible.module_utils.facts import timeout + + +@pytest.fixture +def set_gather_timeout_higher(): + default_timeout = timeout.GATHER_TIMEOUT + timeout.GATHER_TIMEOUT = 5 + yield + timeout.GATHER_TIMEOUT = default_timeout + + +@pytest.fixture +def set_gather_timeout_lower(): + default_timeout = timeout.GATHER_TIMEOUT + timeout.GATHER_TIMEOUT = 2 + yield + timeout.GATHER_TIMEOUT = default_timeout + + +@timeout.timeout +def sleep_amount_implicit(amount): + # implicit refers to the lack of argument to the decorator + time.sleep(amount) + return 'Succeeded after {0} sec'.format(amount) + + +@timeout.timeout(timeout.DEFAULT_GATHER_TIMEOUT + 5) +def sleep_amount_explicit_higher(amount): + # explicit refers to the argument to the decorator + time.sleep(amount) + return 'Succeeded after {0} sec'.format(amount) + + +@timeout.timeout(2) +def sleep_amount_explicit_lower(amount): + # explicit refers to the argument to the decorator + time.sleep(amount) + return 'Succeeded after {0} sec'.format(amount) + + +# +# Tests for how the timeout decorator is specified +# + +def test_defaults_still_within_bounds(): + # If the default changes outside of these bounds, some of the tests will + # no longer test the right thing. Need to review and update the timeouts + # in the other tests if this fails + assert timeout.DEFAULT_GATHER_TIMEOUT >= 4 + + +def test_implicit_file_default_succeeds(): + # amount checked must be less than DEFAULT_GATHER_TIMEOUT + assert sleep_amount_implicit(1) == 'Succeeded after 1 sec' + + +def test_implicit_file_default_timesout(monkeypatch): + monkeypatch.setattr(timeout, 'DEFAULT_GATHER_TIMEOUT', 1) + # sleep_time is greater than the default + sleep_time = timeout.DEFAULT_GATHER_TIMEOUT + 1 + with pytest.raises(timeout.TimeoutError): + assert sleep_amount_implicit(sleep_time) == '(Not expected to succeed)' + + +def test_implicit_file_overridden_succeeds(set_gather_timeout_higher): + # Set sleep_time greater than the default timeout and less than our new timeout + sleep_time = 3 + assert sleep_amount_implicit(sleep_time) == 'Succeeded after {0} sec'.format(sleep_time) + + +def test_implicit_file_overridden_timesout(set_gather_timeout_lower): + # Set sleep_time greater than our new timeout but less than the default + sleep_time = 3 + with pytest.raises(timeout.TimeoutError): + assert sleep_amount_implicit(sleep_time) == '(Not expected to Succeed)' + + +def test_explicit_succeeds(monkeypatch): + monkeypatch.setattr(timeout, 'DEFAULT_GATHER_TIMEOUT', 1) + # Set sleep_time greater than the default timeout and less than our new timeout + sleep_time = 2 + assert sleep_amount_explicit_higher(sleep_time) == 'Succeeded after {0} sec'.format(sleep_time) + + +def test_explicit_timeout(): + # Set sleep_time greater than our new timeout but less than the default + sleep_time = 3 + with pytest.raises(timeout.TimeoutError): + assert sleep_amount_explicit_lower(sleep_time) == '(Not expected to succeed)' + + +# +# Test that exception handling works +# + +@timeout.timeout(1) +def function_times_out(): + time.sleep(2) + + +# This is just about the same test as function_times_out but uses a separate process which is where +# we normally have our timeouts. It's more of an integration test than a unit test. +@timeout.timeout(1) +def function_times_out_in_run_command(am): + am.run_command([sys.executable, '-c', 'import time ; time.sleep(2)']) + + +@timeout.timeout(1) +def function_other_timeout(): + raise TimeoutError('Vanilla Timeout') + + +@timeout.timeout(1) +def function_raises(): + 1 / 0 + + +@timeout.timeout(1) +def function_catches_all_exceptions(): + try: + time.sleep(10) + except BaseException: + raise RuntimeError('We should not have gotten here') + + +def test_timeout_raises_timeout(): + with pytest.raises(timeout.TimeoutError): + assert function_times_out() == '(Not expected to succeed)' + + +@pytest.mark.parametrize('stdin', ({},), indirect=['stdin']) +def test_timeout_raises_timeout_integration_test(am): + with pytest.raises(timeout.TimeoutError): + assert function_times_out_in_run_command(am) == '(Not expected to succeed)' + + +def test_timeout_raises_other_exception(): + with pytest.raises(ZeroDivisionError): + assert function_raises() == '(Not expected to succeed)' + + +def test_exception_not_caught_by_called_code(): + with pytest.raises(timeout.TimeoutError): + assert function_catches_all_exceptions() == '(Not expected to succeed)' |