# Copyright (c) 2016 The CEF Python authors. All rights reserved.
"""Run unit tests. With no arguments all tests are run. Read notes below.
Usage:
_test_runner.py [FILE | _TESTCASE]
Options:
FILE Run tests from single file
_TESTCASE Test cases matching pattern to run eg "file.TestCase".
Calling with this argument is for internal use only.
It has side effects, so don't use it. See comments.
Notes:
- Files starting with "_" are ignored
- If test case name contains "IsolatedTest" word then this test
case will be run using a new instance of Python interpreter.
In such case instead of calling "unittest.main()" use this code:
"import _runner; _runner.main(os.path.basename(__file__))".
- Tested only with TestCase objects. TestSuite usage is untested.
"""
import unittest
import os
import sys
from os.path import dirname, realpath
import re
import subprocess
def main(file_arg=""):
# type: (str) -> None
"""Main entry point."""
# Set working dir to script's current
os.chdir(dirname(realpath(__file__)))
# Script arguments
testcase_arg = ""
if len(sys.argv) > 1:
if ".py" in sys.argv[1]:
file_arg = sys.argv[1]
else:
testcase_arg = sys.argv[1]
# Run tests
runner = TestRunner()
if testcase_arg:
runner.run_testcase(testcase_arg)
elif file_arg:
runner.run_file(file_arg)
else:
runner.run_all()
class TestRunner(object):
"""Customized test runner."""
ran = 0
errors = 0
failures = 0
cefpython_version = "-unknown-"
_suites = None # type: unittest.TestSuite
_isolated_suites = None # type: unittest.TestSuite
_import_errors = None # type: unittest.TestSuite
def _reset_state(self):
# type: () -> None
"""Reset TestRunner state before test discovery."""
self.ran = 0
self.errors = 0
self.failures = 0
self._suites = unittest.TestSuite()
self._isolated_suites = unittest.TestSuite()
self._import_errors = unittest.TestSuite()
# ---- Public methods
def run_testcase(self, testcase):
# type: (str) -> None
"""Run single test case eg 'foo.BarTest'. This is needed to
run single testcase that is marked as IsolatedTest."""
self._discover("[!_]*.py", testcase)
assert not self._count_suites(self._isolated_suites)
if not self._count_suites(self._suites):
print("[_test_runner.py] ERROR: test case not found")
sys.exit(1)
# Import errors found during discovery are ignored
self._run_suites(self._suites)
self._exit()
def run_file(self, filename):
# type: (str) -> None
"""Run test cases from a specific file. This is needed so that
you can use _runner.main() in isolated tests."""
self._discover(filename)
self._run_discovered_suites()
def run_all(self):
# type: () -> None
"""Run all tests from current directory."""
self._discover("[!_]*.py")
self._run_discovered_suites()
# ---- Private methods
def _run_discovered_suites(self):
# type: () -> None
"""Run both normal and isolated suites."""
suites = self._merge_suites(self._import_errors, self._suites)
self._run_suites(suites)
self._run_suites_in_isolation(self._isolated_suites)
self._print_summary()
def _run_suites(self, suites):
# type: (unittest.TestSuite) -> None
"""Run suites."""
if not self._count_suites(suites):
return
runner = unittest.TextTestRunner(verbosity=2, descriptions=True,
buffer=False)
# Update "ran" before running suites, because after ran
# counting them doesn't work (Python 3 issue).
self.ran += self._count_suites(suites)
result = runner.run(suites)
self.errors += len(result.errors)
self.failures += len(result.failures)
def _run_suites_in_isolation(self, suites):
# type: (unittest.TestSuite) -> None
"""Run each suite using new instance of Python interpreter."""
if not self._count_suites(suites):
return
for suite in suites:
# Find test case identifier
testcase_id = ""
for testcase in suite:
testcase_id = testcase.id()
break
# Run test using new instance of Python interpreter
try:
output = subprocess.check_output(
[sys.executable, "_test_runner.py", testcase_id],
stderr=subprocess.STDOUT)
exit_code = 0
except subprocess.CalledProcessError as exc:
output = exc.output
exit_code = exc.returncode
if type(output) != str:
output = output.decode("utf-8", errors="replace")
# Fetch number of sub-tests ran from output
match = re.search(r"^Ran (\d+) sub-tests in \w+", output,
re.MULTILINE)
if match:
self.ran += int(match.group(1))
# Fetch CEF Python version from output
match = re.search(r"^CEF Python (\d+\.\d+)", output,
re.MULTILINE)
if match:
self.cefpython_version = match.group(1)
# Write original output
sys.stdout.write(output)
# If tests failed parse output for errors/failures
if exit_code:
if output:
lines = output.splitlines()
lastline = lines[len(lines)-1]
match = re.search(r"failures=(\d+)", lastline)
if match:
self.failures += int(match.group(1))
match = re.search(r"errors=(\d+)", lastline)
if match:
self.errors += int(match.group(1))
if not self.errors and not self.failures:
self.errors += 1
elif output:
# Test case still might have failed and unittest would not
# detect this. For example when assertion fails
# in ClientHandler in core_test.py .
if "Traceback (most recent call last)" in output\
or "AssertionError" in output:
self.errors += 1
# Update ran
self.ran += self._count_suites(suites)
def _count_suites(self, suites):
# type: (unittest.TestSuite) -> int
count = 0
for suite in suites:
if isinstance(suite, unittest.TestSuite):
for _ in suite:
count += 1
return count
def _merge_suites(self, suites1, suites2):
# type: (unittest.TestSuite, unittest.TestSuite) -> unittest.TestSuite
merged = unittest.TestSuite()
for suite in suites1:
merged.addTest(suite)
for suite in suites2:
merged.addTest(suite)
return merged
def _discover(self, pattern, testcase_name=""):
# type: (str, str) -> None
"""Test discovery using glob pattern from arg or main()."""
self._reset_state()
loader = unittest.TestLoader()
discovered_suite = loader.discover(start_dir=".", pattern=pattern)
for level1_suite in discovered_suite:
for level2_suite in level1_suite:
if isinstance(level2_suite, unittest.TestSuite):
for testcase_obj in level2_suite:
if testcase_name:
if re.match(re.escape(testcase_name),
testcase_obj.id()):
self._suites.addTest(level2_suite)
break
elif "IsolatedTest" in testcase_obj.id():
self._isolated_suites.addTest(level2_suite)
break
else:
self._suites.addTest(level2_suite)
break
elif not testcase_name:
# unittest.loader.ModuleImportFailure
# Warning: If there is an import error in a file
# containing a test case that is being run then there
# won't be any error displayed about it when running
# that single test case. However running a single test
# case is for internal use only, to run test cases in
# isolation with a new instance of Python interpreter.
# Import errors will always be showed when running all
# tests or tests from file.
self._import_errors.addTest(level2_suite)
def _print_summary(self):
# type: () -> None
"""Print summary and exit."""
print("-"*70)
print("[_test_runner.py] CEF Python {ver}"
.format(ver=self.cefpython_version))
print("[_test_runner.py] Python {ver}".format(ver=sys.version[:6]))
print("[_test_runner.py] Ran {ran} tests in total"
.format(ran=self.ran))
if self.errors or self.failures:
failed_str = "[_test_runner.py] FAILED ("
if self.failures:
failed_str += ("failures="+str(self.failures))
if self.errors:
if self.failures:
failed_str += ", "
failed_str += ("errors="+str(self.errors))
failed_str += ")"
print(failed_str)
else:
print("[_test_runner.py] OK")
self._exit()
def _exit(self):
# type: () -> None
"""Exit with appropriate exit code."""
if self.errors or self.failures:
sys.exit(1)
else:
sys.exit(0)
if __name__ == "__main__":
main()