Source code
Revision control
Copy as Markdown
Other Tools
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import argparse
import logging
import os
import subprocess
import tempfile
from concurrent.futures import ThreadPoolExecutor, as_completed, thread
from multiprocessing import cpu_count
import mozinfo
from mach.decorators import Command, CommandArgument
from manifestparser import TestManifest
from manifestparser import filters as mpf
from mozfile import which
from tqdm import tqdm
@Command("python", category="devenv", description="Run Python.")
@CommandArgument(
"--exec-file", default=None, help="Execute this Python file using `exec`"
)
@CommandArgument(
"--ipython",
action="store_true",
default=False,
help="Use ipython instead of the default Python REPL.",
)
@CommandArgument(
"--virtualenv",
default=None,
help="Prepare and use the virtualenv with the provided name. If not specified, "
"then the Mach context is used instead.",
)
@CommandArgument("args", nargs=argparse.REMAINDER)
def python(
command_context,
exec_file,
ipython,
virtualenv,
args,
):
# Avoid logging the command
command_context.log_manager.terminal_handler.setLevel(logging.CRITICAL)
# Note: subprocess requires native strings in os.environ on Windows.
append_env = {"PYTHONDONTWRITEBYTECODE": str("1")}
if virtualenv:
command_context._virtualenv_name = virtualenv
if exec_file:
command_context.activate_virtualenv()
exec(open(exec_file).read())
return 0
if ipython:
if virtualenv:
command_context.virtualenv_manager.ensure()
python_path = which(
"ipython", path=command_context.virtualenv_manager.bin_path
)
if not python_path:
raise Exception(
"--ipython was specified, but the provided "
'--virtualenv doesn\'t have "ipython" installed.'
)
else:
command_context._virtualenv_name = "ipython"
command_context.virtualenv_manager.ensure()
python_path = which(
"ipython", path=command_context.virtualenv_manager.bin_path
)
else:
command_context.virtualenv_manager.ensure()
python_path = command_context.virtualenv_manager.python_path
return command_context.run_process(
[python_path] + args,
pass_thru=True, # Allow user to run Python interactively.
ensure_exit_code=False, # Don't throw on non-zero exit code.
python_unbuffered=False, # Leave input buffered.
append_env=append_env,
)
@Command(
"python-test",
category="testing",
virtualenv_name="python-test",
description="Run Python unit tests with pytest.",
)
@CommandArgument(
"-v", "--verbose", default=False, action="store_true", help="Verbose output."
)
@CommandArgument(
"-j",
"--jobs",
default=None,
type=int,
help="Number of concurrent jobs to run. Default is the number of CPUs "
"in the system.",
)
@CommandArgument(
"-x",
"--exitfirst",
default=False,
action="store_true",
help="Runs all tests sequentially and breaks at the first failure.",
)
@CommandArgument(
"--subsuite",
default=None,
help=(
"Python subsuite to run. If not specified, all subsuites are run. "
"Use the string `default` to only run tests without a subsuite."
),
)
@CommandArgument(
"tests",
nargs="*",
metavar="TEST",
help=(
"Tests to run. Each test can be a single file or a directory. "
"Default test resolution relies on PYTHON_UNITTEST_MANIFESTS."
),
)
@CommandArgument(
"extra",
nargs=argparse.REMAINDER,
metavar="PYTEST ARGS",
help=(
"Arguments that aren't recognized by mach. These will be "
"passed as it is to pytest"
),
)
def python_test(command_context, *args, **kwargs):
try:
tempdir = str(tempfile.mkdtemp(suffix="-python-test"))
os.environ["PYTHON_TEST_TMP"] = tempdir
return run_python_tests(command_context, *args, **kwargs)
finally:
import mozfile
mozfile.remove(tempdir)
def run_python_tests(
command_context,
tests=None,
test_objects=None,
subsuite=None,
verbose=False,
jobs=None,
exitfirst=False,
extra=None,
**kwargs,
):
if test_objects is None:
from moztest.resolve import TestResolver
resolver = command_context._spawn(TestResolver)
# If we were given test paths, try to find tests matching them.
test_objects = resolver.resolve_tests(paths=tests, flavor="python")
else:
# We've received test_objects from |mach test|. We need to ignore
# the subsuite because python-tests don't use this key like other
# harnesses do and |mach test| doesn't realize this.
subsuite = None
mp = TestManifest()
mp.tests.extend(test_objects)
filters = []
if subsuite == "default":
filters.append(mpf.subsuite(None))
elif subsuite:
filters.append(mpf.subsuite(subsuite))
tests = mp.active_tests(filters=filters, disabled=False, python=3, **mozinfo.info)
if not tests:
submsg = "for subsuite '{}' ".format(subsuite) if subsuite else ""
message = (
"TEST-UNEXPECTED-FAIL | No tests collected "
+ "{}(Not in PYTHON_UNITTEST_MANIFESTS?)".format(submsg)
)
command_context.log(logging.WARN, "python-test", {}, message)
return 1
parallel = []
sequential = []
os.environ.setdefault("PYTEST_ADDOPTS", "")
if extra:
os.environ["PYTEST_ADDOPTS"] += " " + " ".join(extra)
installed_requirements = set()
for test in tests:
if (
test.get("requirements")
and test["requirements"] not in installed_requirements
):
command_context.virtualenv_manager.install_pip_requirements(
test["requirements"], quiet=True
)
installed_requirements.add(test["requirements"])
if exitfirst:
sequential = tests
os.environ["PYTEST_ADDOPTS"] += " -x"
else:
for test in tests:
if test.get("sequential"):
sequential.append(test)
else:
parallel.append(test)
jobs = jobs or cpu_count()
return_code = 0
failure_output = []
def on_test_finished(result):
output, ret, test_path = result
if ret:
# Log the output of failed tests at the end so it's easy to find.
failure_output.extend(output)
if not return_code:
command_context.log(
logging.ERROR,
"python-test",
{"test_path": test_path, "ret": ret},
"Setting retcode to {ret} from {test_path}",
)
else:
for line in output:
command_context.log(
logging.INFO, "python-test", {"line": line.rstrip()}, "{line}"
)
return return_code or ret
with tqdm(
total=(len(parallel) + len(sequential)),
unit="Test",
desc="Tests Completed",
initial=0,
) as progress_bar:
try:
with ThreadPoolExecutor(max_workers=jobs) as executor:
futures = []
for test in parallel:
command_context.log(
logging.DEBUG,
"python-test",
{"line": f"Launching thread for test {test['file_relpath']}"},
"{line}",
)
futures.append(
executor.submit(
_run_python_test, command_context, test, jobs, verbose
)
)
try:
for future in as_completed(futures):
progress_bar.clear()
return_code = on_test_finished(future.result())
progress_bar.update(1)
except KeyboardInterrupt:
# Hack to force stop currently running threads.
executor._threads.clear()
thread._threads_queues.clear()
raise
for test in sequential:
test_result = _run_python_test(command_context, test, jobs, verbose)
progress_bar.clear()
return_code = on_test_finished(test_result)
if return_code and exitfirst:
break
progress_bar.update(1)
finally:
progress_bar.clear()
# Now log all failures (even if there was a KeyboardInterrupt or other exception).
for line in failure_output:
command_context.log(
logging.INFO, "python-test", {"line": line.rstrip()}, "{line}"
)
command_context.log(
logging.INFO,
"python-test",
{"return_code": return_code},
"Return code from mach python-test: {return_code}",
)
return return_code
def _run_python_test(command_context, test, jobs, verbose):
output = []
def _log(line):
# Buffer messages if more than one worker to avoid interleaving
if jobs > 1:
output.append(line)
else:
command_context.log(
logging.INFO, "python-test", {"line": line.rstrip()}, "{line}"
)
_log(test["path"])
python = command_context.virtualenv_manager.python_path
cmd = [python, test["path"]]
env = os.environ.copy()
env["PYTHONDONTWRITEBYTECODE"] = "1"
result = subprocess.run(
cmd,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
encoding="UTF-8",
)
return_code = result.returncode
file_displayed_test = False
for line in result.stdout.split(os.linesep):
if not file_displayed_test:
test_ran = "Ran" in line or "collected" in line or line.startswith("TEST-")
if test_ran:
file_displayed_test = True
# Hack to make sure treeherder highlights pytest failures
if "FAILED" in line.rsplit(" ", 1)[-1]:
line = line.replace("FAILED", "TEST-UNEXPECTED-FAIL")
_log(line)
if not file_displayed_test:
return_code = 1
_log(
"TEST-UNEXPECTED-FAIL | No test output (missing mozunit.main() "
"call?): {}".format(test["path"])
)
if verbose:
if return_code != 0:
_log("Test failed: {}".format(test["path"]))
else:
_log("Test passed: {}".format(test["path"]))
return output, return_code, test["path"]