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 schrodinger import get_mmshare_version
from schrodinger.infra import mmjob
from schrodinger.test import mmshare_testfile

from . import exetest


@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


@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


@pytest.fixture
def tmp_cwd(tmpdir_factory, request, monkeypatch):
    nodeid = os.path.basename(request.node.nodeid)
    nodeid = nodeid.replace(':', '_').replace('.', '_')
    # 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)


@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)


@pytest.fixture
def allow_memtest(request):
    # class-based fixtures are not supported.
    return _allow_memtest(request)


@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


@pytest.fixture
def mock_gpgpu_hosts_env():
    try:
        with patch.dict(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)


@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