Commit 5171072c authored by Jason Madden's avatar Jason Madden

Convert test_close_backend_fd to a real test, rename to convention.

parent d6a1b54c
from __future__ import print_function
import os
import unittest
import gevent
from gevent import core
@unittest.skipUnless(
getattr(core, 'LIBEV_EMBED', False),
"Needs embedded libev. "
"hub.loop.fileno is only defined when "
"we embed libev for some reason. "
"Choosing specific backends is also only supported by libev "
"(not libuv), and besides, libuv has a nasty tendency to "
"abort() the process if its FD gets closed. "
)
class Test(unittest.TestCase):
# NOTE that we extend unittest.TestCase, not greentest.TestCase
# Extending the later causes the wrong hub to get used.
assertRaisesRegex = getattr(unittest.TestCase, 'assertRaisesRegex',
getattr(unittest.TestCase, 'assertRaisesRegexp'))
def _check_backend(self, backend):
hub = gevent.get_hub(backend, default=False)
try:
self.assertEqual(hub.loop.backend, backend)
gevent.sleep(0.001)
fileno = hub.loop.fileno()
if fileno is None:
raise unittest.SkipTest("backend %s lacks fileno" % (backend,))
os.close(fileno)
with self.assertRaisesRegex(SystemError, "(libev)"):
gevent.sleep(0.001)
hub.destroy()
self.assertIn('destroyed', repr(hub))
finally:
if hub.loop is not None:
hub.destroy()
def _make_test(count, backend): # pylint:disable=no-self-argument
def test(self):
self._check_backend(backend)
test.__name__ = 'test_' + backend + '_' + str(count)
return test.__name__, test
count = backend = None
for count in range(2):
for backend in core.supported_backends():
name, func = _make_test(count, backend)
locals()[name] = func
name = func = None
del count
del backend
del _make_test
if __name__ == '__main__':
unittest.main()
from __future__ import print_function
import os
import gevent
from gevent import core
if getattr(core, 'LIBEV_EMBED', False):
# hub.loop.fileno is only defined when
# we embed libev for some reason.
# Choosing specific backends is also only supported by libev
# (not libuv), and besides, libuv has a nasty tendency to
# abort() the process if its FD gets closed.
for count in range(2):
for backend in core.supported_backends():
hub = gevent.get_hub(backend, default=False)
assert hub.loop.backend == backend, (hub.loop.backend, backend)
gevent.sleep(0.001)
fileno = hub.loop.fileno()
if fileno is not None:
print('%s. Testing %r: %r' % (count, backend, hub))
os.close(fileno)
try:
gevent.sleep(0.001)
except SystemError as ex:
if '(libev)' in str(ex):
print('The error is expected: %s' % ex)
else:
raise
else:
raise AssertionError('gevent.sleep() is expected to fail after loop fd was closed')
else:
print('%s. %r lacks fileno()' % (count, backend))
hub.destroy()
assert 'destroyed' in repr(hub), repr(hub)
......@@ -3,3 +3,4 @@ test__monkey_ssl_warning.py
test___monitor.py
test__monkey_scope.py
test__ares_timeout.py
test__close_backend_fd.py
......@@ -21,3 +21,4 @@ test__events.py
test__monkey_scope.py
test__iwait.py
test__ares_timeout.py
test__close_backend_fd.py
......@@ -4,7 +4,7 @@ test__api_timeout.py
test__ares_host_result.py
test__ares_timeout.py # explicitly uses ares resolver
# uses socket test__backdoor.py
test_close_backend_fd.py
test__close_backend_fd.py
test__core_async.py
test__core_callback.py
test__core_loop_run.py
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment