def _get_running_loop():
try:
return asyncio.get_running_loop()
- except AttributeError:
+ except AttributeError: # pragma: no cover
return asyncio.get_event_loop()
self.recvfrom.set_result((data, addr))
self.recvfrom = None
- def error_received(self, exc):
+ def error_received(self, exc): # pragma: no cover
if self.recvfrom:
self.recvfrom.set_exception(exc)
self.writer.close()
try:
await self.writer.wait_closed()
- except AttributeError:
+ except AttributeError: # pragma: no cover
pass
async def getpeername(self):
import asyncio
import socket
+import time
import unittest
import dns.asyncbackend
def testSleep(self):
async def run():
before = time.time()
- self.backend.sleep(0.1)
+ await self.backend.sleep(0.1)
after = time.time()
self.assertTrue(after - before >= 0.1)
+ self.async_run(run)
try:
import trio