modules/libjuju/tests/unit/test_loop.py

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