Handle chain exceptions in tasks submitted by the multiprocess module (#1984).
(pytester: Pytester)
| 1435 | |
| 1436 | |
| 1437 | def test_exception_handling_no_traceback(pytester: Pytester) -> None: |
| 1438 | """Handle chain exceptions in tasks submitted by the multiprocess module (#1984).""" |
| 1439 | p1 = pytester.makepyfile( |
| 1440 | """ |
| 1441 | from multiprocessing import Pool |
| 1442 | |
| 1443 | def process_task(n): |
| 1444 | assert n == 10 |
| 1445 | |
| 1446 | def multitask_job(): |
| 1447 | tasks = [1] |
| 1448 | with Pool(processes=1) as pool: |
| 1449 | pool.map(process_task, tasks) |
| 1450 | |
| 1451 | def test_multitask_job(): |
| 1452 | multitask_job() |
| 1453 | """ |
| 1454 | ) |
| 1455 | pytester.syspathinsert() |
| 1456 | result = pytester.runpytest(p1, "--tb=long") |
| 1457 | result.stdout.fnmatch_lines( |
| 1458 | [ |
| 1459 | "====* FAILURES *====", |
| 1460 | "*multiprocessing.pool.RemoteTraceback:*", |
| 1461 | "Traceback (most recent call last):", |
| 1462 | "*assert n == 10", |
| 1463 | "The above exception was the direct cause of the following exception:", |
| 1464 | "> * multitask_job()", |
| 1465 | ] |
| 1466 | ) |
| 1467 | |
| 1468 | |
| 1469 | @pytest.mark.skipif("'__pypy__' in sys.builtin_module_names") |
nothing calls this directly
no test coverage detected