Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

asyncio: sync with Tulip 

* Tulip issue 194: Don't use sys.getrefcount() in unit tests
* signal.set_wakeup_fd() can now raise an OSError on Python 3.5
This commit is contained in:
Victor Stinner 2014-07-23 18:21:45 +02:00
parent 8966759b03
commit c4c464911a
3 changed files with 13 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Lib/asyncio/unix_events.py
View File

@ -74,7 +74,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
# event loop running in another thread cannot add a signal
# handler.
signal.set_wakeup_fd(self._csock.fileno())
except ValueError as exc:
except (ValueError, OSError) as exc:
raise RuntimeError(str(exc))
handle = events.Handle(callback, args, self)
@ -93,7 +93,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
if not self._signal_handlers:
try:
signal.set_wakeup_fd(-1)
except ValueError as nexc:
except (ValueError, OSError) as nexc:
logger.info('set_wakeup_fd(-1) failed: %s', nexc)
if exc.errno == errno.EINVAL:
@ -138,7 +138,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
if not self._signal_handlers:
try:
signal.set_wakeup_fd(-1)
except ValueError as exc:
except (ValueError, OSError) as exc:
logger.info('set_wakeup_fd(-1) failed: %s', exc)
return True

4
Lib/test/test_asyncio/test_selector_events.py
View File

@ -672,6 +672,8 @@ class SelectorTransportTests(test_utils.TestCase):
def test_connection_lost(self):
exc = OSError()
tr = _SelectorTransport(self.loop, self.sock, self.protocol, None)
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)
tr._call_connection_lost(exc)
self.protocol.connection_lost.assert_called_with(exc)
@ -679,8 +681,6 @@ class SelectorTransportTests(test_utils.TestCase):
self.assertIsNone(tr._sock)
self.assertIsNone(tr._protocol)
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)

17
Lib/test/test_asyncio/test_unix_events.py
View File

@ -435,6 +435,8 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
def test__call_connection_lost(self):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)
err = None
tr._call_connection_lost(err)
@ -442,13 +444,13 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
self.pipe.close.assert_called_with()
self.assertIsNone(tr._protocol)
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
def test__call_connection_lost_with_err(self):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)
err = OSError()
tr._call_connection_lost(err)
@ -456,9 +458,6 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
self.pipe.close.assert_called_with()
self.assertIsNone(tr._protocol)
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
@ -717,6 +716,8 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
def test__call_connection_lost(self):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)
err = None
tr._call_connection_lost(err)
@ -724,13 +725,13 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
self.pipe.close.assert_called_with()
self.assertIsNone(tr._protocol)
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
def test__call_connection_lost_with_err(self):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)
err = OSError()
tr._call_connection_lost(err)
@ -738,8 +739,6 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
self.pipe.close.assert_called_with()
self.assertIsNone(tr._protocol)
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
def test_close(self):