Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:param bool expected_timeout: if Executor raises TimeoutExpired or not
"""
kwargs = {
'command': HTTP_NORMAL_CMD,
'url': 'http://{0}:{1}/badpath'.format(HOST, PORT),
'timeout': 2
} # type: Dict[str, Any]
if accepted_status:
kwargs['status'] = accepted_status
executor = HTTPExecutor(**kwargs)
if not expected_timeout:
executor.start()
executor.stop()
else:
with pytest.raises(TimeoutExpired):
executor.start()
executor.stop()
def test_slow_method_server_timed_out(method):
"""Check if timeout properly expires."""
http_method_slow_cmd = '{python} {srv} {host}:{port} False {method}'.format(
python=sys.executable,
srv=TEST_SERVER_PATH,
host=HOST,
port=PORT,
method=method
)
executor = HTTPExecutor(
http_method_slow_cmd,
'http://{0}:{1}/'.format(HOST, PORT), method=method, timeout=1
)
with pytest.raises(TimeoutExpired) as exc:
executor.start()
assert executor.running() is False
assert 'timed out after' in str(exc.value)
def test_start_and_timeout():
"""Test if executor will properly times out."""
executor = UnixSocketExecutor(
SOCKET_SERVER_CMD + " 10", socket_name=SOCKET_PATH, timeout=5
)
with pytest.raises(TimeoutExpired):
executor.start()
assert executor.running() is False
def test_timeout_error():
"""Check if timeout properly expires."""
executor = PidExecutor(SLEEP, FILENAME, timeout=1)
with pytest.raises(TimeoutExpired):
executor.start()
assert executor.running() is False
def test_it_raises_error_on_timeout():
"""Check if TimeoutExpired gets raised correctly."""
command = 'bash -c "sleep 10 && nc -l 3000"'
executor = TCPExecutor(command, host='localhost', port=3000, timeout=2)
with pytest.raises(TimeoutExpired):
executor.start()
assert executor.running() is False
try:
os.killpg(self.process.pid, sig)
except OSError as err:
if err.errno in IGNORED_ERROR_CODES:
pass
else:
raise
def process_stopped():
"""Return True only only when self.process is not running."""
return self.running() is False
self._set_timeout(timeout)
try:
self.wait_for(process_stopped)
except TimeoutExpired:
log.warning("Timeout expired, killing process", process=self)
pass
self._kill_all_kids(sig)
self._clear_process()
return self