| 31 | self.GreenletExit = self.patching('gevent.greenlet.GreenletExit') |
| 32 | |
| 33 | def test_sched(self): |
| 34 | self.greenlet.Greenlet = object |
| 35 | x = Timer() |
| 36 | self.greenlet.Greenlet = Mock() |
| 37 | x._Greenlet.spawn_later = Mock() |
| 38 | x._GreenletExit = KeyError |
| 39 | entry = Mock() |
| 40 | g = x._enter(1, 0, entry) |
| 41 | assert x.queue |
| 42 | |
| 43 | x._entry_exit(g) |
| 44 | g.kill.assert_called_with() |
| 45 | assert not x._queue |
| 46 | |
| 47 | x._queue.add(g) |
| 48 | x.clear() |
| 49 | x._queue.add(g) |
| 50 | g.kill.side_effect = KeyError() |
| 51 | x.clear() |
| 52 | |
| 53 | g = x._Greenlet() |
| 54 | g.cancel() |
| 55 | |
| 56 | |
| 57 | class test_TaskPool: |