Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_start_check_executor():
"""Validate Executor base class having NotImplemented methods."""
executor = Executor(SLEEP_300)
with pytest.raises(NotImplementedError):
executor.pre_start_check()
with pytest.raises(NotImplementedError):
executor.after_start_check()
def test_executor_ignores_processes_exiting_with_0():
"""
Test process exit detection.
Subprocess exiting with zero should be tolerated in order to support
double-forking applications.
"""
# We execute a process that will return zero. In order to give the process
# enough time to return we keep the polling loop spinning for a second.
executor = Executor(['bash', '-c', 'exit 0'], timeout=1.0)
executor.pre_start_check = mock.Mock(return_value=False) # type: ignore
executor.after_start_check = mock.Mock(return_value=False) # type: ignore
with pytest.raises(TimeoutExpired):
# We keep the post-checks spinning forever so it eventually times out.
executor.start()
# Both checks should have been called.
assert executor.pre_start_check.called is True # type: ignore
assert executor.after_start_check.called is True # type: ignore
def test_executor_raises_if_process_exits_with_error():
"""
Test process exit detection.
If the process exits with an error while checks are being polled, executor
should raise an exception.
"""
error_code = 12
failing_executor = Executor(
['bash', '-c', f'exit {error_code!s}'],
timeout=5
)
failing_executor.pre_start_check = mock.Mock( # type: ignore
return_value=False)
# After-start check will keep returning False to let the process terminate.
failing_executor.after_start_check = mock.Mock( # type: ignore
return_value=False)
with pytest.raises(ProcessExitedWithError) as exc:
failing_executor.start()
assert exc.value.exit_code == 12
error_msg = f'exited with a non-zero code: {error_code!s}'
assert error_msg in str(exc.value)