How to use the mirakuru.Executor function in mirakuru

To help you get started, we’ve selected a few mirakuru examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
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()
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
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
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
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)