Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_future_wrap_future(self):
from uvloop.loop import _wrap_future
def run(arg):
return (arg, threading.get_ident())
ex = concurrent.futures.ThreadPoolExecutor(1)
f1 = ex.submit(run, 'oi')
f2 = _wrap_future(f1, loop=self.loop)
res, ident = self.loop.run_until_complete(f2)
self.assertIsInstance(f2, asyncio.Future)
self.assertEqual(res, 'oi')
self.assertNotEqual(ident, threading.get_ident())
def test_future_wrap_future_future(self):
from uvloop.loop import _wrap_future
f1 = self.create_future()
f2 = _wrap_future(f1)
self.assertIs(f1, f2)
def test_future_wrap_future_cancel(self):
from uvloop.loop import _wrap_future
f1 = concurrent.futures.Future()
f2 = _wrap_future(f1, loop=self.loop)
f2.cancel()
test_utils.run_briefly(self.loop)
self.assertTrue(f1.cancelled())
self.assertTrue(f2.cancelled())
def test_future_wrap_future_cancel2(self):
from uvloop.loop import _wrap_future
f1 = concurrent.futures.Future()
f2 = _wrap_future(f1, loop=self.loop)
f1.set_result(42)
f2.cancel()
test_utils.run_briefly(self.loop)
self.assertFalse(f1.cancelled())
self.assertEqual(f1.result(), 42)
self.assertTrue(f2.cancelled())