<commit_before># Copyright (C) 2015-2016 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.

from __future__ import absolute_import
from __future__ import print_function

from builtins import str
from builtins import range
from builtins import object
import filecmp
from abc import abstractmethod, ABCMeta
from struct import pack, unpack
from uuid import uuid4

from toil.job import Job
from toil.fileStores import FileID
from toil.fileStores.cachingFileStore import IllegalDeletionCacheError, CacheUnbalancedError, CachingFileStore
from toil.test import ToilTest, needs_aws, needs_azure, needs_google, slow, travis_test
from toil.leader import FailedJobsException
from toil.jobStores.abstractJobStore import NoSuchFileException
from toil.realtimeLogger import RealtimeLogger

import collections
import inspect
import logging
import os
import random
import signal
import sys
import time
import pytest
import subprocess

# Python 3 compatibility imports
from six.moves import xrange
from future.utils import with_metaclass

if sys.version_info[0] < 3:
    # Define a usable FileNotFoundError as will be raised by os.remove on a
    # nonexistent file.
    FileNotFoundError = OSError

# Some tests take too long on the AWS and Azure Job stores and are unquitable for CI.  They can be
# be run during manual tests by setting this to False.
testingIsAutomatic = True

logger = logging.getLogger(__name__)


class hidden(object):
    """
    Hiding the abstract test classes from the Unittest loader so it can be inherited in different
    test suites for the different job stores.
    """
    class AbstractFileStoreTest(with_metaclass(ABCMeta, ToilTest)):
        """
        An abstract base class for testing the various general functions described in
        :class:toil.fileStores.abstractFileStore.AbstractFileStore
        """
        # This is overwritten in the inheriting classs
        jobStoreType = None

        def _getTestJobStore(self):
            if self.jobStoreType == 'file':
                return self._getTestJobStorePath()
            elif self.jobStoreType == 'aws':
                return 'aws:%s:cache-tests-%s' % (self.awsRegion(), uuid4())
            elif self.jobStoreType == 'azure':
                accountName = os.getenv('TOIL_AZURE_KEYNAME')
                return 'azure:%s:cache-tests-%s' % (accountName, str(uuid4()))
            elif self.jobStoreType == 'google':
                projectID = os.getenv('TOIL_GOOGLE_PROJECTID')
                return 'google:%s:cache-tests-%s' % (projectID, str(uuid4()))
            else:
                raise RuntimeError('Illegal job store type.')

        def setUp(self):
            super(hidden.AbstractFileStoreTest, self).setUp()
            testDir = self._createTempDir()
            self.options = Job.Runner.getDefaultOptions(self._getTestJobStore())
            self.options.logLevel = 'DEBUG'
            self.options.realTimeLogging = True
            self.options.workDir = testDir
            self.options.clean = 'always'
            self.options.logFile = os.path.join(testDir, 'logFile')

        @staticmethod
        def _uselessFunc(job):
            """
            I do nothing.  Don't judge me.
            """
            return None

        # Sanity test
        @travis_test
        def testToilIsNotBroken(self):
            """
            Runs a simple DAG to test if if any features other that caching were broken.
            """
            A = Job.wrapJobFn(self._uselessFunc)
            B = Job.wrapJobFn(self._uselessFunc)
            C = Job.wrapJobFn(self._uselessFunc)
            D = Job.wrapJobFn(self._uselessFunc)
            A.addChild(B)
            A.addChild(C)
            B.addChild(D)
            C.addChild(D)
            Job.Runner.startToil(A, self.options)

        # Test filestore operations.  This is a slightly less intense version of the cache specific
        # test `testReturnFileSizes`
        @slow
        def testFileStoreOperations(self):
            """
            Write a couple of files to the jobstore.  Delete a couple of them.  Read back written
            and locally deleted files.
            """
            workdir = self._createTempDir(purpose='nonLocalDir')
            F = Job.wrapJobFn(self._testFileStoreOperations,
                              nonLocalDir=workdir,
                              numIters=30, disk='2G')
            Job.Runner.startToil(F, self.options)

        @staticmethod
        def _testFileStoreOperations(job, nonLocalDir, numIters=100):
            """
            Aux function for testFileStoreOperations Conduct numIters operations.
            """
            work_dir = job.fileStore.getLocalTempDir()
            writtenFiles = {}  # fsID: (size, isLocal)
            localFileIDs = set()
            # Add one file for the sake of having something in the job store
            writeFileSize = random.randint(0, 30)
            cls = hidden.AbstractNonCachingFileStoreTest
            fsId, _ = cls._writeFileToJobStore(job, isLocalFile=True, nonLocalDir=nonLocalDir,
                                               fileMB=writeFileSize)
            writtenFiles[fsId] = writeFileSize
            localFileIDs.add(list(writtenFiles.keys())[0])
            i = 0
            while i <= numIters:
                randVal = random.random()
                if randVal < 0.33:  # Write
                    writeFileSize = random.randint(0, 30)
                    isLocalFile = True if random.random() <= 0.5 else False
                    fsID, _ = cls._writeFileToJobStore(job, isLocalFile=isLocalFile,
                                                       nonLocalDir=nonLocalDir,
                                                       fileMB=writeFileSize)
                    writtenFiles[fsID] = writeFileSize
                    localFileIDs.add(fsID)
                else:
                    if len(writtenFiles) == 0:
                        continue
                    else:
                        fsID, rdelFileSize = random.choice(list(writtenFiles.items()))
                        rdelRandVal = random.random()
                    if randVal < 0.66:  # Read
                        mutable = True if random.random() <= 0.5 else False
                        cache = True if random.random() <= 0.5 else False
                        job.fileStore.readGlobalFile(fsID, '/'.join([work_dir, str(uuid4())]),
                                                     cache=cache, mutable=mutable)
                        localFileIDs.add(fsID)
                    else:  # Delete
                        if rdelRandVal <= 0.5:  # Local Delete
                            if fsID not in localFileIDs:
                                continue
                            job.fileStore.deleteLocalFile(fsID)
                        else:  # Global Delete
                            job.fileStore.deleteGlobalFile(fsID)
                            writtenFiles.pop(fsID)
                        if fsID in localFileIDs:
                            localFileIDs.remove(fsID)
                i += 1

        @staticmethod
        def _writeFileToJobStore(job, isLocalFile, nonLocalDir=None, fileMB=1):
            """
            This function creates a file and writes it to the jobstore.

            :param bool isLocalFile: Is the file local(T) or Non-Local(F)?
            :param str nonLocalDir: A dir to write the file to.  If unspecified, a local directory
                                    is created.
            :param int fileMB: Size of the created file in MB
            """
            if isLocalFile:
                work_dir = job.fileStore.getLocalTempDir()
            else:
                assert nonLocalDir is not None
                work_dir = nonLocalDir
            with open(os.path.join(work_dir, str(uuid4())), 'wb') as testFile:
                testFile.write(os.urandom(fileMB * 1024 * 1024))

            return job.fileStore.writeGlobalFile(testFile.name), testFile

    class AbstractNonCachingFileStoreTest(with_metaclass(ABCMeta, AbstractFileStoreTest)):
        """
        Abstract tests for the the various functions in
        :class:toil.fileStores.nonCachingFileStore.NonCachingFileStore. These
        tests are general enough that they can also be used for
        :class:toil.fileStores.CachingFileStore.
        """

        def setUp(self):
            super(hidden.AbstractNonCachingFileStoreTest, self).setUp()
            self.options.disableCaching = True

    class AbstractCachingFileStoreTest(with_metaclass(ABCMeta, AbstractFileStoreTest)):
        """
        Abstract tests for the the various cache-related functions in
        :class:toil.fileStores.cachingFileStore.CachingFileStore.
        """

        def setUp(self):
            super(hidden.AbstractCachingFileStoreTest, self).setUp()
            self.options.disableCaching = False

        @slow
        def testExtremeCacheSetup(self):
            """
            Try to create the cache with bad worker active and then have 10 child jobs try to run in
            the chain.  This tests whether the cache is created properly even when the job crashes
            randomly.
            """
            if testingIsAutomatic and self.jobStoreType != 'file':
                self.skipTest("To save time")
            self.options.retryCount = 20
            self.options.badWorker = 0.5
            self.options.badWorkerFailInterval = 0.1
            for test in range(0, 20):
                E = Job.wrapJobFn(self._uselessFunc)
                F = Job.wrapJobFn(self._uselessFunc)
                jobs = {}
                for i in range(0, 10):
                    jobs[i] = Job.wrapJobFn(self._uselessFunc)
                    E.addChild(jobs[i])
                    jobs[i].addChild(F)
                Job.Runner.startToil(E, self.options)

        @slow
        def testCacheEvictionPartialEvict(self):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests 10MB of disk requiring eviction of the 1st file.  Ensure that the
            behavior is as expected.
            """
            self._testValidityOfCacheEvictTest()

            # Explicitly set clean to always so even the failed cases get cleaned (This will
            # overwrite the value set in setUp if it is ever changed in the future)
            self.options.clean = 'always'

            self._testCacheEviction(file1MB=20, file2MB=30, diskRequestMB=10)

        @slow
        def testCacheEvictionTotalEvict(self):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests 10MB of disk requiring eviction of the 1st file.  Ensure that the
            behavior is as expected.
            """
            self._testValidityOfCacheEvictTest()

            # Explicitly set clean to always so even the failed cases get cleaned (This will
            # overwrite the value set in setUp if it is ever changed in the future)
            self.options.clean = 'always'

            self._testCacheEviction(file1MB=20, file2MB=30, diskRequestMB=30)

        @slow
        def testCacheEvictionFailCase(self):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests 10MB of disk requiring eviction of the 1st file.  Ensure that the
            behavior is as expected.
            """
            self._testValidityOfCacheEvictTest()

            # Explicitly set clean to always so even the failed cases get cleaned (This will
            # overwrite the value set in setUp if it is ever changed in the future)
            self.options.clean = 'always'

            self._testCacheEviction(file1MB=20, file2MB=30, diskRequestMB=60)

        def _testValidityOfCacheEvictTest(self):
            # If the job store and cache are on the same file system, file sizes are accounted for
            # by the job store and are not reflected in the cache hence this test is redundant.
            if not self.options.jobStore.startswith(('aws', 'azure', 'google')):
                workDirDev = os.stat(self.options.workDir).st_dev
                jobStoreDev = os.stat(os.path.dirname(self.options.jobStore)).st_dev
                if workDirDev == jobStoreDev:
                    self.skipTest('Job store and working directory are on the same filesystem.')

        def _testCacheEviction(self, file1MB, file2MB, diskRequestMB):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests either 10, 30 or 60MB -- requiring eviction of 1 file, both files,
            or results in an error due to lack of space, respectively.  Ensure that the behavior is
            as expected.
            """
            self.options.retryCount = 0
            if diskRequestMB > 50:
                # This can be non int as it will never reach _probeJobReqs
                expectedResult = 'Fail'<commit_msg>Make all cahcing file store/file job store tests run locally and pass<commit_after># Copyright (C) 2015-2016 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.

