(self)
| 83 | assert x.num_processes == 3 |
| 84 | |
| 85 | def test_terminate_job(self): |
| 86 | func = Mock() |
| 87 | pool = TaskPool(10) |
| 88 | pool.on_start() |
| 89 | pool.on_apply(func) |
| 90 | |
| 91 | assert len(pool._pool_map.keys()) == 1 |
| 92 | pid = list(pool._pool_map.keys())[0] |
| 93 | greenlet = pool._pool_map[pid] |
| 94 | greenlet.link.assert_called_once() |
| 95 | |
| 96 | pool.terminate_job(pid) |
| 97 | import gevent |
| 98 | |
| 99 | gevent.kill.assert_called_once() |
| 100 | |
| 101 | def test_make_killable_target(self): |
| 102 | def valid_target(): |
nothing calls this directly
no test coverage detected