@pytest.mark.skipif((sys.version_info[:2] <= (3, 3)), reason='Python 3.4+ shows chained exceptions on multiprocess')
def test_exception_handling_no_traceback(testdir):
 p1 = testdir.makepyfile('\n        from multiprocessing import Pool\n\n        def process_task(n):\n            assert n == 10\n\n        def multitask_job():\n            tasks = [1]\n            with Pool(processes=1) as pool:\n                pool.map(process_task, tasks)\n\n        def test_multitask_job():\n            multitask_job()\n    ')
 result = testdir.runpytest(p1, '--tb=long')
 result.stdout.fnmatch_lines(['====* FAILURES *====', '*multiprocessing.pool.RemoteTraceback:*', 'Traceback (most recent call last):', '*assert n == 10', 'The above exception was the direct cause of the following exception:', '> * multitask_job()'])

@ pytest . mark . skipif ( ( sys . version_info [ : 2 ] <= ( 3 , 3 ) ) , reason = 'Python 3.4+ shows chained exceptions on multiprocess' ) def test_exception_handling_no_traceback ( testdir ) : p1 = testdir . makepyfile ( '\n        from multiprocessing import Pool\n\n        def process_task(n):\n            assert n == 10\n\n        def multitask_job():\n            tasks = [1]\n            with Pool(processes=1) as pool:\n                pool.map(process_task, tasks)\n\n        def test_multitask_job():\n            multitask_job()\n    ' ) result = testdir . runpytest ( p1 , '--tb=long' ) result . stdout . fnmatch_lines ( [ '====* FAILURES *====' , '*multiprocessing.pool.RemoteTraceback:*' , 'Traceback (most recent call last):' , '*assert n == 10' , 'The above exception was the direct cause of the following exception:' , '> * multitask_job()' ] )
handle chain exceptions in tasks submitted by the multiprocess module .