from __future__ import absolute_import
from __future__ import print_function

from builtins import str
from builtins import range
from builtins import object
import filecmp
from abc import abstractmethod, ABCMeta
from struct import pack, unpack
from uuid import uuid4

from toil.job import Job
from toil.fileStores import FileID
from toil.fileStores.cachingFileStore import IllegalDeletionCacheError, CacheUnbalancedError, CachingFileStore
from toil.test import ToilTest, needs_aws, needs_azure, needs_google, slow, travis_test
from toil.leader import FailedJobsException
from toil.jobStores.abstractJobStore import NoSuchFileException
from toil.realtimeLogger import RealtimeLogger

import collections
import inspect
import logging
import os
import random
import signal
import sys
import time
import pytest
import subprocess

# Python 3 compatibility imports
from six.moves import xrange
from future.utils import with_metaclass

if sys.version_info[0] < 3:
    # Define a usable FileNotFoundError as will be raised by os.remove on a
    # nonexistent file.
    FileNotFoundError = OSError

# Some tests take too long on the AWS and Azure Job stores and are unquitable for CI.  They can be
# be run during manual tests by setting this to False.
testingIsAutomatic = True

logger = logging.getLogger(__name__)


class hidden(object):
    """
    Hiding the abstract test classes from the Unittest loader so it can be inherited in different
    test suites for the different job stores.
    """
    class AbstractFileStoreTest(with_metaclass(ABCMeta, ToilTest)):
        """
        An abstract base class for testing the various general functions described in
        :class:toil.fileStores.abstractFileStore.AbstractFileStore
        """
        # This is overwritten in the inheriting classs
        jobStoreType = None

        def _getTestJobStore(self):
            if self.jobStoreType == 'file':
                return self._getTestJobStorePath()
            elif self.jobStoreType == 'aws':
                return 'aws:%s:cache-tests-%s' % (self.awsRegion(), uuid4())
            elif self.jobStoreType == 'azure':
                accountName = os.getenv('TOIL_AZURE_KEYNAME')
                return 'azure:%s:cache-tests-%s' % (accountName, str(uuid4()))
            elif self.jobStoreType == 'google':
                projectID = os.getenv('TOIL_GOOGLE_PROJECTID')
                return 'google:%s:cache-tests-%s' % (projectID, str(uuid4()))
            else:
                raise RuntimeError('Illegal job store type.')

        def setUp(self):
            super(hidden.AbstractFileStoreTest, self).setUp()
            testDir = self._createTempDir()
            self.options = Job.Runner.getDefaultOptions(self._getTestJobStore())
            self.options.logLevel = 'DEBUG'
            self.options.realTimeLogging = True
            self.options.workDir = testDir
            self.options.clean = 'always'
            self.options.logFile = os.path.join(testDir, 'logFile')

        @staticmethod
        def _uselessFunc(job):
            """
            I do nothing.  Don't judge me.
            """
            return None

        # Sanity test
        @travis_test
        def testToilIsNotBroken(self):
            """
            Runs a simple DAG to test if if any features other that caching were broken.
            """
            A = Job.wrapJobFn(self._uselessFunc)
            B = Job.wrapJobFn(self._uselessFunc)
            C = Job.wrapJobFn(self._uselessFunc)
            D = Job.wrapJobFn(self._uselessFunc)
            A.addChild(B)
            A.addChild(C)
            B.addChild(D)
            C.addChild(D)
            Job.Runner.startToil(A, self.options)

        # Test filestore operations.  This is a slightly less intense version of the cache specific
        # test `testReturnFileSizes`
        @slow
        def testFileStoreOperations(self):
            """
            Write a couple of files to the jobstore.  Delete a couple of them.  Read back written
            and locally deleted files.
            """
            workdir = self._createTempDir(purpose='nonLocalDir')
            F = Job.wrapJobFn(self._testFileStoreOperations,
                              nonLocalDir=workdir,
                              numIters=30, disk='2G')
            Job.Runner.startToil(F, self.options)

        @staticmethod
        def _testFileStoreOperations(job, nonLocalDir, numIters=100):
            """
            Aux function for testFileStoreOperations Conduct numIters operations.
            """
            work_dir = job.fileStore.getLocalTempDir()
            writtenFiles = {}  # fsID: (size, isLocal)
            localFileIDs = set()
            # Add one file for the sake of having something in the job store
            writeFileSize = random.randint(0, 30)
            cls = hidden.AbstractNonCachingFileStoreTest
            fsId, _ = cls._writeFileToJobStore(job, isLocalFile=True, nonLocalDir=nonLocalDir,
                                               fileMB=writeFileSize)
            writtenFiles[fsId] = writeFileSize
            localFileIDs.add(list(writtenFiles.keys())[0])
            i = 0
            while i <= numIters:
                randVal = random.random()
                if randVal < 0.33:  # Write
                    writeFileSize = random.randint(0, 30)
                    isLocalFile = True if random.random() <= 0.5 else False
                    fsID, _ = cls._writeFileToJobStore(job, isLocalFile=isLocalFile,
                                                       nonLocalDir=nonLocalDir,
                                                       fileMB=writeFileSize)
                    writtenFiles[fsID] = writeFileSize
                    localFileIDs.add(fsID)
                else:
                    if len(writtenFiles) == 0:
                        continue
                    else:
                        fsID, rdelFileSize = random.choice(list(writtenFiles.items()))
                        rdelRandVal = random.random()
                    if randVal < 0.66:  # Read
                        mutable = True if random.random() <= 0.5 else False
                        cache = True if random.random() <= 0.5 else False
                        job.fileStore.readGlobalFile(fsID, '/'.join([work_dir, str(uuid4())]),
                                                     cache=cache, mutable=mutable)
                        localFileIDs.add(fsID)
                    else:  # Delete
                        if rdelRandVal <= 0.5:  # Local Delete
                            if fsID not in localFileIDs:
                                continue
                            job.fileStore.deleteLocalFile(fsID)
                        else:  # Global Delete
                            job.fileStore.deleteGlobalFile(fsID)
                            writtenFiles.pop(fsID)
                        if fsID in localFileIDs:
                            localFileIDs.remove(fsID)
                i += 1

        @staticmethod
        def _writeFileToJobStore(job, isLocalFile, nonLocalDir=None, fileMB=1):
            """
            This function creates a file and writes it to the jobstore.

            :param bool isLocalFile: Is the file local(T) or Non-Local(F)?
            :param str nonLocalDir: A dir to write the file to.  If unspecified, a local directory
                                    is created.
            :param int fileMB: Size of the created file in MB
            """
            if isLocalFile:
                work_dir = job.fileStore.getLocalTempDir()
            else:
                assert nonLocalDir is not None
                work_dir = nonLocalDir
            with open(os.path.join(work_dir, str(uuid4())), 'wb') as testFile:
                testFile.write(os.urandom(fileMB * 1024 * 1024))

            return job.fileStore.writeGlobalFile(testFile.name), testFile

    class AbstractNonCachingFileStoreTest(with_metaclass(ABCMeta, AbstractFileStoreTest)):
        """
        Abstract tests for the the various functions in
        :class:toil.fileStores.nonCachingFileStore.NonCachingFileStore. These
        tests are general enough that they can also be used for
        :class:toil.fileStores.CachingFileStore.
        """

        def setUp(self):
            super(hidden.AbstractNonCachingFileStoreTest, self).setUp()
            self.options.disableCaching = True

    class AbstractCachingFileStoreTest(with_metaclass(ABCMeta, AbstractFileStoreTest)):
        """
        Abstract tests for the the various cache-related functions in
        :class:toil.fileStores.cachingFileStore.CachingFileStore.
        """

        def setUp(self):
            super(hidden.AbstractCachingFileStoreTest, self).setUp()
            self.options.disableCaching = False

        @slow
        def testExtremeCacheSetup(self):
            """
            Try to create the cache with bad worker active and then have 10 child jobs try to run in
            the chain.  This tests whether the cache is created properly even when the job crashes
            randomly.
            """
            if testingIsAutomatic and self.jobStoreType != 'file':
                self.skipTest("To save time")
            self.options.retryCount = 20
            self.options.badWorker = 0.5
            self.options.badWorkerFailInterval = 0.1
            for test in range(0, 20):
                E = Job.wrapJobFn(self._uselessFunc)
                F = Job.wrapJobFn(self._uselessFunc)
                jobs = {}
                for i in range(0, 10):
                    jobs[i] = Job.wrapJobFn(self._uselessFunc)
                    E.addChild(jobs[i])
                    jobs[i].addChild(F)
                Job.Runner.startToil(E, self.options)

        @slow
        def testCacheEvictionPartialEvict(self):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests 10MB of disk requiring eviction of the 1st file.  Ensure that the
            behavior is as expected.
            """
            self._testValidityOfCacheEvictTest()

            # Explicitly set clean to always so even the failed cases get cleaned (This will
            # overwrite the value set in setUp if it is ever changed in the future)
            self.options.clean = 'always'

            self._testCacheEviction(file1MB=20, file2MB=30, diskRequestMB=10)

        @slow
        def testCacheEvictionTotalEvict(self):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests 10MB of disk requiring eviction of the 1st file.  Ensure that the
            behavior is as expected.
            """
            self._testValidityOfCacheEvictTest()

            # Explicitly set clean to always so even the failed cases get cleaned (This will
            # overwrite the value set in setUp if it is ever changed in the future)
            self.options.clean = 'always'

            self._testCacheEviction(file1MB=20, file2MB=30, diskRequestMB=30)

        @slow
        def testCacheEvictionFailCase(self):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests 10MB of disk requiring eviction of the 1st file.  Ensure that the
            behavior is as expected.
            """
            self._testValidityOfCacheEvictTest()

            # Explicitly set clean to always so even the failed cases get cleaned (This will
            # overwrite the value set in setUp if it is ever changed in the future)
            self.options.clean = 'always'

            self._testCacheEviction(file1MB=20, file2MB=30, diskRequestMB=60)

        def _testValidityOfCacheEvictTest(self):
            # If the job store and cache are on the same file system, file
            # sizes are accounted for by the job store and are not reflected in
            # the cache hence this test is redundant (caching will be free).
            return
            if not self.options.jobStore.startswith(('aws', 'azure', 'google')):
                workDirDev = os.stat(self.options.workDir).st_dev
                jobStoreDev = os.stat(os.path.dirname(self.options.jobStore)).st_dev
                if workDirDev == jobStoreDev:
                    self.skipTest('Job store and working directory are on the same filesystem.')

        def _testCacheEviction(self, file1MB, file2MB, diskRequestMB):
            """
            Ensure the cache eviction happens as expected.  Two files (20MB and 30MB) are written
            sequentially into the job store in separate jobs.  The cache max is force set to 50MB.
            A Third Job requests either 10, 30 or 60MB -- requiring eviction of 1 file, both files,
            or results in an error due to lack of space, respectively.  Ensure that the behavior is
            as expected.
            """
            self.options.retryCount = 0
            if diskRequestMB > 50:
                # This can be non int as it will never reach _probeJobReqs
                expectedResult = 'Fail'
{'fraction_non_alphanumeric': 0.06869688385269122, 'fraction_numerical': 0.010127478753541077, 'mean_word_length': 1.9051537907622673, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '109192', 'n_tokens_mistral': 7482, 'n_tokens_neox': 7127, 'n_words': 2385}