<s> from __future__ import absolute_import <EOL> import unittest <EOL> import os <EOL> import random <EOL> from toil . lib . bioio import getTempFile <EOL> from toil . job import Job , JobGraphDeadlockException <EOL> from toil . test import ToilTest <EOL> class JobTest ( ToilTest ) : <EOL> """<STR_LIT>""" <EOL> def testStatic ( self ) : <EOL> """<STR_LIT>""" <EOL> outFile = getTempFile ( rootDir = self . _createTempDir ( ) ) <EOL> try : <EOL> A = Job . wrapFn ( fn1Test , "<STR_LIT:A>" , outFile ) <EOL> B = Job . wrapFn ( fn1Test , A . rv ( ) , outFile ) <EOL> C = Job . wrapFn ( fn1Test , B . rv ( ) , outFile ) <EOL> D = Job . wrapFn ( fn1Test , C . rv ( ) , outFile ) <EOL> E = Job . wrapFn ( fn1Test , D . rv ( ) , outFile ) <EOL> F = Job . wrapFn ( fn1Test , E . rv ( ) , outFile ) <EOL> A . addChild ( B ) <EOL> A . addChild ( C ) <EOL> B . addChild ( C ) <EOL> B . addFollowOn ( E ) <EOL> C . addFollowOn ( D ) <EOL> A . addFollowOn ( F ) <EOL> options = Job . Runner . getDefaultOptions ( self . _getTestJobStorePath ( ) ) <EOL> options . logLevel = "<STR_LIT>" <EOL> options . retryCount = <NUM_LIT:100> <EOL> options . badWorker = <NUM_LIT:0.5> <EOL> options . badWorkerFailInterval = <NUM_LIT> <EOL> Job . Runner . startToil ( A , options ) <EOL> self . assertEquals ( open ( outFile , '<STR_LIT:r>' ) . readline ( ) , "<STR_LIT>" ) <EOL> finally : <EOL> os . remove ( outFile ) <EOL> def testStatic2 ( self ) : <EOL> """<STR_LIT>""" <EOL> outFile = getTempFile ( rootDir = self . _createTempDir ( ) ) <EOL> try : <EOL> A = Job . wrapFn ( fn1Test , "<STR_LIT:A>" , outFile ) <EOL> B = Job . wrapFn ( fn1Test , A . rv ( ) , outFile ) <EOL> C = Job . wrapFn ( fn1Test , B . rv ( ) , outFile ) <EOL> D = Job . wrapFn ( fn1Test , C . rv ( ) , outFile ) <EOL> A . addChild ( B ) <EOL> A . addFollowOn ( C ) <EOL> C . addChild ( D ) <EOL> options = Job . Runner . getDefaultOptions ( self . _getTestJobStorePath ( ) ) <EOL> options . logLevel = "<STR_LIT>" <EOL> options . retryCount = <NUM_LIT:100> <EOL> options . badWorker = <NUM_LIT:0.5> <EOL> options . badWorkerFailInterval = <NUM_LIT> <EOL> Job . Runner . startToil ( A , options ) <EOL> self . assertEquals ( open ( outFile , '<STR_LIT:r>' ) . readline ( ) , "<STR_LIT>" ) <EOL> finally : <EOL> os . remove ( outFile ) <EOL> def testDeadlockDetection ( self ) : <EOL> """<STR_LIT>""" <EOL> for test in xrange ( <NUM_LIT:100> ) : <EOL> nodeNumber = random . choice ( xrange ( <NUM_LIT:2> , <NUM_LIT:20> ) ) <EOL> childEdges = self . makeRandomDAG ( nodeNumber ) <EOL> adjacencyList = self . getAdjacencyList ( nodeNumber , childEdges ) <EOL> self . assertTrue ( self . isAcyclic ( adjacencyList ) ) <EOL> followOnEdges = self . addRandomFollowOnEdges ( adjacencyList ) <EOL> self . assertTrue ( self . isAcyclic ( adjacencyList ) ) <EOL> rootJob = self . makeJobGraph ( nodeNumber , childEdges , followOnEdges , None ) <EOL> rootJob . checkJobGraphAcylic ( ) <EOL> rootJob . checkJobGraphConnected ( ) <EOL> self . assertEquals ( rootJob . getRootJobs ( ) , set ( ( rootJob , ) ) ) <EOL> childEdges2 = childEdges . copy ( ) <EOL> childEdges2 . add ( ( nodeNumber , <NUM_LIT:1> ) ) <EOL> rootJob2 = self . makeJobGraph ( nodeNumber + <NUM_LIT:1> , childEdges2 , followOnEdges , None ) <EOL> try : <EOL> rootJob2 . checkJobGraphConnected ( ) <EOL> self . assertTrue ( False ) <EOL> except JobGraphDeadlockException : <EOL> pass <EOL> def checkChildEdgeCycleDetection ( fNode , tNode ) : <EOL> childEdges . add ( ( fNode , tNode ) ) <EOL> adjacencyList [ fNode ] . add ( tNode ) <EOL> self . assertTrue ( not self . isAcyclic ( adjacencyList ) ) <EOL> try : <EOL> self . makeJobGraph ( nodeNumber , childEdges , <EOL> followOnEdges , None ) . checkJobGraphAcylic ( ) <EOL> self . assertTrue ( False ) <EOL> except JobGraphDeadlockException : <EOL> pass <EOL> childEdges . remove ( ( fNode , tNode ) ) <EOL> adjacencyList [ fNode ] . remove ( tNode ) <EOL> self . makeJobGraph ( nodeNumber , childEdges , <EOL> followOnEdges , None ) . checkJobGraphAcylic ( ) <EOL> fNode , tNode = self . getRandomEdge ( nodeNumber ) <EOL> while fNode not in self . reachable ( tNode , adjacencyList ) : <EOL> fNode , tNode = self . getRandomEdge ( nodeNumber ) <EOL> checkChildEdgeCycleDetection ( fNode , tNode ) <EOL> node = random . choice ( xrange ( nodeNumber ) ) <EOL> checkChildEdgeCycleDetection ( node , node ) <EOL> def checkFollowOnEdgeCycleDetection ( fNode , tNode ) : <EOL> followOnEdges . add ( ( fNode , tNode ) ) <EOL> try : <EOL> self . makeJobGraph ( nodeNumber , childEdges , <EOL> followOnEdges , None ) . checkJobGraphAcylic ( ) <EOL> except JobGraphDeadlockException : <EOL> pass <EOL> followOnEdges . remove ( ( fNode , tNode ) ) <EOL> self . makeJobGraph ( nodeNumber , childEdges , <EOL> followOnEdges , None ) . checkJobGraphAcylic ( ) <EOL> fNode , tNode = self . getRandomEdge ( nodeNumber ) <EOL> while fNode not in self . reachable ( tNode , adjacencyList ) : <EOL> fNode , tNode = self . getRandomEdge ( nodeNumber ) <EOL> checkFollowOnEdgeCycleDetection ( fNode , tNode ) <EOL> node = random . choice ( xrange ( nodeNumber ) ) <EOL> checkFollowOnEdgeCycleDetection ( node , node ) <EOL> fNode , tNode = self . getRandomEdge ( nodeNumber ) <EOL> if ( len ( self . reachable ( tNode , adjacencyList ) . intersection ( self . reachable ( fNode , adjacencyList ) ) ) > <NUM_LIT:0> and <EOL> ( fNode , tNode ) not in childEdges ) : <EOL> checkFollowOnEdgeCycleDetection ( fNode , tNode ) <EOL> def testEvaluatingRandomDAG ( self ) : <EOL> """<STR_LIT>""" <EOL> jobStore = self . _getTestJobStorePath ( ) <EOL> for test in xrange ( <NUM_LIT:30> ) : <EOL> tempDir = self . _createTempDir ( purpose = '<STR_LIT>' ) <EOL> nodeNumber = random . choice ( xrange ( <NUM_LIT:2> , <NUM_LIT:20> ) ) <EOL> childEdges = self . makeRandomDAG ( nodeNumber ) <EOL> adjacencyList = self . getAdjacencyList ( nodeNumber , childEdges ) <EOL> self . assertTrue ( self . isAcyclic ( adjacencyList ) ) <EOL> followOnEdges = self . addRandomFollowOnEdges ( adjacencyList ) <EOL> self . assertTrue ( self . isAcyclic ( adjacencyList ) ) <EOL> rootJob = self . makeJobGraph ( nodeNumber , childEdges , followOnEdges , tempDir ) <EOL> options = Job . Runner . getDefaultOptions ( "<STR_LIT>" % ( jobStore , test ) ) <EOL> options . retryCount = <NUM_LIT:100> <EOL> options . badWorker = <NUM_LIT:0.5> <EOL> options . badWorkerFailInterval = <NUM_LIT> <EOL> Job . Runner . startToil ( rootJob , options ) <EOL> for i in xrange ( nodeNumber ) : <EOL> with open ( os . path . join ( tempDir , str ( i ) ) , '<STR_LIT:r>' ) as fH : <EOL> ordering = map ( int , fH . readline ( ) . split ( ) ) <EOL> self . assertEquals ( int ( ordering [ - <NUM_LIT:1> ] ) , i ) <EOL> for j in ordering [ : - <NUM_LIT:1> ] : <EOL> adjacencyList [ int ( j ) ] . add ( int ( ordering [ - <NUM_LIT:1> ] ) ) <EOL> if not self . isAcyclic ( adjacencyList ) : <EOL> print "<STR_LIT>" , ordering <EOL> print "<STR_LIT>" , childEdges <EOL> print "<STR_LIT>" , followOnEdges <EOL> print "<STR_LIT>" , adjacencyList <EOL> self . assertTrue ( self . isAcyclic ( adjacencyList ) ) <EOL> @ staticmethod <EOL> def getRandomEdge ( nodeNumber ) : <EOL> assert nodeNumber > <NUM_LIT:1> <EOL> fNode = random . choice ( xrange ( nodeNumber - <NUM_LIT:1> ) ) <EOL> return ( fNode , random . choice ( xrange ( fNode , nodeNumber ) ) ) <EOL> @ staticmethod <EOL> def makeRandomDAG ( nodeNumber ) : <EOL> """<STR_LIT>""" <EOL> edgeNumber = random . choice ( xrange ( nodeNumber - <NUM_LIT:1> , <NUM_LIT:1> + ( nodeNumber * ( nodeNumber - <NUM_LIT:1> ) ) / <NUM_LIT:2> ) ) <EOL> edges = set ( map ( lambda i : ( random . choice ( xrange ( i ) ) , i ) , xrange ( <NUM_LIT:1> , nodeNumber ) ) ) <EOL> while edgeNumber < len ( edges ) : <EOL> edges . add ( JobTest . getRandomEdge ( nodeNumber ) ) <EOL> return edges <EOL> @ staticmethod <EOL> def getAdjacencyList ( nodeNumber , edges ) : <EOL> """<STR_LIT>""" <EOL> adjacencyList = [ set ( ) for i in xrange ( nodeNumber ) ] <EOL> for fNode , tNode in edges : <EOL> adjacencyList [ fNode ] . add ( tNode ) <EOL> return adjacencyList <EOL> @ staticmethod <EOL> def reachable ( node , adjacencyList , followOnAdjacencyList = None ) : <EOL> """<STR_LIT>""" <EOL> visited = set ( ) <EOL> def dfs ( fNode ) : <EOL> if fNode not in visited : <EOL> visited . add ( fNode ) <EOL> for tNode in adjacencyList [ fNode ] : <EOL> dfs ( tNode ) <EOL> if followOnAdjacencyList != None : <EOL> for tNode in followOnAdjacencyList [ fNode ] : <EOL> dfs ( tNode ) <EOL> dfs ( node ) <EOL> return visited <EOL> @ staticmethod <EOL> def addRandomFollowOnEdges ( childAdjacencyList ) : <EOL> """<STR_LIT>""" <EOL> def makeAugmentedAdjacencyList ( ) : <EOL> augmentedAdjacencyList = map ( lambda i : childAdjacencyList [ i ] . union ( followOnAdjacencyList [ i ] ) , <EOL> range ( len ( childAdjacencyList ) ) ) <EOL> def addImpliedEdges ( node , followOnEdges ) : <EOL> visited = set ( ) <EOL> def f ( node ) : <EOL> if node not in visited : <EOL> visited . add ( node ) <EOL> for i in followOnEdges : <EOL> augmentedAdjacencyList [ node ] . add ( i ) <EOL> map ( f , childAdjacencyList [ node ] ) <EOL> map ( f , followOnAdjacencyList [ node ] ) <EOL> map ( f , childAdjacencyList [ node ] ) <EOL> for node in xrange ( len ( followOnAdjacencyList ) ) : <EOL> addImpliedEdges ( node , followOnAdjacencyList [ node ] ) <EOL> return augmentedAdjacencyList <EOL> followOnEdges = set ( ) <EOL> followOnAdjacencyList = map ( lambda i : set ( ) , childAdjacencyList ) <EOL> while random . random ( ) > <NUM_LIT> : <EOL> fNode , tNode = JobTest . getRandomEdge ( len ( childAdjacencyList ) ) <EOL> fDescendants = reduce ( lambda i , j : i . union ( j ) , map ( lambda c : <EOL> JobTest . reachable ( c , childAdjacencyList , <EOL> followOnAdjacencyList ) , <EOL> childAdjacencyList [ fNode ] ) , set ( ) ) <EOL> fDescendants . add ( fNode ) <EOL> followOnAdjacencyList [ fNode ] . add ( tNode ) <EOL> augmentedAdjacencyList = makeAugmentedAdjacencyList ( ) <EOL> if JobTest . isAcyclic ( augmentedAdjacencyList ) : <EOL> followOnEdges . add ( ( fNode , tNode ) ) <EOL> else : <EOL> followOnAdjacencyList [ fNode ] . remove ( tNode ) <EOL> childAdjacencyList [ : ] = makeAugmentedAdjacencyList ( ) [ : ] <EOL> return followOnEdges <EOL> @ staticmethod <EOL> def makeJobGraph ( nodeNumber , childEdges , followOnEdges , outPath ) : <EOL> """<STR_LIT>""" <EOL> jobsToPromisesMap = { } <EOL> def makeJob ( string ) : <EOL> promises = [ ] <EOL> job = Job . wrapFn ( fn2Test , promises , string , <EOL> os . path . join ( outPath , string ) if outPath != None else None ) <EOL> jobsToPromisesMap [ job ] = promises <EOL> return job <EOL> jobs = map ( lambda i : makeJob ( str ( i ) ) , xrange ( nodeNumber ) ) <EOL> for fNode , tNode in childEdges : <EOL> jobs [ fNode ] . addChild ( jobs [ tNode ] ) <EOL> for fNode , tNode in followOnEdges : <EOL> jobs [ fNode ] . addFollowOn ( jobs [ tNode ] ) <EOL> def getRandomPredecessor ( job ) : <EOL> predecessor = random . choice ( list ( job . _directPredecessors ) ) <EOL> while random . random ( ) > <NUM_LIT:0.5> and len ( predecessor . _directPredecessors ) > <NUM_LIT:0> : <EOL> predecessor = random . choice ( list ( predecessor . _directPredecessors ) ) <EOL> return predecessor <EOL> while random . random ( ) > <NUM_LIT> : <EOL> job = random . choice ( jobsToPromisesMap . keys ( ) ) <EOL> promises = jobsToPromisesMap [ job ] <EOL> if len ( job . _directPredecessors ) > <NUM_LIT:0> : <EOL> predecessor = getRandomPredecessor ( job ) <EOL> promises . append ( predecessor . rv ( ) ) <EOL> return jobs [ <NUM_LIT:0> ] <EOL> @ staticmethod <EOL> def isAcyclic ( adjacencyList ) : <EOL> """<STR_LIT>""" <EOL> def cyclic ( fNode , visited , stack ) : <EOL> if fNode not in visited : <EOL> visited . add ( fNode ) <EOL> assert fNode not in stack <EOL> stack . append ( fNode ) <EOL> for tNode in adjacencyList [ fNode ] : <EOL> if cyclic ( tNode , visited , stack ) : <EOL> return True <EOL> assert stack . pop ( ) == fNode <EOL> return fNode in stack <EOL> visited = set ( ) <EOL> for i in xrange ( len ( adjacencyList ) ) : <EOL> if cyclic ( i , visited , [ ] ) : <EOL> return False <EOL> return True <EOL> def fn1Test ( string , outputFile , promises = [ ] ) : <EOL> """<STR_LIT>""" <EOL> rV = string + chr ( ord ( string [ - <NUM_LIT:1> ] ) + <NUM_LIT:1> ) <EOL> with open ( outputFile , '<STR_LIT:w>' ) as fH : <EOL> fH . write ( rV ) <EOL> return rV <EOL> def fn2Test ( pStrings , s , outputFile ) : <EOL> """<STR_LIT>""" <EOL> with open ( outputFile , '<STR_LIT:w>' ) as fH : <EOL> fH . write ( "<STR_LIT:U+0020>" . join ( pStrings ) + "<STR_LIT:U+0020>" + s ) <EOL> return s <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
