Browse Source

[tests] add (colored) progress output

master
Stefan Bühler 11 years ago
parent
commit
8ffcd3e246
  1. 55
      tests/base.py
  2. 70
      tests/logfile.py
  3. 18
      tests/requests.py
  4. 4
      tests/runtests.py

55
tests/base.py

@ -25,6 +25,7 @@ A test intance has the following attributes:
GroupTest will set the vhost of subtests to the vhost of the GroupTest
if the subtest doesn't provide a config
* runnable: whether to call Run
* todo: whether the test is expected to fail
You can create files and directories in Prepare with TestBase.{PrepareVHostFile,PrepareFile,PrepareDir};
they will get removed on cleanup automatically (if the test was successful).
@ -79,6 +80,7 @@ class TestBase(object):
name = None
vhost = None
runnable = True
todo = False
def __init__(self):
self._test_cleanup_files = []
@ -119,7 +121,7 @@ var.vhosts = var.vhosts + [ "%s" : ${
print >> sys.stderr, "Test %s failed:" % (self.name)
print >> sys.stderr, traceback.format_exc(10)
print >> Env.log, "[Done] Running test %s [result=%s]" % (self.name, failed and "Failed" or "Succeeded")
self._test_failed = failed
self._test_failed = failed and not self.todo
return not failed
def _cleanup(self):
@ -210,12 +212,18 @@ class Tests(object):
self.tests = [] # all tests (we always prepare/cleanup all tests)
self.tests_dict = { }
self.config = None
self.testname_len = 60
self.prepared_dirs = { }
self.prepared_files = { }
self.failed = False
self.stat_pass = 0
self.stat_fail = 0
self.stat_todo = 0
self.stat_done = 0
self.add_service(Lighttpd())
def add_test(self, test):
@ -223,11 +231,12 @@ class Tests(object):
if self.tests_dict.has_key(name):
raise BaseException("Test '%s' already defined" % name)
self.tests_dict[name] = test
for f in self.tests_filter:
if name.startswith(f):
if test.runnable:
if test.runnable:
for f in self.tests_filter:
if name.startswith(f):
self.run.append(test)
break
self.testname_len = max(self.testname_len, len(name))
break
self.tests.append(test)
def add_service(self, service):
@ -322,13 +331,47 @@ allow-listen {{ ip "127.0.0.1:{Env.port}"; }}
raise
print >> Env.log, "[Done] Preparing services"
def _progress(self, i, n):
s = str(n)
return ("[{0:>%i}" % len(s)).format(i) + "/" + s + "]"
def Run(self):
COLOR_RESET = Env.color and "\033[0m" or ""
COLOR_BLUE = Env.color and "\033[1;34m" or ""
COLOR_GREEN = Env.color and "\033[1;32m" or ""
COLOR_YELLOW = Env.color and "\033[1;33m" or ""
COLOR_RED = Env.color and "\033[1;38m" or ""
COLOR_CYAN = Env.color and "\033[1;36m" or ""
PASS = COLOR_GREEN + "[PASS]" + COLOR_RESET
FAIL = COLOR_RED + "[FAIL]" + COLOR_RESET
TODO = COLOR_YELLOW + "[TODO]" + COLOR_RESET
DONE = COLOR_YELLOW + "[DONE]" + COLOR_RESET
testcount = len(self.run)
print >> Env.log, "[Start] Running tests"
fmt = COLOR_BLUE + " {0:<%i} " % self.testname_len
failed = False
i = 1
for t in self.run:
if not t._run(): failed = True
result = t._run()
if t.todo:
print >> sys.stdout, COLOR_CYAN + self._progress(i, testcount) + fmt.format(t.name) + (result and DONE or TODO)
if result:
self.stat_done += 1
else:
self.stat_todo += 1
else:
print >> sys.stdout, COLOR_CYAN + self._progress(i, testcount) + fmt.format(t.name) + (result and PASS or FAIL)
if result:
self.stat_pass += 1
else:
self.stat_fail += 1
failed = True
i += 1
self.failed = failed
print >> sys.stdout, ("%i out of %i tests passed (%.2f%%), %i tests failed, %i todo items, %i todo items are ready" %
(self.stat_pass, testcount, (100.0 * self.stat_pass)/testcount, self.stat_fail, self.stat_todo, self.stat_done))
print >> Env.log, "[Done] Running tests [result=%s]" % (failed and "Failed" or "Succeeded")
return not failed

70
tests/logfile.py

@ -2,7 +2,7 @@
import time
__all__ = [ 'LogFile' ]
__all__ = [ 'LogFile', 'RemoveEscapeSeq' ]
ATTRS = [ 'closed', 'encoding', 'errors', 'mode', 'name', 'newlines', 'softspace' ]
@ -84,3 +84,71 @@ class LogFile(object):
def writelines(self, *args):
return self.write(''.join(args))
def xreadlines(self, *args, **kwargs): return self.file.xreadlines(*args, **kwargs)
class RemoveEscapeSeq(object):
def __init__(self, file):
self.file = file
self.escape_open = False
def __enter__(self, *args, **kwargs): return self.file.__enter__(*args, **kwargs)
def __exit__(self, *args, **kwargs): return self.file.__exit__(*args, **kwargs)
def __iter__(self, *args, **kwargs): return self.file.__iter__(*args, **kwargs)
def __repr__(self, *args, **kwargs): return self.file.__repr__(*args, **kwargs)
def __delattr__(self, name):
if name in ATTRS:
return delattr(self.file, name)
else:
return super(RemoveEscapeSeq, self).__delattr__(name, value)
def __getattr__(self, name):
if name in ATTRS:
return getattr(self.file, name)
else:
return super(RemoveEscapeSeq, self).__getattr__(name, value)
def __getattribute__(self, name):
if name in ATTRS:
return self.file.__getattribute__(name)
else:
return object.__getattribute__(self, name)
def __setattr__(self, name, value):
if name in ATTRS:
return setattr(self.file, name, value)
else:
return super(RemoveEscapeSeq, self).__setattr__(name, value)
def close(self, *args, **kwargs): return self.file.close(*args, **kwargs)
def fileno(self, *args, **kwargs):
pass
def flush(self, *args, **kwargs): return self.file.flush(*args, **kwargs)
def isatty(self, *args, **kwargs): return False
def next(self, *args, **kwargs): return self.file.next(*args, **kwargs)
def read(self, *args, **kwargs): return self.file.read(*args, **kwargs)
def readinto(self, *args, **kwargs): return self.file.readinto(*args, **kwargs)
def readline(self, *args, **kwargs): return self.file.readline(*args, **kwargs)
def readlines(self, *args, **kwargs): return self.file.readlines(*args, **kwargs)
def seek(self, *args, **kwargs):
pass
def tell(self, *args, **kwargs): return self.file.tell(*args, **kwargs)
def truncate(self, *args, **kwargs):
pass
def write(self, str):
while str != "":
if self.escape_open:
l = str.split('m', 1)
if len(l) == 2:
self.escape_open = False
str = l[1]
else:
return
else:
l = str.split('\033', 1)
self.file.write(l[0])
if len(l) == 2:
self.escape_open = True
str = l[1]
else:
return
def writelines(self, *args):
return self.write(''.join(args))
def xreadlines(self, *args, **kwargs): return self.file.xreadlines(*args, **kwargs)

18
tests/requests.py

@ -84,16 +84,16 @@ class CurlRequest(TestBase):
def dump(self):
c = self.curl
sys.stdout.flush()
print >> sys.stdout, "Dumping request for test '%s'" % self.name
print >> sys.stdout, "Curl request: URL = %s://%s:%i%s" % (self.SCHEME, self.vhost, Env.port + self.PORT, self.URL)
print >> sys.stdout, "Curl response code: %i " % (c.getinfo(pycurl.RESPONSE_CODE))
print >> sys.stdout, "Curl response headers:"
Env.log.flush()
print >> Env.log, "Dumping request for test '%s'" % self.name
print >> Env.log, "Curl request: URL = %s://%s:%i%s" % (self.SCHEME, self.vhost, Env.port + self.PORT, self.URL)
print >> Env.log, "Curl response code: %i " % (c.getinfo(pycurl.RESPONSE_CODE))
print >> Env.log, "Curl response headers:"
for (k, v) in self.resp_header_list:
print >> sys.stdout, " %s: %s" % (k, v)
print >> sys.stdout, "Curl response body:"
print >> sys.stdout, self.buffer.getvalue()
sys.stdout.flush()
print >> Env.log, " %s: %s" % (k, v)
print >> Env.log, "Curl response body:"
print >> Env.log, self.buffer.getvalue()
Env.log.flush()
def _checkResponse(self):
c = self.curl

4
tests/runtests.py

@ -3,7 +3,7 @@
import os
import sys
from logfile import LogFile
from logfile import LogFile, RemoveEscapeSeq
from base import Env, Tests
@ -52,11 +52,13 @@ Env.sourcedir = os.path.abspath(os.path.dirname(__file__))
Env.luadir = os.path.join(os.path.dirname(Env.sourcedir), "doc")
Env.debugRequests = options.debug_requests
Env.strace = options.strace
Env.color = sys.stdin.isatty()
Env.dir = mkdtemp(dir = os.getcwd())
Env.defaultwww = os.path.join(Env.dir, "www", "default")
Env.log = open(os.path.join(Env.dir, "tests.log"), "w")
if Env.color: Env.log = RemoveEscapeSeq(Env.log)
sys.stderr = LogFile(sys.stderr, **{ "[stderr]": Env.log })
sys.stdout = LogFile(sys.stdout, **{ "[stdout]": Env.log })
Env.log = LogFile(Env.log)

Loading…
Cancel
Save