<s> """<STR_LIT>""" <EOL> import os <EOL> import json <EOL> import time <EOL> import inspect <EOL> TEST_DATA_LOCATION_ENVIRONMENT_VARIABLE = "<STR_LIT>" <EOL> class TimedOutException ( Exception ) : <EOL> pass <EOL> def isCurrentlyTesting ( ) : <EOL> return os . getenv ( TEST_DATA_LOCATION_ENVIRONMENT_VARIABLE ) is not None <EOL> def recordTest ( testName , elapsedTime , metadata , ** kwargs ) : <EOL> if not ( isinstance ( elapsedTime , float ) or elapsedTime is None ) : <EOL> raise UserWarning ( <EOL> "<STR_LIT>" <EOL> ) <EOL> if not isCurrentlyTesting ( ) : <EOL> raise UserWarning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" + TEST_DATA_LOCATION_ENVIRONMENT_VARIABLE + <EOL> "<STR_LIT>" ) <EOL> targetPath = os . getenv ( TEST_DATA_LOCATION_ENVIRONMENT_VARIABLE ) <EOL> if metadata is None : <EOL> metadata = { } <EOL> if kwargs is not None : <EOL> metadata . update ( kwargs ) <EOL> perfLogEntry = { <EOL> "<STR_LIT:name>" : str ( testName ) , <EOL> "<STR_LIT:time>" : elapsedTime if isinstance ( elapsedTime , float ) else None , <EOL> "<STR_LIT>" : metadata <EOL> } <EOL> if targetPath == '<STR_LIT:->' : <EOL> print "<STR_LIT>" % ( elapsedTime , testName ) <EOL> else : <EOL> with open ( targetPath , "<STR_LIT>" ) as f : <EOL> f . write ( json . dumps ( perfLogEntry ) + "<STR_LIT:\n>" ) <EOL> def recordThroughputTest ( testName , runtime , n , baseMultiplier , metadata ) : <EOL> recordTest ( testName , <EOL> runtime / n * baseMultiplier , <EOL> metadata , <EOL> n = n , <EOL> baseMultiplier = baseMultiplier , <EOL> actualTime = runtime ) <EOL> def testThroughput ( testName , <EOL> testFunOfN , <EOL> setupFunOfN = None , <EOL> transformOfN = None , <EOL> metadata = None , <EOL> maxNToSearch = <NUM_LIT> , <EOL> baseMultiplier = <NUM_LIT:1> , <EOL> timeoutInSec = <NUM_LIT> ) : <EOL> counter = <NUM_LIT:0> <EOL> unitOfWork = counter <EOL> unitsOfWorkCompleted = <NUM_LIT:0> <EOL> runtime = None <EOL> timeUsed = <NUM_LIT:0> <EOL> while timeUsed < timeoutInSec and counter <= maxNToSearch : <EOL> try : <EOL> counter += <NUM_LIT:1> <EOL> unitOfWork = counter <EOL> if transformOfN is not None : <EOL> unitOfWork = transformOfN ( counter ) <EOL> if setupFunOfN is not None : <EOL> setupFunOfN ( unitOfWork ) <EOL> t0 = time . time ( ) <EOL> testFunOfN ( unitOfWork ) <EOL> runtime = time . time ( ) - t0 <EOL> if timeUsed < timeoutInSec : <EOL> unitsOfWorkCompleted += unitOfWork <EOL> timeUsed += runtime <EOL> except TimedOutException : <EOL> break <EOL> assert runtime is not None <EOL> assert unitsOfWorkCompleted > <NUM_LIT:0> <EOL> if isCurrentlyTesting ( ) : <EOL> recordThroughputTest ( testName , timeUsed , unitsOfWorkCompleted , baseMultiplier , metadata ) <EOL> def loadTestsFromFile ( testFileName ) : <EOL> with open ( testFileName , "<STR_LIT:rb>" ) as f : <EOL> return [ json . loads ( x ) for x in f . readlines ( ) ] <EOL> def PerfTest ( testName ) : <EOL> """<STR_LIT>""" <EOL> def decorator ( f ) : <EOL> meta = { <EOL> '<STR_LIT:file>' : "<STR_LIT:/>" . join ( inspect . getmodule ( f ) . __name__ . split ( "<STR_LIT:.>" ) ) + "<STR_LIT>" , <EOL> '<STR_LIT>' : inspect . getsourcelines ( f ) [ <NUM_LIT:1> ] <EOL> } <EOL> def innerTestFun ( * args , ** kwds ) : <EOL> t0 = time . time ( ) <EOL> try : <EOL> result = f ( * args , ** kwds ) <EOL> except : <EOL> if isCurrentlyTesting ( ) : <EOL> recordTest ( testName , None , meta ) <EOL> raise <EOL> if isCurrentlyTesting ( ) : <EOL> recordTest ( testName , time . time ( ) - t0 , meta ) <EOL> return result <EOL> innerTestFun . __name__ = f . __name__ <EOL> return innerTestFun <EOL> return decorator <EOL> def getCurrentStackframeFileAndLine ( framesAbove ) : <EOL> curStack = inspect . currentframe ( ) <EOL> above = inspect . getouterframes ( curStack ) <EOL> twoAbove = above [ framesAbove ] [ <NUM_LIT:0> ] <EOL> return { <EOL> '<STR_LIT:file>' : "<STR_LIT:/>" . join ( inspect . getmodule ( twoAbove ) . __name__ . split ( "<STR_LIT:.>" ) ) + "<STR_LIT>" , <EOL> '<STR_LIT>' : inspect . getsourcelines ( twoAbove ) [ <NUM_LIT:1> ] <EOL> } <EOL> class RecordAsPerfTest : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> self . t0 = None <EOL> def __enter__ ( self , * args , ** kwds ) : <EOL> self . t0 = time . time ( ) <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> if type is None : <EOL> recordTest ( self . name , time . time ( ) - self . t0 , { } ) </s>
