# 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 and sys.argv[1].startswith("--"):

# Will allow to pass custom args like --debug to isolated tests

# (main_test.py for example).

global CUSTOM_CMDLINE_ARG

CUSTOM_CMDLINE_ARG = sys.argv[1]

elif 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:

"""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

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

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,

CUSTOM_CMDLINE_ARG],

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} {arch}"

.format(ver=platform.python_version(),

arch=platform.architecture()[0]))

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 all unit tests succeeded")

self._exit()

def _exit(self):

# type: () -> None

"""Exit with appropriate exit code."""

if self.errors or self.failures:

sys.exit(1)

else: