Source code for toil.test.src.restartDAGTest

# Copyright (C) 2015-2021 Regents of the University of California
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import logging
import os
import shutil
import signal

from toil.common import Toil
from toil.exceptions import FailedJobsException
from toil.job import Job
from toil.test import ToilTest, slow

logger = logging.getLogger(__name__)




class RestartDAGTest(ToilTest):
    """
    Tests that restarted job DAGs don't run children of jobs that failed in the first run till the
    parent completes successfully in the restart.
    """



    def setUp(self):
        super().setUp()
        self.tempDir = self._createTempDir(purpose="tempDir")
        self.testJobStore = self._getTestJobStorePath()




    def tearDown(self):
        super().tearDown()
        shutil.rmtree(self.testJobStore)




    @slow
    def testRestartedWorkflowSchedulesCorrectJobsOnFailedParent(self):
        self._testRestartedWorkflowSchedulesCorrectJobs("raise")




    @slow
    def testRestartedWorkflowSchedulesCorrectJobsOnKilledParent(self):
        self._testRestartedWorkflowSchedulesCorrectJobs("kill")


    def _testRestartedWorkflowSchedulesCorrectJobs(self, failType):
        """
        Creates a diamond DAG
            /->passingParent-\
        root                 |-->child
           \\->failingParent--/

        where root and passingParent are guaranteed to pass, while failingParent will fail.
        child should not run on start or restart and we assert that by ensuring that a file create
        iff child is run is not present on the system.

        :param str failType: Does failingParent fail on an assertionError, or is it killed.
        """
        # Specify options
        options = Job.Runner.getDefaultOptions(self.testJobStore)
        options.logLevel = "DEBUG"
        options.retryCount = 0
        options.clean = "never"

        parentFile = os.path.join(self.tempDir, "parent")
        childFile = os.path.join(self.tempDir, "child")

        # Make the first job
        root = Job.wrapJobFn(passingFn)
        passingParent = Job.wrapJobFn(passingFn)
        failingParent = Job.wrapJobFn(failingFn, failType=failType, fileName=parentFile)
        child = Job.wrapJobFn(passingFn, fileName=childFile)

        # define the DAG
        root.addChild(passingParent)
        root.addChild(failingParent)
        passingParent.addChild(child)
        failingParent.addChild(child)

        failReasons = []

        assert not os.path.exists(childFile)

        # Run the test
        for runMode in "start", "restart":
            self.errorRaised = None
            try:
                with Toil(options) as toil:
                    if runMode == "start":
                        toil.start(root)
                    else:
                        toil.restart()
            except Exception as e:
                logger.exception(e)
                self.errorRaised = e
            finally:
                # The processing of an AssertionError and FailedJobsException is the same so we do
                # it together in this finally clause.
                if self.errorRaised is not None:
                    if not os.path.exists(parentFile):
                        failReasons.append(
                            'The failing parent file did not exist on toil "%s".'
                            % runMode
                        )
                    if os.path.exists(childFile):
                        failReasons.append(
                            "The child file existed.  i.e. the child was run on "
                            'toil "%s".' % runMode
                        )
                    if isinstance(self.errorRaised, FailedJobsException):
                        if self.errorRaised.numberOfFailedJobs != 3:
                            failReasons.append(
                                'FailedJobsException was raised on toil "%s" but '
                                "the number of failed jobs (%s) was not 3."
                                % (runMode, self.errorRaised.numberOfFailedJobs)
                            )
                    elif isinstance(self.errorRaised, AssertionError):
                        failReasons.append(
                            "Toil raised an AssertionError instead of a "
                            'FailedJobsException on toil "%s".' % runMode
                        )
                    else:
                        failReasons.append("Toil raised error: %s" % self.errorRaised)
                    self.errorRaised = None
                    options.restart = True
                else:
                    self.fail('No errors were raised on toil "%s".' % runMode)
        if failReasons:
            self.fail(
                "Test failed for ({}) reasons:\n\t{}".format(
                    len(failReasons), "\n\t".join(failReasons)
                )
            )





def passingFn(job, fileName=None):
    """
    This function is guaranteed to pass as it does nothing out of the ordinary.  If fileName is
    provided, it will be created.

    :param str fileName: The name of a file that must be created if provided.
    """
    if fileName is not None:
        # Emulates system touch.
        open(fileName, "w").close()





def failingFn(job, failType, fileName):
    """
    This function is guaranteed to fail via a raised assertion, or an os.kill

    :param job: Job
    :param str failType: 'raise' or 'kill
    :param str fileName: The name of a file that must be created.
    """
    assert failType in ("raise", "kill")
    # Use that function to avoid code redundancy
    passingFn(job, fileName)

    if failType == "raise":
        assert False
    else:
        os.kill(os.getpid(), signal.SIGKILL)