grachevko/wait-for-it
1
0
Fork 0
Code Pull Requests 1 Actions Releases Activity

Fixes to test script for flake8

Browse Source
This commit is contained in:
Douglas Gibbons
2017-05-17 10:18:55 -07:00
parent f9b79b2e51
commit 8ed81e3803
2 changed files with 48 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.travis.yml
View File

@ -2,4 +2,6 @@ language: python
python: python:
- "2.7" - "2.7"
script: python test/wait-for-it.py script:
- python test/wait-for-it.py

95
test/wait-for-it.py
View File

@ -1,5 +1,4 @@
import unittest import unittest
import subprocess
import shlex import shlex
from subprocess import Popen, PIPE from subprocess import Popen, PIPE
import os import os
@ -11,59 +10,58 @@ MISSING_ARGS_TEXT = "Error: you need to provide a host and port to test."
HELP_TEXT = "Usage:" # Start of help text HELP_TEXT = "Usage:" # Start of help text
DIVIDE_LINE = '-'*71 # Output line of dashes DIVIDE_LINE = '-'*71 # Output line of dashes
"""
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.
"""
class TestWaitForIt(unittest.TestCase):
class TestWaitForIt(unittest.TestCase):
def execute(self,cmd): """
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.
"""
def execute(self, cmd):
"""Executes a command and returns exit code, STDOUT, STDERR""" """Executes a command and returns exit code, STDOUT, STDERR"""
args = shlex.split(cmd) args = shlex.split(cmd)
proc = Popen(args, stdout=PIPE, stderr=PIPE) proc = Popen(args, stdout=PIPE, stderr=PIPE)
out, err = proc.communicate() out, err = proc.communicate()
exitcode = proc.returncode exitcode = proc.returncode
return exitcode, out, err return exitcode, out, err
def open_local_port(self,host="localhost", port=8929, timeout=5): def open_local_port(self, host="localhost", port=8929, timeout=5):
s = socket.socket() s = socket.socket()
s.bind((host,port)) s.bind((host, port))
s.listen(timeout) s.listen(timeout)
return s return s
def check_args(self,args,stdout_regex,stderr_regex,exitcode): def check_args(self, args, stdout_regex, stderr_regex, exitcode):
command = self.wait_script + " " + args command = self.wait_script + " " + args
actual_exitcode, out, err = self.execute(command) actual_exitcode, out, err = self.execute(command)
# Check stderr # Check stderr
msg = ("Failed check that STDERR:\n"+ msg = ("Failed check that STDERR:\n" +
DIVIDE_LINE + "\n" + err + "\n" + DIVIDE_LINE + DIVIDE_LINE + "\n" + err + "\n" + DIVIDE_LINE +
"\nmatches:\n" + "\nmatches:\n" +
DIVIDE_LINE + "\n" + stderr_regex + "\n" + DIVIDE_LINE) DIVIDE_LINE + "\n" + stderr_regex + "\n" + DIVIDE_LINE)
self.assertIsNotNone(re.match(stderr_regex,err,re.DOTALL),msg) self.assertIsNotNone(re.match(stderr_regex, err, re.DOTALL), msg)
# Check STDOUT # Check STDOUT
msg = ("Failed check that STDOUT:\n"+ msg = ("Failed check that STDOUT:\n" +
DIVIDE_LINE + "\n" + out + "\n" + DIVIDE_LINE + DIVIDE_LINE + "\n" + out + "\n" + DIVIDE_LINE +
"\nmatches:\n" + "\nmatches:\n" +
DIVIDE_LINE + "\n" + stdout_regex + "\n" + DIVIDE_LINE) DIVIDE_LINE + "\n" + stdout_regex + "\n" + DIVIDE_LINE)
self.assertIsNotNone(re.match(stdout_regex,out,re.DOTALL),msg) self.assertIsNotNone(re.match(stdout_regex, out, re.DOTALL), msg)
# Check exit code # Check exit code
self.assertEqual(actual_exitcode,exitcode) self.assertEqual(actual_exitcode, exitcode)
def setUp(self): def setUp(self):
script_path = os.path.dirname(sys.argv[0]) script_path = os.path.dirname(sys.argv[0])
parent_path = os.path.abspath(os.path.join(script_path, os.pardir)) parent_path = os.path.abspath(os.path.join(script_path, os.pardir))
self.wait_script = os.path.join(parent_path,"wait-for-it.sh") self.wait_script = os.path.join(parent_path, "wait-for-it.sh")
def test_no_args(self): def test_no_args(self):
""" """
Check that no aruments returns the missing args text and the correct Check that no aruments returns the missing args text and the
return code correct return code
""" """
self.check_args( self.check_args(
"", "",
@ -73,15 +71,15 @@ class TestWaitForIt(unittest.TestCase):
) )
# Return code should be 1 when called with no args # Return code should be 1 when called with no args
exitcode, out, err = self.execute(self.wait_script) exitcode, out, err = self.execute(self.wait_script)
self.assertEqual(exitcode,1) self.assertEqual(exitcode, 1)
def test_help(self): def test_help(self):
""" Check that help text is printed with --help argument """ """ Check that help text is printed with --help argument """
self.check_args( self.check_args(
"--help", "--help",
"", "",
HELP_TEXT, HELP_TEXT,
1 1
) )
def test_no_port(self): def test_no_port(self):
@ -112,7 +110,7 @@ class TestWaitForIt(unittest.TestCase):
0 0
) )
soc.close() soc.close()
def test_combined_host_port(self): def test_combined_host_port(self):
""" """
Tests that wait-for-it.sh returns correctly after establishing a Tests that wait-for-it.sh returns correctly after establishing a
@ -126,7 +124,7 @@ class TestWaitForIt(unittest.TestCase):
0 0
) )
soc.close() soc.close()
def test_port_failure_with_timeout(self): 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
@ -134,7 +132,7 @@ class TestWaitForIt(unittest.TestCase):
self.check_args( self.check_args(
"localhost:8929 --timeout=1", "localhost:8929 --timeout=1",
"", "",
".*wait-for-it.sh: timeout occurred after waiting 1 seconds for localhost:8929", ".*timeout occurred after waiting 1 seconds for localhost:8929",
124 124
) )
@ -151,21 +149,19 @@ class TestWaitForIt(unittest.TestCase):
) )
soc.close() soc.close()
def test_failed_command_execution(self): def test_failed_command_execution(self):
""" """
Check command failure. The command in question outputs STDERR and an Check command failure. The command in question outputs STDERR and
exit code of 2 an exit code of 2
""" """
soc = self.open_local_port(port=8929) soc = self.open_local_port(port=8929)
self.check_args( self.check_args(
"localhost:8929 -- ls not_real_file", "localhost:8929 -- ls not_real_file",
"", "",
".*No such file or directory\n", ".*No such file or directory\n",
2 2
) )
soc.close() soc.close()
def test_command_after_connection_failure(self): def test_command_after_connection_failure(self):
""" """
@ -175,10 +171,9 @@ class TestWaitForIt(unittest.TestCase):
self.check_args( self.check_args(
"localhost:8929 --timeout=1 -- echo \"CMD OUTPUT\"", "localhost:8929 --timeout=1 -- echo \"CMD OUTPUT\"",
"CMD OUTPUT", "CMD OUTPUT",
".*wait-for-it.sh: timeout occurred after waiting 1 seconds for localhost:8929", ".*timeout occurred after waiting 1 seconds for localhost:8929",
0 0
) )
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()