Skip to content
Snippets Groups Projects
Commit 0fe88338 authored by Marcel Huber's avatar Marcel Huber
Browse files

try fixing stuck builds

parent 4f55896a
No related branches found
No related tags found
No related merge requests found
Pipeline #260954 canceled
Stage: test
Hide whitespace changes
Inline Side-by-side
test/wait-for-it.py
+ 78
− 100
View file @ 0fe88338
import unittest
import shlex
from subprocess import Popen, PIPE
import os
import sys
import socket
import re
has_timeout_param = True
try:
from subprocess32 import Popen, PIPE, STDOUT, TimeoutExpired
except:
try:
from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
except:
from subprocess import Popen, PIPE, STDOUT
class TimeoutExpired(Exception):
def __init__(self):
pass
has_timeout_param = False
MISSING_ARGS_TEXT = "Error: you need to provide a host and port to test."
HELP_TEXT = "Usage:" # Start of help text
DIVIDE_LINE = '-'*71 # Output line of dashes
DIVIDE_LINE = '-' * 71 # Output line of dashes
class TestWaitForIt(unittest.TestCase):
"""
TestWaitForIt tests the wait-for-it.sh shell script.
The wait-for-it.sh script is assumed to be in the parent directory to
the test script.
"""
"""TestWaitForIt tests the wait-for-it.sh shell script.
def execute(self, cmd):
"""Executes a command and returns exit code, STDOUT, STDERR"""
The wait-for-it.sh script is assumed to be in the parent directory
to the test script.
"""
def execute(self, cmd, stdincontent=None, timeout=10):
_kwargs = {'input': stdincontent}
if has_timeout_param:
_kwargs['timeout'] = timeout
args = shlex.split(cmd)
proc = Popen(args, stdout=PIPE, stderr=PIPE)
out, err = proc.communicate()
exitcode = proc.returncode
return exitcode, out, err
try:
_stdout, _stderr = proc.communicate(**_kwargs)
except TimeoutExpired:
_kwargs = {'input': None}
if has_timeout_param:
_kwargs['timeout'] = 5
proc.kill()
_stdout, _stderr = proc.communicate(**_kwargs)
except OSError:
pass
finally:
returncode = proc.poll()
return (returncode, _stdout, _stderr)
def open_local_port(self, host="localhost", port=8929, timeout=5):
s = socket.socket()
......@@ -37,17 +62,15 @@ class TestWaitForIt(unittest.TestCase):
actual_exitcode, out, err = self.execute(command)
# Check stderr
msg = ("Failed check that STDERR:\n" +
DIVIDE_LINE + "\n" + err + "\n" + DIVIDE_LINE +
"\nmatches:\n" +
DIVIDE_LINE + "\n" + stderr_regex + "\n" + DIVIDE_LINE)
msg = ("Failed check that STDERR:\n" + DIVIDE_LINE + "\n" + err +
"\n" + DIVIDE_LINE + "\nmatches:\n" + DIVIDE_LINE + "\n" +
stderr_regex + "\n" + DIVIDE_LINE)
self.assertIsNotNone(re.match(stderr_regex, err, re.DOTALL), msg)
# Check STDOUT
msg = ("Failed check that STDOUT:\n" +
DIVIDE_LINE + "\n" + out + "\n" + DIVIDE_LINE +
"\nmatches:\n" +
DIVIDE_LINE + "\n" + stdout_regex + "\n" + DIVIDE_LINE)
msg = ("Failed check that STDOUT:\n" + DIVIDE_LINE + "\n" + out +
"\n" + DIVIDE_LINE + "\nmatches:\n" + DIVIDE_LINE + "\n" +
stdout_regex + "\n" + DIVIDE_LINE)
self.assertIsNotNone(re.match(stdout_regex, out, re.DOTALL), msg)
# Check exit code
......@@ -59,121 +82,76 @@ class TestWaitForIt(unittest.TestCase):
self.wait_script = os.path.join(parent_path, "wait-for-it.sh")
def test_no_args(self):
"""
Check that no aruments returns the missing args text and the
correct return code
"""
self.check_args(
"",
"^$",
MISSING_ARGS_TEXT,
1
)
"""Check that no aruments returns the missing args text and the correct
return code."""
self.check_args("", "^$", MISSING_ARGS_TEXT, 1)
# Return code should be 1 when called with no args
exitcode, out, err = self.execute(self.wait_script)
self.assertEqual(exitcode, 1)
self.assertEqual(exitcode, 1)
def test_help(self):
""" Check that help text is printed with --help argument """
self.check_args(
"--help",
"",
HELP_TEXT,
1
)
"""Check that help text is printed with --help argument."""
self.check_args("--help", "", HELP_TEXT, 1)
def test_no_port(self):
""" Check with missing port argument """
self.check_args(
"--host=localhost",
"",
MISSING_ARGS_TEXT,
1
)
"""Check with missing port argument."""
self.check_args("--host=localhost", "", MISSING_ARGS_TEXT, 1)
def test_no_host(self):
""" Check with missing hostname argument """
self.check_args(
"--port=80",
"",
MISSING_ARGS_TEXT,
1
)
"""Check with missing hostname argument."""
self.check_args("--port=80", "", MISSING_ARGS_TEXT, 1)
def test_host_port(self):
""" Check that --host and --port args work correctly """
"""Check that --host and --port args work correctly."""
soc = self.open_local_port(port=8929)
self.check_args(
"--host=localhost --port=8929 --timeout=1",
"",
"wait-for-it.sh: waiting 1 seconds for localhost:8929",
0
)
"--host=localhost --port=8929 --timeout=1", "",
"wait-for-it.sh: waiting 1 seconds for localhost:8929", 0)
soc.close()
def test_combined_host_port(self):
"""
Tests that wait-for-it.sh returns correctly after establishing a
connectionm using combined host and ports
"""
"""Tests that wait-for-it.sh returns correctly after establishing a
connectionm using combined host and ports."""
soc = self.open_local_port(port=8929)
self.check_args(
"localhost:8929 --timeout=1",
"",
"wait-for-it.sh: waiting 1 seconds for localhost:8929",
0
)
"localhost:8929 --timeout=1", "",
"wait-for-it.sh: waiting 1 seconds for localhost:8929", 0)
soc.close()
def test_port_failure_with_timeout(self):
"""
Note exit status of 124 is exected, passed from the timeout command
"""
"""Note exit status of 124 is exected, passed from the timeout
command."""
self.check_args(
"localhost:8929 --timeout=1",
"",
"localhost:8929 --timeout=1", "",
".*timeout occurred after waiting 1 seconds for localhost:8929",
124
)
124)
def test_command_execution(self):
"""
Checks that a command executes correctly after a port test passes
"""
"""Checks that a command executes correctly after a port test
passes."""
soc = self.open_local_port(port=8929)
self.check_args(
"localhost:8929 -- echo \"CMD OUTPUT\"",
"CMD OUTPUT",
".*wait-for-it.sh: localhost:8929 is available after 0 seconds",
0
)
"localhost:8929 -- echo \"CMD OUTPUT\"", "CMD OUTPUT",
".*wait-for-it.sh: localhost:8929 is available after 0 seconds", 0)
soc.close()
def test_failed_command_execution(self):
"""
Check command failure. The command in question outputs STDERR and
an exit code of 2
"""Check command failure.
The command in question outputs STDERR and an exit code of 2
"""
soc = self.open_local_port(port=8929)
self.check_args(
"localhost:8929 -- ls not_real_file",
"",
".*No such file or directory\n",
2
)
self.check_args("localhost:8929 -- ls not_real_file", "",
".*No such file or directory\n", 2)
soc.close()
def test_command_after_connection_failure(self):
"""
Test that a command still runs even if a connection times out
and that the return code is correct for the comand being run
"""
"""Test that a command still runs even if a connection times out and
that the return code is correct for the comand being run."""
self.check_args(
"localhost:8929 --timeout=1 -- echo \"CMD OUTPUT\"",
"CMD OUTPUT",
".*timeout occurred after waiting 1 seconds for localhost:8929",
0
)
"localhost:8929 --timeout=1 -- echo \"CMD OUTPUT\"", "CMD OUTPUT",
".*timeout occurred after waiting 1 seconds for localhost:8929", 0)
if __name__ == '__main__':
unittest.main()
tox.ini 0 → 100644
+ 33
− 0
View file @ 0fe88338
# Tox (http://tox.testrun.org/) is a tool for running tests
# in multiple virtualenvs. This configuration file will run the
# test suite on all supported python versions. To use it, "pip install tox"
# and then run "tox" from this directory.
[tox]
envlist = py[27]
skip_missing_interpreters = True
# do not fail on missing setup.py
skipsdist = true
#pip_pre=False
[testenv]
deps =
subprocess32
commands =
python test/wait-for-it.py
[testenv:reformat]
basepython =
python2.7
deps =
yapf
docformatter
whitelist_externals =
find
bash
commands =
- bash -c \
"for n in $(find test '(' -name '*.py' ')'); do \
yapf --in-place $n || echo reformat failed at $n; \
docformatter --in-place $n; \
done"
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment