How to use the mirakuru.base.SimpleExecutor 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_command(command):
    """Check that the command and command parts are equivalent."""

    executor = SimpleExecutor(command)
    assert executor.command == SLEEP_300
    assert executor.command_parts == SLEEP_300.split()
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_context_stopped():
    """Start for context, and shuts it for nested context."""
    executor = SimpleExecutor(SLEEP_300)
    with executor:
        assert executor.running() is True
        with executor.stopped():
            assert executor.running() is False
        assert executor.running() is True

    assert executor.running() is False
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_custom_signal_stop():
    """Start process and shuts it down using signal SIGQUIT."""
    executor = SimpleExecutor(SLEEP_300, sig_stop=signal.SIGQUIT)
    executor.start()
    assert executor.running() is True
    executor.stop()
    assert executor.running() is False
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_executor_methods_returning_self():
    """Test if SimpleExecutor lets to chain start, stop and kill methods."""
    executor = SimpleExecutor(SLEEP_300).start().stop().kill().stop()
    assert not executor.running()

    # Check if context manager returns executor to use it in 'as' phrase:
    with SimpleExecutor(SLEEP_300) as executor:
        assert executor.running()

    with SimpleExecutor(SLEEP_300).start().stopped() as executor:
        assert not executor.running()

    assert SimpleExecutor(SLEEP_300).start().stop().output
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_running_process(command):
    """Start process and shuts it down."""
    executor = SimpleExecutor(command)
    executor.start()
    assert executor.running() is True
    executor.stop()
    assert executor.running() is False

    # check proper __str__ and __repr__ rendering:
    assert 'SimpleExecutor' in repr(executor)
    assert SLEEP_300 in str(executor)
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_running_context():
    """Start process and shuts it down."""
    executor = SimpleExecutor(SLEEP_300)
    with executor:
        assert executor.running() is True

    assert executor.running() is False
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_stopping_not_yet_running_executor():
    """
    Test if SimpleExecutor can be stopped even it was never running.

    We must make sure that it's possible to call .stop() and SimpleExecutor
    will not raise any exception and .start() can be called afterwards.
    """
    executor = SimpleExecutor(SLEEP_300)
    executor.stop()
    executor.start()
    assert executor.running() is True
    executor.stop()
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_executor_methods_returning_self():
    """Test if SimpleExecutor lets to chain start, stop and kill methods."""
    executor = SimpleExecutor(SLEEP_300).start().stop().kill().stop()
    assert not executor.running()

    # Check if context manager returns executor to use it in 'as' phrase:
    with SimpleExecutor(SLEEP_300) as executor:
        assert executor.running()

    with SimpleExecutor(SLEEP_300).start().stopped() as executor:
        assert not executor.running()

    assert SimpleExecutor(SLEEP_300).start().stop().output
github ClearcodeHQ / mirakuru / tests / executors / test_executor.py View on Github external
def test_process_output(command):
    """Start process, check output and shut it down."""
    executor = SimpleExecutor(command)
    executor.start()

    assert executor.output().read() == 'foobar\n'
    executor.stop()
github ClearcodeHQ / mirakuru / src / mirakuru / output.py View on Github external
"""Executor that awaits for appearance of a predefined banner in output."""
import platform
import re
import select
from typing import Union, List, Any, TypeVar, Tuple, IO, Optional

from mirakuru.base import SimpleExecutor


IS_DARWIN = platform.system() == 'Darwin'


OutputExecutorType = TypeVar("OutputExecutorType", bound="OutputExecutor")


class OutputExecutor(SimpleExecutor):
    """Executor that awaits for string output being present in output."""

    def __init__(self,
                 command: Union[str, List[str], Tuple[str, ...]],
                 banner: str,
                 **kwargs: Any) -> None:
        """
        Initialize OutputExecutor executor.

        :param (str, list) command: command to be run by the subprocess
        :param str banner: string that has to appear in process output -
            should compile to regular expression.
        :param bool shell: same as the `subprocess.Popen` shell definition
        :param int timeout: number of seconds to wait for the process to start
            or stop. If None or False, wait indefinitely.
        :param float sleep: how often to check for start/stop condition