"""
Schrodinger-specific modification to pytest startup.
"""
import argparse
import enum
import faulthandler
import glob
import os
import pathlib
import sys
import warnings
import pytest
import schrodinger
import schrodinger.job.util
from schrodinger.job import jobcontrol
from schrodinger.job import server
from schrodinger.utils.sysinfo import is_display_present
from schrodinger.test.hypothesis import hypothesis_profiles
from schrodinger.test.jobserver import SCHRODINGER_JOBSERVER_CONFIG_FILE
from schrodinger.utils import mmutil
from schrodinger.utils.env import prepend_sys_path
from . import _i_am_buildbot
from . import faulthandler_setup
from . import reporter
from .warnings import mark_warnings_as_errors
CURRENT_SESSION = None
hypothesis_profiles.register_profiles()
[docs]@enum.unique
class SchrodingerIniOptions(enum.Enum):
ALLOW_REMOTE_JOBS = "allow_remote_jobs"
DISALLOW_MOCK_IN_SWIG = "disallow_mock_in_swig"
WARNINGS_AS_ERRORS = "warnings_as_errors"
[docs]class DeferXdistPlugin:
"""
Plugin to defer pytest-xdist hook functions.
"""
[docs] def pytest_testnodedown(self, node, error):
"""
Check for core dumps if a node crashes
"""
if not error:
return
proc = node.gateway._io.popen
# Empirically, the process is not done at this point;
# do a short wait for it to finish
returncode = proc.wait(timeout=1)
pid = proc.pid
reporter.print_crashed_process(pid, returncode)
[docs]def addoption(parser):
"""
Add Schrodinger options to run the post tests, or just the fastest tests.
This is a pytest hook.
"""
def product(product_name):
"""
Hunt for a product. Allows a failure for uninstalled product before
tests are discovered.
"""
product_exec = schrodinger.job.util.hunt(product_name)
if not product_exec:
raise argparse.ArgumentTypeError(
f'Product "{product_name}" is not installed.')
return (product_name, product_exec)
group = parser.getgroup('Schrodinger options')
group.addoption('--run-in-dir',
action="store_true",
help='Execute each test in the directory of the test file.')
group.addoption(
'--pypath',
action='append',
help=('Add the requested path to the PYTHONPATH before executing '
'tests.'))
parser.addini(
'pypath',
help=(
'Add the requested path to the PYTHONPATH before executing tests.'))
# Setting default product based on -FROM, this makes -FROM and --product
# synonyms
default = os.environ.get('SCHRODINGER_PRODUCT', None)
help_msg = ('Also available as -FROM. Sets environment as if called with '
'$SCHRODINGER/run -FROM <PRODUCT>. Also adds the requested '
"product's bin directory to the PYTHONPATH.")
if default:
help_msg += f' (DEFAULT={default})'
default = product(default)
group.addoption('--product',
type=product,
dest='from_product',
default=default,
help=help_msg)
parser.addini(
'src_dirname',
help="Directory name of product source repository, eg: 'maestro-src'")
group.addoption('--fast',
action="store_true",
help='Skip tests marked as as slow.')
# This is never used in automated builds. I use it all the time, though,
# to run both post tests and not post tests for a directory.
group.addoption('--post_test',
'--post-test',
action="store_true",
help=("Include tests that are run from the "
"post_test' target."))
group.addoption(
'--post_test-only',
'--post_test_only',
'--post-test-only',
action="store_true",
help=("Run only the tests that rely on more than mmshare (i.e. the "
"'post_test' target)."))
if sys.platform.startswith('linux'):
msg = "Execute non python tests using valgrind"
else:
msg = argparse.SUPPRESS
group.addoption('--memtest', action="store_true", help=msg)
group.addoption('--default-feature-flags',
action='store_true',
help=('Use the feature flag settings from the '
'installation, ignoring those in .schrodinger'))
group.addoption(
'--no-display',
action='store_true',
help='Simulate running on a computer with no display available. '
'Tests marked require_display will be skipped, and the '
'qapplication will not be started')
parser.addini(SchrodingerIniOptions.DISALLOW_MOCK_IN_SWIG.value,
type="bool",
default=True,
help="Raise a TypeError if a MagicMock is passed in to SWIG")
parser.addini(SchrodingerIniOptions.WARNINGS_AS_ERRORS.value,
type="bool",
default=False,
help="Turn any python warnings into errors")
# It would be nice to kill each thread after running at most X tests. I
# like that idea so we can avoid the OOM errors on win32.
parser.addini(
SchrodingerIniOptions.ALLOW_REMOTE_JOBS.value,
type="bool",
default=False,
help="If True, allow launching remote job server jobs, "
"otherwise hide remote JOB_SERVERs to make job manager access faster")
[docs]def extend_pythonpath(additional_paths):
"""
Add "additional_paths" to the PYTHONPATH. Also adds schrodinger.test and
the test_modules directory.
"""
if not additional_paths:
additional_paths = []
# Need to manipulate both sys.path and PYTHONPATH to deal with multiple
# processes.
pythonpath = os.environ.get('PYTHONPATH', [])
if pythonpath:
pythonpath = [pythonpath]
else:
pythonpath = []
def add_to_pypath(dirname):
"""
Adds to sys.path and PYTHONPATH in case processes are spawned. This
definitely happens, for instance, when running tests in parallel.
"""
pythonpath.append(dirname)
sys.path.append(dirname)
for dirname in additional_paths:
add_to_pypath(dirname)
os.environ['PYTHONPATH'] = os.pathsep.join(pythonpath)
def _relaunch_pytest(*additional_arguments):
"""Relaunch py.test, possibly with some additional arguments"""
with prepend_sys_path(os.environ['MMSHARE_EXEC']):
import toplevel
import shlex
# regenerate the command line that pytest was called with, but add
# -FROM product.
pytest_utility = shlex.split(os.environ['SCHRODINGER_COMMANDLINE'])[0]
basecmd = [
toplevel.__file__, pytest_utility, 'mmshare', '', 'MMSHARE_EXEC',
os.path.basename(sys.executable), '-m', 'pytest'
]
cmd = basecmd + list(additional_arguments) + sys.argv[1:]
return toplevel.main(cmd)
[docs]def cmdline_main(config):
"""
Run the py.test main loop.
Only affects if an option was requested that necessitates restarting
toplevel.
"""
if config.option.default_feature_flags:
if os.environ.get('SCHRODINGER_FEATURE_FLAGS', None) == '':
# Already set to ignore feature flags from .schrodinger.
config.option.default_feature_flags = False
else:
os.environ['SCHRODINGER_FEATURE_FLAGS'] = ''
# Turn on DEV DEBUG to make sure custom exception handler is not invoked
os.environ["SCHRODINGER_DEV_DEBUG"] = "True"
if config.option.from_product:
# Make sure that toplevel was used to set the product directory.
# if not, overload pytest's main function.
schrodinger_product = os.environ.get('SCHRODINGER_PRODUCT')
if schrodinger_product != config.option.from_product[0]:
if schrodinger_product and schrodinger_product != 'mmshare':
raise pytest.UsageError(
"Product set using -FROM and --product must match, "
f"currently -FROM={schrodinger_product} and "
f"--product={config.option.from_product[0]}")
# pytest was called with --product <product name>, but not -FROM.
# This means that toplevel needs to be invoked to set the
# environment up correctly.
return _relaunch_pytest('-FROM', config.option.from_product[0])
if config.option.default_feature_flags:
return _relaunch_pytest()
# https://github.com/pytest-dev/pytest/issues/6936
warnings.filterwarnings("ignore",
message="The TerminalReporter.writer",
category=pytest.PytestDeprecationWarning)
if mmutil.feature_flag_is_enabled(
mmutil.JOB_SERVER) and not jobcontrol.get_backend():
allow_remote_jobs = SCHRODINGER_JOBSERVER_CONFIG_FILE in os.environ or config.getini(
SchrodingerIniOptions.ALLOW_REMOTE_JOBS.value)
if not allow_remote_jobs:
os.environ[
SCHRODINGER_JOBSERVER_CONFIG_FILE] = "REMOTE_JOBS_HIDDEN_FROM_PYTEST"
server.ensure_localhost_server_running()
if config.getini(SchrodingerIniOptions.WARNINGS_AS_ERRORS.value):
mark_warnings_as_errors()
[docs]class CurrentSession:
[docs] def __init__(self, rootdir, pluginmanager):
self.rootdir = rootdir
self.pluginmanager = pluginmanager
[docs]def set_current_session(config):
"""
:param config: current config object for pytest
:type config: pytest.config.Config
Sets a module level variable to refer to in case of crash.
"""
global CURRENT_SESSION
CURRENT_SESSION = CurrentSession(config.rootdir, config.pluginmanager)
[docs]def can_write_bytecode():
"""
Return whether we can write bytecode to the __pycache__ directory.
"""
test_file = pathlib.Path(
pathlib.Path(__file__).parent) / "__pycache__" / ".writeable_file.py"
try:
test_file.write_text("")
except (FileNotFoundError, PermissionError):
return False
return True
[docs]def disable_bytecode_if_not_writable():
"""
Determines if we need to disable bytecode writing, due to pytest using
atomicwrites package, which uses mkstemp. It will attempt to create 2
billion files on windows, per file imported by pytest.
https://bugs.python.org/issue22107
"""
if sys.platform.startswith("win32") and not can_write_bytecode():
sys.dont_write_bytecode = True
[docs]def get_schrodinger_package_dirs():
"""
Get directories under the schrodinger module that contain code from product
repositories outside of mmshare
"""
schrodinger_dir = os.path.dirname(schrodinger.__file__)
package_dirs = glob.glob(os.path.join(schrodinger_dir, '*', 'packages'))
package_dirs.extend(
glob.glob(os.path.join(schrodinger_dir, 'application', '*',
'packages')))
package_dirs.extend(
glob.glob(os.path.join(schrodinger_dir, 'application', 'epik')))
return package_dirs