Testing multiple threads calling tasks.
(self, manager)
| 111 | @flaky |
| 112 | @pytest.mark.skip(reason="Broken test") |
| 113 | def test_multithread_producer(self, manager): |
| 114 | """Testing multiple threads calling tasks.""" |
| 115 | set_multiprocessing_start_method() |
| 116 | |
| 117 | from multiprocessing.pool import ThreadPool |
| 118 | pool = ThreadPool(20) |
| 119 | ret = pool.map(_producer, range(120)) |
| 120 | assert list(ret) == list(range(120)) |
| 121 | |
| 122 | @flaky |
| 123 | def test_ignore_result(self, manager): |
nothing calls this directly
no test coverage detected