55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
import aioeventlet
|
|
import greenlet
|
|
import tests
|
|
|
|
|
|
class WrapGreenletTests(tests.TestCase):
|
|
def test_wrap_greenlet(self):
|
|
def func(value):
|
|
return value * 3
|
|
|
|
gl = greenlet.greenlet(func)
|
|
fut = aioeventlet.wrap_greenthread(gl)
|
|
gl.switch(5)
|
|
result = self.loop.run_until_complete(fut)
|
|
self.assertEqual(result, 15)
|
|
|
|
def test_wrap_greenlet_exc(self):
|
|
def func():
|
|
raise ValueError(7)
|
|
|
|
gl = greenlet.greenlet(func)
|
|
fut = aioeventlet.wrap_greenthread(gl)
|
|
gl.switch()
|
|
self.assertRaises(ValueError, self.loop.run_until_complete, fut)
|
|
|
|
def test_wrap_greenlet_no_run_attr(self):
|
|
gl = greenlet.greenlet()
|
|
msg = "wrap_greenthread: the run attribute of the greenlet is not set"
|
|
self.assertRaisesRegexp(RuntimeError, msg,
|
|
aioeventlet.wrap_greenthread, gl)
|
|
|
|
def test_wrap_greenlet_running(self):
|
|
def func(value):
|
|
gl = greenlet.getcurrent()
|
|
return aioeventlet.wrap_greenthread(gl)
|
|
|
|
gl = greenlet.greenlet(func)
|
|
msg = "wrap_greenthread: the greenthread is running"
|
|
self.assertRaisesRegexp(RuntimeError, msg, gl.switch, 5)
|
|
|
|
def test_wrap_greenlet_dead(self):
|
|
def func(value):
|
|
return value * 3
|
|
|
|
gl = greenlet.greenlet(func)
|
|
gl.switch(5)
|
|
msg = "wrap_greenthread: the greenthread already finished"
|
|
self.assertRaisesRegexp(RuntimeError, msg, aioeventlet.wrap_greenthread, gl)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import unittest
|
|
unittest.main()
|
|
|