from pyParallelRunAndWrite import parallelRunAndWrite as prw from time import sleep, time import os from random import randint import re rangeSize = 10 outFname = "output.txt" errFname = "err.txt" errRegex='prw arg \d+ pid \d+' def raiseOrWork(arg): if(randint(0,100)%2 ==0): raise Exception("prw arg {} pid {} ".format(arg, os.getpid())) return work(arg) def work(arg): ret=str(arg)+str(os.getpid()) sleep(randint(0,3)) return ret def verifyOutput(): ok=True with open(outFname, 'r') as f: for line in f.read(): if(re.match('\d*',line) is None) : ok=False assert ok def verifyErrFile(): found=False with open(errFname, 'r') as f: for line in f.read(): if(re.match('Exception: '+errRegex,line) is None) : found=True assert found def verifyFiles(): verifyOutput() verifyErrFile() def testErrFileDebug(): prw.parallelRunAndWrite(raiseOrWork, range(rangeSize), outFname, errFileName=errFname, debug=True, bar=False) verifyFiles() def testErrFile(): prw.parallelRunAndWrite(raiseOrWork, range(rangeSize), outFname, errFileName=errFname) verifyFiles() def testNoErrFile(): prw.parallelRunAndWrite(work, range(rangeSize), outFname) verifyOutput() def testRaise(): from pytest import raises with raises(Exception, match=r''+errRegex): prw.parallelRunAndWrite(raiseOrWork, range(rangeSize), outFname) verifyFiles()