Source code for schrodinger.test.fakejob

import os
from unittest.mock import patch

from schrodinger.infra import mmjob
from schrodinger.job import jobcontrol
from schrodinger.Qt import QtCore

from . import mmshare_testfile


def get_fake_job():
    return get_job(mmshare_testfile("job_test_files/isabel-0-42f280f5"))


def get_fake_running_job():
    return get_job(mmshare_testfile("job_test_files/isabel-0-42f280f6"))


def get_job(filename):
    """
    Return a job object from a text file.
    """
    with patch('schrodinger.job.jobcontrol.Job.readAgain'):
        job = jobcontrol.Job(job_id=os.path.basename(filename))
        with open(filename) as fh:
            job._cpp_job = mmjob.job_from_job_record_text(fh.read())
            return job


class FakeJobDownloader(QtCore.QObject):
    """
    Drop-in replacement for the JobDownloader class, immediately emits a signal
    indicating a successful download.
    """
    downloadFinished = QtCore.pyqtSignal(str)

    def __init__(self, jobid):
        super().__init__()
        self.output = ''

    def download(self):
        self.downloadFinished.emit(self.output)


class FakeFailedJobDownloader(FakeJobDownloader):
    """
    Like FakeJobDownloader, but emits its signal with a non-empty string,
    indicating a download failure.
    """
    FAKE_OUTPUT = 'fake download failed'

    def __init__(self, jobid):
        super().__init__(jobid)
        self.output = FakeFailedJobDownloader.FAKE_OUTPUT


class FakeJobLauncher(QtCore.QObject):
    """
    Drop-in replacement for the JobLauncher class, immediately emits a signal
    indicating a successful download.
    """

    # hack to provide class var to show last value
    last_job_cmd = None

    jobStarted = QtCore.pyqtSignal(jobcontrol.Job)
    jobLaunchFailed = QtCore.pyqtSignal(str)

    def __init__(self, job_cmd):
        # set cls variable
        FakeJobLauncher.last_job_cmd = job_cmd
        super().__init__()

    def launch(self):
        QtCore.QTimer.singleShot(0, self._launchFinished)

    def _launchFinished(self):
        self.jobStarted.emit(get_fake_job())