1# -*- coding: utf-8 -*- 2# (c) 2017, Toshio Kuratomi <tkuratomi@ansible.com> 3# 4# This file is part of Ansible 5# 6# Ansible is free software: you can redistribute it and/or modify 7# it under the terms of the GNU General Public License as published by 8# the Free Software Foundation, either version 3 of the License, or 9# (at your option) any later version. 10# 11# Ansible is distributed in the hope that it will be useful, 12# but WITHOUT ANY WARRANTY; without even the implied warranty of 13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 14# GNU General Public License for more details. 15# 16# You should have received a copy of the GNU General Public License 17# along with Ansible. If not, see <http://www.gnu.org/licenses/>. 18 19# Make coding more python3-ish 20from __future__ import (absolute_import, division) 21__metaclass__ = type 22 23import sys 24import time 25 26import pytest 27 28from ansible.module_utils.facts import timeout 29 30 31@pytest.fixture 32def set_gather_timeout_higher(): 33 default_timeout = timeout.GATHER_TIMEOUT 34 timeout.GATHER_TIMEOUT = 5 35 yield 36 timeout.GATHER_TIMEOUT = default_timeout 37 38 39@pytest.fixture 40def set_gather_timeout_lower(): 41 default_timeout = timeout.GATHER_TIMEOUT 42 timeout.GATHER_TIMEOUT = 2 43 yield 44 timeout.GATHER_TIMEOUT = default_timeout 45 46 47@timeout.timeout 48def sleep_amount_implicit(amount): 49 # implicit refers to the lack of argument to the decorator 50 time.sleep(amount) 51 return 'Succeeded after {0} sec'.format(amount) 52 53 54@timeout.timeout(timeout.DEFAULT_GATHER_TIMEOUT + 5) 55def sleep_amount_explicit_higher(amount): 56 # explicit refers to the argument to the decorator 57 time.sleep(amount) 58 return 'Succeeded after {0} sec'.format(amount) 59 60 61@timeout.timeout(2) 62def sleep_amount_explicit_lower(amount): 63 # explicit refers to the argument to the decorator 64 time.sleep(amount) 65 return 'Succeeded after {0} sec'.format(amount) 66 67 68# 69# Tests for how the timeout decorator is specified 70# 71 72def test_defaults_still_within_bounds(): 73 # If the default changes outside of these bounds, some of the tests will 74 # no longer test the right thing. Need to review and update the timeouts 75 # in the other tests if this fails 76 assert timeout.DEFAULT_GATHER_TIMEOUT >= 4 77 78 79def test_implicit_file_default_succeeds(): 80 # amount checked must be less than DEFAULT_GATHER_TIMEOUT 81 assert sleep_amount_implicit(1) == 'Succeeded after 1 sec' 82 83 84def test_implicit_file_default_timesout(monkeypatch): 85 monkeypatch.setattr(timeout, 'DEFAULT_GATHER_TIMEOUT', 1) 86 # sleep_time is greater than the default 87 sleep_time = timeout.DEFAULT_GATHER_TIMEOUT + 1 88 with pytest.raises(timeout.TimeoutError): 89 assert sleep_amount_implicit(sleep_time) == '(Not expected to succeed)' 90 91 92def test_implicit_file_overridden_succeeds(set_gather_timeout_higher): 93 # Set sleep_time greater than the default timeout and less than our new timeout 94 sleep_time = 3 95 assert sleep_amount_implicit(sleep_time) == 'Succeeded after {0} sec'.format(sleep_time) 96 97 98def test_implicit_file_overridden_timesout(set_gather_timeout_lower): 99 # Set sleep_time greater than our new timeout but less than the default 100 sleep_time = 3 101 with pytest.raises(timeout.TimeoutError): 102 assert sleep_amount_implicit(sleep_time) == '(Not expected to Succeed)' 103 104 105def test_explicit_succeeds(monkeypatch): 106 monkeypatch.setattr(timeout, 'DEFAULT_GATHER_TIMEOUT', 1) 107 # Set sleep_time greater than the default timeout and less than our new timeout 108 sleep_time = 2 109 assert sleep_amount_explicit_higher(sleep_time) == 'Succeeded after {0} sec'.format(sleep_time) 110 111 112def test_explicit_timeout(): 113 # Set sleep_time greater than our new timeout but less than the default 114 sleep_time = 3 115 with pytest.raises(timeout.TimeoutError): 116 assert sleep_amount_explicit_lower(sleep_time) == '(Not expected to succeed)' 117 118 119# 120# Test that exception handling works 121# 122 123@timeout.timeout(1) 124def function_times_out(): 125 time.sleep(2) 126 127 128# This is just about the same test as function_times_out but uses a separate process which is where 129# we normally have our timeouts. It's more of an integration test than a unit test. 130@timeout.timeout(1) 131def function_times_out_in_run_command(am): 132 am.run_command([sys.executable, '-c', 'import time ; time.sleep(2)']) 133 134 135@timeout.timeout(1) 136def function_other_timeout(): 137 raise TimeoutError('Vanilla Timeout') 138 139 140@timeout.timeout(1) 141def function_raises(): 142 1 / 0 143 144 145@timeout.timeout(1) 146def function_catches_all_exceptions(): 147 try: 148 time.sleep(10) 149 except BaseException: 150 raise RuntimeError('We should not have gotten here') 151 152 153def test_timeout_raises_timeout(): 154 with pytest.raises(timeout.TimeoutError): 155 assert function_times_out() == '(Not expected to succeed)' 156 157 158@pytest.mark.parametrize('stdin', ({},), indirect=['stdin']) 159def test_timeout_raises_timeout_integration_test(am): 160 with pytest.raises(timeout.TimeoutError): 161 assert function_times_out_in_run_command(am) == '(Not expected to succeed)' 162 163 164def test_timeout_raises_other_exception(): 165 with pytest.raises(ZeroDivisionError): 166 assert function_raises() == '(Not expected to succeed)' 167 168 169def test_exception_not_caught_by_called_code(): 170 with pytest.raises(timeout.TimeoutError): 171 assert function_catches_all_exceptions() == '(Not expected to succeed)' 172