<s> from __future__ import absolute_import <EOL> import logging <EOL> import os <EOL> import tempfile <EOL> import unittest <EOL> import shutil <EOL> import re <EOL> import subprocess <EOL> from bd2k . util . files import mkdir_p <EOL> from toil import toilPackageDirPath <EOL> log = logging . getLogger ( __name__ ) <EOL> class ToilTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> _tempBaseDir = None <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( ToilTest , cls ) . setUpClass ( ) <EOL> cls . _tempDirs = [ ] <EOL> tempBaseDir = os . environ . get ( '<STR_LIT>' , None ) <EOL> if tempBaseDir is not None and not os . path . isabs ( tempBaseDir ) : <EOL> tempBaseDir = os . path . abspath ( os . path . join ( cls . _projectRootPath ( ) , tempBaseDir ) ) <EOL> mkdir_p ( tempBaseDir ) <EOL> cls . _tempBaseDir = tempBaseDir <EOL> @ classmethod <EOL> def _getUtilScriptPath ( cls , script_name ) : <EOL> return os . path . join ( toilPackageDirPath ( ) , '<STR_LIT>' , script_name + '<STR_LIT>' ) <EOL> @ classmethod <EOL> def _projectRootPath ( cls ) : <EOL> """<STR_LIT>""" <EOL> assert re . search ( r'<STR_LIT>' , __file__ ) <EOL> projectRootPath = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> packageComponents = __name__ . split ( '<STR_LIT:.>' ) <EOL> expectedSuffix = os . path . join ( '<STR_LIT:src>' , * packageComponents ) <EOL> assert projectRootPath . endswith ( expectedSuffix ) <EOL> projectRootPath = projectRootPath [ : - len ( expectedSuffix ) ] <EOL> return projectRootPath <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> if cls . _tempBaseDir is None : <EOL> while cls . _tempDirs : <EOL> tempDir = cls . _tempDirs . pop ( ) <EOL> if os . path . exists ( tempDir ) : <EOL> shutil . rmtree ( tempDir ) <EOL> else : <EOL> cls . _tempDirs = [ ] <EOL> super ( ToilTest , cls ) . tearDownClass ( ) <EOL> def setUp ( self ) : <EOL> log . info ( "<STR_LIT>" , self . id ( ) ) <EOL> super ( ToilTest , self ) . setUp ( ) <EOL> def _createTempDir ( self , purpose = None ) : <EOL> prefix = [ '<STR_LIT>' , '<STR_LIT:test>' , self . id ( ) ] <EOL> if purpose : prefix . append ( purpose ) <EOL> prefix . append ( '<STR_LIT>' ) <EOL> temp_dir_path = tempfile . mkdtemp ( dir = self . _tempBaseDir , prefix = '<STR_LIT:->' . join ( prefix ) ) <EOL> self . _tempDirs . append ( temp_dir_path ) <EOL> return temp_dir_path <EOL> def tearDown ( self ) : <EOL> super ( ToilTest , self ) . tearDown ( ) <EOL> log . info ( "<STR_LIT>" , self . id ( ) ) <EOL> def _getTestJobStorePath ( self ) : <EOL> path = self . _createTempDir ( purpose = '<STR_LIT>' ) <EOL> os . rmdir ( path ) <EOL> return path <EOL> try : <EOL> from _pytest . mark import MarkDecorator <EOL> except ImportError : <EOL> def _mark_test ( name , test_item ) : <EOL> return test_item <EOL> else : <EOL> def _mark_test ( name , test_item ) : <EOL> return MarkDecorator ( name ) ( test_item ) <EOL> def needs_aws ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> import boto <EOL> except ImportError : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> dot_boto_path = os . path . expanduser ( '<STR_LIT>' ) <EOL> dot_aws_credentials_path = os . path . expanduser ( '<STR_LIT>' ) <EOL> hv_uuid_path = '<STR_LIT>' <EOL> if ( os . path . exists ( dot_boto_path ) <EOL> or os . path . exists ( dot_aws_credentials_path ) <EOL> or os . path . exists ( hv_uuid_path ) and file_begins_with ( hv_uuid_path , '<STR_LIT>' ) ) : <EOL> return test_item <EOL> else : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ( test_item ) <EOL> def file_begins_with ( path , prefix ) : <EOL> with open ( path ) as f : <EOL> return f . read ( len ( prefix ) ) == prefix <EOL> def needs_azure ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> import azure . storage <EOL> except ImportError : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> from toil . jobStores . azureJobStore import credential_file_path <EOL> full_credential_file_path = os . path . expanduser ( credential_file_path ) <EOL> if not os . path . exists ( full_credential_file_path ) : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" % credential_file_path ) ( test_item ) <EOL> return test_item <EOL> def needs_gridengine ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> with open ( os . devnull , '<STR_LIT>' ) as devnull : <EOL> subprocess . Popen ( '<STR_LIT>' , stdout = devnull , stderr = devnull , stdin = devnull ) <EOL> except OSError : <EOL> return unittest . skip ( "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> return test_item <EOL> def needs_mesos ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> import mesos . native <EOL> except ImportError : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> return test_item <EOL> def needs_parasol ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> with open ( os . devnull , '<STR_LIT>' ) as devnull : <EOL> subprocess . Popen ( '<STR_LIT>' , stdout = devnull , stderr = devnull , stdin = devnull ) <EOL> except OSError : <EOL> return unittest . skip ( "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> return test_item <EOL> def needs_slurm ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> with open ( os . devnull , '<STR_LIT>' ) as devnull : <EOL> subprocess . Popen ( '<STR_LIT>' , stdout = devnull , stderr = devnull , stdin = devnull ) <EOL> except OSError : <EOL> return unittest . skip ( "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> return test_item <EOL> def needs_encryption ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> import nacl <EOL> except ImportError : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> return test_item <EOL> def needs_cwl ( test_item ) : <EOL> """<STR_LIT>""" <EOL> test_item = _mark_test ( '<STR_LIT>' , test_item ) <EOL> try : <EOL> import cwltool <EOL> except ImportError : <EOL> return unittest . skip ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ( test_item ) <EOL> except : <EOL> raise <EOL> else : <EOL> return test_item <EOL> methodNamePartRegex = re . compile ( '<STR_LIT>' ) <EOL> def make_tests ( generalMethod , targetClass = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> def pop ( d ) : <EOL> """<STR_LIT>""" <EOL> k , v = next ( kwargs . iteritems ( ) ) <EOL> del d [ k ] <EOL> return k , v <EOL> def permuteIntoLeft ( left , rPrmName , right ) : <EOL> """<STR_LIT>""" <EOL> for prmValName , lDict in left . items ( ) : <EOL> for rValName , rVal in right . items ( ) : <EOL> nextPrmVal = ( '<STR_LIT>' % ( rPrmName , rValName . title ( ) ) ) <EOL> if methodNamePartRegex . match ( nextPrmVal ) is None : <EOL> raise RuntimeError ( "<STR_LIT>" % pvName ) <EOL> aggDict = dict ( lDict ) <EOL> aggDict [ rPrmName ] = rVal <EOL> left [ prmValName + nextPrmVal ] = aggDict <EOL> left . pop ( prmValName ) <EOL> def insertMethodToClass ( ) : <EOL> """<STR_LIT>""" <EOL> def fx ( self , prms = prms ) : <EOL> return generalMethod ( self , ** prms ) <EOL> setattr ( targetClass , '<STR_LIT>' % ( generalMethod . __name__ , prmNames ) , fx ) <EOL> left = { } <EOL> prmName , vals = pop ( kwargs ) <EOL> for valName , val in vals . items ( ) : <EOL> pvName = '<STR_LIT>' % ( prmName , valName . title ( ) ) <EOL> if methodNamePartRegex . match ( pvName ) is None : <EOL> raise RuntimeError ( "<STR_LIT>" % pvName ) <EOL> left [ pvName ] = { prmName : val } <EOL> while len ( kwargs ) > <NUM_LIT:0> : <EOL> permuteIntoLeft ( left , * pop ( kwargs ) ) <EOL> targetClass = targetClass or generalMethod . im_class <EOL> for prmNames , prms in left . items ( ) : <EOL> insertMethodToClass ( ) </s>
