I am trying to test a function that might crash, using multiprocessing.Process(). The code written below restarts the call if the function doesn't respond ideally within 10 minutes which is working completely fine, even the self.assert...
tests are working fine. The problem comes out to be with the coverage, for some reason, running coverage over my files shows that my_func
is being called but nothing inside it is getting covered while testing whereas the tests (which require the whole function to be executed without crashing) are passing.
while True:
p = multiprocessing.Process(
target=my_func, args=(return_dict) # return_dict to get the return values
)
p.start()
p.join(600)
if p.is_alive():
print(
"Taking too long, "
+ "KILLING IT and starting AGAIN."
)
p.kill()
p.join()
else:
break
Is there something that I am doing wrong? Please let me know if the information seems incomplete, I'll add my tests and the function then.
For coverage to cover code running in sub-processes you need to actually specify it.
Add import coverage
import statement in your file
Add coverage.process_startup()
on top of every multiprocessing.Process()
statement, in my case something like this -
coverage.process_startup()
p = multiprocessing.Process( target=my_func, args=(return_dict) )
....rest of the code
Create a .coveragerc
file in your root directory and add the following lines to specify covering subprocesses and multiprocessing.
[run]
concurrency=multiprocessing
Set a path variable called COVERAGE_PROCESS_START
to store the location of .coveragerc
file
set COVERAGE_PROCESS_START=%cd%\.coveragerc // Windows
export COVERAGE_PROCESS_START=$PWD/.coveragerc // UNIX/LINUX
You're all set, run the tests using coverage run -m unittest
, notice how now there are more than 1 coverage report files generated. To combine them run coverage combine
and finally for the report run coverage report