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_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()
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
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
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
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)
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
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()
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
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()
"""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