test_loop.py

   1 import asyncio
   2 import unittest
   3 import juju.loop
   4
   5
   6 class TestLoop(unittest.TestCase):
   7     def setUp(self):
   8         # new event loop for each test
   9         policy = asyncio.get_event_loop_policy()
  10         self.loop = policy.new_event_loop()
  11         policy.set_event_loop(self.loop)
  12
  13     def tearDown(self):
  14         self.loop.close()
  15
  16     def test_run(self):
  17         assert asyncio.get_event_loop() == self.loop
  18         async def _test():
  19             return 'success'
  20         self.assertEqual(juju.loop.run(_test()), 'success')
  21
  22     def test_run_interrupt(self):
  23         async def _test():
  24             juju.loop.run._sigint = True
  25         self.assertRaises(KeyboardInterrupt, juju.loop.run, _test())
  26
  27     def test_run_exception(self):
  28         async def _test():
  29             raise ValueError()
  30         self.assertRaises(ValueError, juju.loop.run, _test())