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_slow_method_server_starting(method):
"""
Test whether or not executor awaits for slow starting servers.
Simple example. You run Gunicorn and it is working but you have to
wait for worker processes.
"""
http_method_slow_cmd = '{python} {srv} {host}:{port} False {method}'.format(
python=sys.executable,
srv=TEST_SERVER_PATH,
host=HOST,
port=PORT,
method=method
)
with HTTPExecutor(
http_method_slow_cmd,
'http://{0}:{1}/'.format(HOST, PORT), method=method, timeout=30
) as executor:
assert executor.running() is True
connect_to_server()
def test_fail_if_other_running():
"""Test raising AlreadyRunning exception when port is blocked."""
executor = HTTPExecutor(
HTTP_NORMAL_CMD, 'http://{0}:{1}/'.format(HOST, PORT),
)
executor2 = HTTPExecutor(
HTTP_NORMAL_CMD, 'http://{0}:{1}/'.format(HOST, PORT),
)
with executor:
assert executor.running() is True
with pytest.raises(AlreadyRunning):
executor2.start()
with pytest.raises(AlreadyRunning) as exc:
with executor2:
pass
assert 'seems to be already running' in str(exc.value)
def test_fail_if_other_running():
"""Test raising AlreadyRunning exception when port is blocked."""
executor = HTTPExecutor(
HTTP_NORMAL_CMD, 'http://{0}:{1}/'.format(HOST, PORT),
)
executor2 = HTTPExecutor(
HTTP_NORMAL_CMD, 'http://{0}:{1}/'.format(HOST, PORT),
)
with executor:
assert executor.running() is True
with pytest.raises(AlreadyRunning):
executor2.start()
with pytest.raises(AlreadyRunning) as exc:
with executor2:
pass
def test_shell_started_server_stops():
"""Test if executor terminates properly executor with shell=True."""
executor = HTTPExecutor(
HTTP_NORMAL_CMD,
'http://{0}:{1}/'.format(HOST, PORT),
timeout=20,
shell=True
)
with pytest.raises(socket.error):
connect_to_server()
with executor:
assert executor.running() is True
connect_to_server()
assert executor.running() is False
with pytest.raises(socket.error):
def test_slow_post_payload_server_starting():
"""
Test whether or not executor awaits for slow starting servers.
Simple example. You run Gunicorn and it is working but you have to
wait for worker processes.
"""
http_method_slow_cmd = '{python} {srv} {host}:{port} False {method}'.format(
python=sys.executable,
srv=TEST_SERVER_PATH,
host=HOST,
port=PORT,
method='Key'
)
with HTTPExecutor(
http_method_slow_cmd,
'http://{0}:{1}/'.format(HOST, PORT),
method='POST',
timeout=30,
payload={'key': 'hole'}
) as executor:
assert executor.running() is True
connect_to_server()
def test_stopping_brutally():
"""
Test if SimpleExecutor is stopping insubordinate process.
Check if the process that doesn't react to SIGTERM signal will be killed
by executor with SIGKILL automatically.
"""
host_port = "127.0.0.1:8000"
cmd = f'{sys.executable} {TEST_SERVER_PATH} {host_port} True'
executor = HTTPExecutor(cmd, f'http://{host_port!s}/', timeout=20)
executor.start()
assert executor.running() is True
stop_at = time.time() + 10
executor.stop()
assert executor.running() is False
assert stop_at <= time.time(), "Subprocess killed earlier than in 10 secs"
def test_default_port():
"""
Test default port for the base TCP check.
Check if HTTP executor fills in the default port for the TCP check
from the base class if no port is provided in the URL.
"""
executor = HTTPExecutor(HTTP_NORMAL_CMD, 'http://{0}/'.format(HOST))
assert executor.url.port is None
assert executor.port == PORT
assert TCPExecutor.pre_start_check(executor) is False
executor.start()
assert TCPExecutor.pre_start_check(executor) is True
executor.stop()
def test_http_status_codes(accepted_status, expected_timeout):
"""
Test how 'status' argument influences executor start.
:param int|str accepted_status: Executor 'status' value
: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()
from mock import patch
from mirakuru import HTTPExecutor, TCPExecutor
from mirakuru import TimeoutExpired, AlreadyRunning
from tests import TEST_SERVER_PATH, HTTP_SERVER_CMD
HOST = "127.0.0.1"
PORT = 7987
HTTP_NORMAL_CMD = '{0} {1}'.format(HTTP_SERVER_CMD, PORT)
HTTP_SLOW_CMD = '{python} {srv} {host}:{port}' \
.format(python=sys.executable, srv=TEST_SERVER_PATH, host=HOST, port=PORT)
slow_server_executor = partial( # pylint: disable=invalid-name
HTTPExecutor,
HTTP_SLOW_CMD,
'http://{0}:{1}/'.format(HOST, PORT),
)
def connect_to_server():
"""Connect to http server and assert 200 response."""
conn = HTTPConnection(HOST, PORT)
conn.request('GET', '/')
assert conn.getresponse().status == OK
conn.close()
def test_executor_starts_and_waits():
"""Test if process awaits for HEAD request to be completed."""
command = 'bash -c "sleep 3 && {0}"'.format(HTTP_NORMAL_CMD)
class ChainConfigType(click.ParamType):
name = "chain-config"
def get_metavar(self, param): # pylint: disable=unused-argument,no-self-use
return ":"
def convert(self, value, param, ctx): # pylint: disable=unused-argument
name, _, rpc_url = value.partition(":")
if name.startswith("http"):
self.fail(f"Invalid value: {value}. Use {self.get_metavar(None)}.")
return name, rpc_url
class HTTPExecutor(mirakuru.HTTPExecutor):
def start(self, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
""" Merged copy paste from the inheritance chain with modified stdout/err behaviour """
if self.pre_start_check():
# Some other executor (or process) is running with same config:
raise AlreadyRunning(self)
if self.process is None:
command = self.command
if not self._shell:
command = self.command_parts
env = os.environ.copy()
env[ENV_UUID] = self._uuid
popen_kwargs = {
"shell": self._shell,
"stdin": subprocess.PIPE,