Source code for schrodinger.test.pytest.fixture
"""
Features that a developer might use when writing tests.
"""
import contextlib
import os
import pathlib
import shutil
import sys
from unittest.mock import patch
import pytest
from rdkit import Chem
from schrodinger import get_mmshare_version
from schrodinger.infra import mmjob
from schrodinger.test import mmshare_testfile
from schrodinger.test.fixtures import safe_patch_os_environ
from . import exetest
[docs]@pytest.fixture
def unittest_preset_manager():
"""
A fixture that mocks out the PresetManager with a PresetManager
that uses the current working directory as the presets directory.
This prevents reliance on system state.
"""
def _getPresetsDirectory(self):
return os.path.join('.', 'foo', self._panel_name)
with patch('schrodinger.models.presets.PresetManager._getPresetsDirectory',
_getPresetsDirectory):
yield
[docs]@pytest.fixture
def json_version():
"""
A fixture that mocks out the version used by
schrodinger.models.json.JsonableClassMixin. By default, the version returned
will be the current mmshare version. The return_value of the fixture can be
adjusted to change the version. Example::
def my_test(json_version):
from schrodinger.models import json
from schrodinger import get_mmshare_version
foo = json.JsonableClassMixin()
assert foo.get_version() == get_mmshare_version()
json_version.return_value = 10
assert foo.get_version() == 10
"""
with patch('schrodinger.models.json.JsonableClassMixin.get_version'
) as get_version_mock:
get_version_mock.return_value = get_mmshare_version()
yield get_version_mock
[docs]@pytest.fixture
def tmp_cwd(tmpdir_factory, request, monkeypatch):
nodeid = os.path.basename(request.node.nodeid)
# replace special characters that break directory paths
for badchar in ':.[]':
nodeid = nodeid.replace(badchar, '_')
# nodeid = 'sort_test.sdaf'
d = tmpdir_factory.mktemp(nodeid)
setattr(request.node, 'schro_tmp_cwd', str(d))
cwd = os.getcwd()
try:
os.chdir(d)
yield str(d)
finally:
os.chdir(cwd)
[docs]@pytest.fixture
def log_to_stdout():
"""Log to stdout instead of files"""
with patch('logging.FileHandler._open', return_value=sys.stdout):
yield
class _allow_memtest:
"""
Run a command from a Python test.
For legacy mmlibs tests
Raises an exception if the command fails or if memtest fails.
"""
def __init__(self, request):
self.node = request.node
def check_call(self, cmd, env=None, stdout=None, stderr=None):
# This is all stuff that is required for memtest to accurately
# find suppressions, for instance.
t = exetest.ExecutableTest.from_parent(name=self.node.fspath.basename,
parent=self.node)
exe_candidates = [cmd[0], f"{cmd[0]}.exe"]
exe_candidates.extend([shutil.which(exe) for exe in exe_candidates])
exe_candidates = [
pathlib.Path(exe) for exe in exe_candidates if exe is not None
]
executable = None
for cand in exe_candidates:
if cand.exists():
executable = os.fspath(cand.absolute())
break
if executable is None:
raise RuntimeError(f"executable {cmd[0]} not found")
cmd = [executable] + cmd[1:]
t.command = cmd
t.runtest(capture=False, env=env, stdout=stdout, stderr=stderr)
[docs]@pytest.fixture
def allow_memtest(request):
# class-based fixtures are not supported.
return _allow_memtest(request)
[docs]@pytest.fixture
def fast_jobdj():
"""
Use this fixture to make jobDJ use minimal delays in updating, at the
expense of throttling the CPU.
"""
with patch("schrodinger.job.queue.USE_JOB_CONTROL_MESSAGES", False):
with patch("schrodinger.job.queue.MONITOR_DELAY", 0.1):
yield
[docs]@pytest.fixture
def mock_gpgpu_hosts_env():
try:
with safe_patch_os_environ({
"SCHRODINGER_HOSTS":
mmshare_testfile("job_test_files/gpgpu.hosts")
}):
mmjob.mmjob_hosts_reload()
yield
finally:
mmjob.mmjob_hosts_reload()
class _MockTaskHelper(contextlib.ExitStack):
"""
Helper class to mock task methods that are needed for the task to finish.
The mocked task will have task.status == task.FAILED
"""
def patch_run(self, task):
"""
Patch task.run and force the task to fail
:rtype: mock.Mock
"""
return self._patch_task(task, "run")
def patch__launchCmd(self, task):
"""
Patch task._launchCmd and force the task to fail
:rtype: mock.Mock
"""
return self._patch_task(task, "_launchCmd")
def _patch_task(self, task, method_name):
patcher = patch.object(task, method_name, side_effect=RuntimeError)
return self.enter_context(patcher)
[docs]@pytest.fixture
def mock_task_helper():
"""
Fixture for mocking task methods that are required for the task to finish
without leaving the task with status RUNNING.
Usage::
def test_foo_task(mock_task_helper):
task = FooTask()
mock_task_helper.patch_run(task)
task.start()
assert task.status is task.FAILED
"""
with _MockTaskHelper() as helper:
yield helper