Merge pull request #1087 from gevent/issue1062

Bump Travis Python versions

CHANGES.rst

@@ -29,6 +29,8 @@
 - The internal, undocumented class ``gevent._socket3._fileobject`` has
   been removed. See :issue:`1084`.
 
+- Travis CI tests on Python 3.7.0a4 and PyPy 2.7 5.10.0 and PyPy 3.5
+  5.10.1.
 
 1.3a1 (2018-01-27)
 ==================

Makefile

@@ -126,9 +126,9 @@ PY27=$(BUILD_RUNTIMES)/snakepit/python2.7.14
 PY34=$(BUILD_RUNTIMES)/snakepit/python3.4.7
 PY35=$(BUILD_RUNTIMES)/snakepit/python3.5.4
 PY36=$(BUILD_RUNTIMES)/snakepit/python3.6.4
-PY37=$(BUILD_RUNTIMES)/snakepit/python3.7.0a3
-PYPY=$(BUILD_RUNTIMES)/snakepit/pypy590
-PYPY3=$(BUILD_RUNTIMES)/snakepit/pypy3.5_590
+PY37=$(BUILD_RUNTIMES)/snakepit/python3.7.0a4
+PYPY=$(BUILD_RUNTIMES)/snakepit/pypy5100
+PYPY3=$(BUILD_RUNTIMES)/snakepit/pypy3.5_5101
 
 TOOLS=$(BUILD_RUNTIMES)/tools
 
@@ -140,8 +140,6 @@ TOOL_PYTHON=$(TOOL_VIRTUALENV)/bin/python
 TOOL_PIP=$(TOOL_VIRTUALENV)/bin/pip
 TOOL_INSTALL=$(TOOL_PIP) install --upgrade
 
-$(PY278):
-	scripts/install.sh 2.7.8
 
 $(PY27):
 	scripts/install.sh 2.7
@@ -176,7 +174,7 @@ develop:
 # Then start installing our deps so they can be cached. Note that use of --build-options / --global-options / --install-options
 # disables the cache.
 # We need wheel>=0.26 on Python 3.5. See previous revisions.
-	GEVENTSETUP_EV_VERIFY=3 python -m pip install -U -r dev-requirements.txt
+	GEVENTSETUP_EV_VERIFY=3 ${PYTHON} -m pip install -U -r dev-requirements.txt
 	${PYTHON} scripts/travis.py fold_end install
 
 test-py27: $(PY27)
@@ -192,13 +190,13 @@ test-py36: $(PY36)
 	PYTHON=python3.6.4 PATH=$(BUILD_RUNTIMES)/versions/python3.6.4/bin:$(PATH) make develop allbackendtest
 
 test-py37: $(PY37)
-	PYTHON=python3.7.0a3 PATH=$(BUILD_RUNTIMES)/versions/python3.7.0a3/bin:$(PATH) make develop allbackendtest
+	PYTHON=python3.7.0a4 PATH=$(BUILD_RUNTIMES)/versions/python3.7.0a4/bin:$(PATH) make develop allbackendtest
 
 test-pypy: $(PYPY)
-	PYTHON=$(PYPY) PATH=$(BUILD_RUNTIMES)/versions/pypy590/bin:$(PATH) make develop cffibackendtest coverage_combine
+	PYTHON=$(PYPY) PATH=$(BUILD_RUNTIMES)/versions/pypy5100/bin:$(PATH) make develop cffibackendtest coverage_combine
 
 test-pypy3: $(PYPY3)
-	PYTHON=$(PYPY3) PATH=$(BUILD_RUNTIMES)/versions/pypy3.5_590/bin:$(PATH) make develop basictest
+	PYTHON=$(PYPY3) PATH=$(BUILD_RUNTIMES)/versions/pypy3.5_5101/bin:$(PATH) make develop basictest
 
 test-py27-noembed: $(PY27)
 	cd deps/libev && ./configure --disable-dependency-tracking && make

scripts/install.sh

@@ -97,13 +97,13 @@ for var in "$@"; do
       install 3.6.4 python3.6.4
       ;;
     3.7)
-      install 3.7.0a3 python3.7.0a3
+      install 3.7.0a4 python3.7.0a4
       ;;
     pypy)
-      install pypy2.7-5.9.0 pypy590
+      install pypy2.7-5.10.0 pypy5100
       ;;
     pypy3)
-      install pypy3.5-5.9.0 pypy3.5_590
+      install pypy3.5-5.10.1 pypy3.5_5101
       ;;
   esac
 done

src/gevent/resolver_thread.py

@@ -3,7 +3,7 @@
 Native thread-based hostname resolver.
 """
 import _socket
-from gevent._compat import text_type
+
 from gevent.hub import get_hub
 
 
@@ -11,7 +11,7 @@ __all__ = ['Resolver']
 
 
 # trigger import of encodings.idna to avoid https://github.com/gevent/gevent/issues/349
-text_type('foo').encode('idna')
+u'foo'.encode('idna')
 
 
 class Resolver(object):

src/greentest/3.7/test_asyncore.py

@@ -726,14 +726,10 @@ class BaseTestAPI:
     def test_create_socket(self):
         s = asyncore.dispatcher()
         s.create_socket(self.family)
+        self.assertEqual(s.socket.type, socket.SOCK_STREAM)
         self.assertEqual(s.socket.family, self.family)
-        SOCK_NONBLOCK = getattr(socket, 'SOCK_NONBLOCK', 0)
-        sock_type = socket.SOCK_STREAM | SOCK_NONBLOCK
-        if hasattr(socket, 'SOCK_CLOEXEC'):
-            self.assertIn(s.socket.type,
-                          (sock_type | socket.SOCK_CLOEXEC, sock_type))
-        else:
-            self.assertEqual(s.socket.type, sock_type)
+        self.assertEqual(s.socket.gettimeout(), 0)
+        self.assertFalse(s.socket.get_inheritable())
 
     def test_bind(self):
         if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:

src/greentest/3.7/test_socket.py

@@ -1444,7 +1444,7 @@ class GeneralModuleTests(unittest.TestCase):
         socket.gethostbyname(domain)
         socket.gethostbyname_ex(domain)
         socket.getaddrinfo(domain,0,socket.AF_UNSPEC,socket.SOCK_STREAM)
-        # this may not work if the forward lookup choses the IPv6 address, as that doesn't
+        # this may not work if the forward lookup chooses the IPv6 address, as that doesn't
         # have a reverse entry yet
         # socket.gethostbyaddr('испытание.python.org')
 
@@ -1577,6 +1577,22 @@ class GeneralModuleTests(unittest.TestCase):
             self.assertEqual(str(s.family), 'AddressFamily.AF_INET')
             self.assertEqual(str(s.type), 'SocketKind.SOCK_STREAM')
 
+    def test_socket_consistent_sock_type(self):
+        SOCK_NONBLOCK = getattr(socket, 'SOCK_NONBLOCK', 0)
+        SOCK_CLOEXEC = getattr(socket, 'SOCK_CLOEXEC', 0)
+        sock_type = socket.SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC
+
+        with socket.socket(socket.AF_INET, sock_type) as s:
+            self.assertEqual(s.type, socket.SOCK_STREAM)
+            s.settimeout(1)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
+            s.settimeout(0)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
+            s.setblocking(True)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
+            s.setblocking(False)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
+
     @unittest.skipIf(os.name == 'nt', 'Will not work on Windows')
     def test_uknown_socket_family_repr(self):
         # Test that when created with a family that's not one of the known
@@ -1589,9 +1605,18 @@ class GeneralModuleTests(unittest.TestCase):
         # On Windows this trick won't work, so the test is skipped.
         fd, path = tempfile.mkstemp()
         self.addCleanup(os.unlink, path)
-        with socket.socket(family=42424, type=13331, fileno=fd) as s:
-            self.assertEqual(s.family, 42424)
-            self.assertEqual(s.type, 13331)
+        unknown_family = max(socket.AddressFamily.__members__.values()) + 1
+
+        unknown_type = max(
+            kind
+            for name, kind in socket.SocketKind.__members__.items()
+            if name not in {'SOCK_NONBLOCK', 'SOCK_CLOEXEC'}
+        ) + 1
+
+        with socket.socket(
+                family=unknown_family, type=unknown_type, fileno=fd) as s:
+            self.assertEqual(s.family, unknown_family)
+            self.assertEqual(s.type, unknown_type)
 
     @unittest.skipUnless(hasattr(os, 'sendfile'), 'test needs os.sendfile()')
     def test__sendfile_use_sendfile(self):
@@ -4399,7 +4424,7 @@ class UnbufferedFileObjectClassTestCase(FileObjectClassTestCase):
         self.write_file.write(self.write_msg)
         self.write_file.flush()
         self.evt2.set()
-        # Avoid cloding the socket before the server test has finished,
+        # Avoid closing the socket before the server test has finished,
         # otherwise system recv() will return 0 instead of EWOULDBLOCK.
         self.serv_finished.wait(5.0)
 
@@ -4533,6 +4558,10 @@ class NetworkConnectionNoServer(unittest.TestCase):
         expected_errnos = [ errno.ECONNREFUSED, ]
         if hasattr(errno, 'ENETUNREACH'):
             expected_errnos.append(errno.ENETUNREACH)
+        if hasattr(errno, 'EADDRNOTAVAIL'):
+            # bpo-31910: socket.create_connection() fails randomly
+            # with EADDRNOTAVAIL on Travis CI
+            expected_errnos.append(errno.EADDRNOTAVAIL)
 
         self.assertIn(cm.exception.errno, expected_errnos)
 
@@ -4671,7 +4700,7 @@ class TCPTimeoutTest(SocketTCPTest):
                          'test needs signal.alarm()')
     def testInterruptedTimeout(self):
         # XXX I don't know how to do this test on MSWindows or any other
-        # plaform that doesn't support signal.alarm() or os.kill(), though
+        # platform that doesn't support signal.alarm() or os.kill(), though
         # the bug should have existed on all platforms.
         self.serv.settimeout(5.0)   # must be longer than alarm
         class Alarm(Exception):
@@ -5080,7 +5109,7 @@ class InheritanceTest(unittest.TestCase):
     def test_SOCK_CLOEXEC(self):
         with socket.socket(socket.AF_INET,
                            socket.SOCK_STREAM | socket.SOCK_CLOEXEC) as s:
-            self.assertTrue(s.type & socket.SOCK_CLOEXEC)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
             self.assertFalse(s.get_inheritable())
 
     def test_default_inheritable(self):
@@ -5132,8 +5161,6 @@ class InheritanceTest(unittest.TestCase):
                              0)
 
 
-    @unittest.skipUnless(hasattr(socket, "socketpair"),
-                         "need socket.socketpair()")
     def test_socketpair(self):
         s1, s2 = socket.socketpair()
         self.addCleanup(s1.close)
@@ -5147,11 +5174,15 @@ class InheritanceTest(unittest.TestCase):
 class NonblockConstantTest(unittest.TestCase):
     def checkNonblock(self, s, nonblock=True, timeout=0.0):
         if nonblock:
-            self.assertTrue(s.type & socket.SOCK_NONBLOCK)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
             self.assertEqual(s.gettimeout(), timeout)
+            self.assertTrue(
+                fcntl.fcntl(s, fcntl.F_GETFL, os.O_NONBLOCK) & os.O_NONBLOCK)
         else:
-            self.assertFalse(s.type & socket.SOCK_NONBLOCK)
+            self.assertEqual(s.type, socket.SOCK_STREAM)
             self.assertEqual(s.gettimeout(), None)
+            self.assertFalse(
+                fcntl.fcntl(s, fcntl.F_GETFL, os.O_NONBLOCK) & os.O_NONBLOCK)
 
     @support.requires_linux_version(2, 6, 28)
     def test_SOCK_NONBLOCK(self):
@@ -5295,7 +5326,7 @@ class SendfileUsingSendTest(ThreadedTCPSocketTest):
     Test the send() implementation of socket.sendfile().
     """
 
-    FILESIZE = (10 * 1024 * 1024)  # 10MB
+    FILESIZE = (10 * 1024 * 1024)  # 10 MiB
     BUFSIZE = 8192
     FILEDATA = b""
     TIMEOUT = 2
@@ -5571,6 +5602,9 @@ class LinuxKernelCryptoAPI(unittest.TestCase):
         else:
             return sock
 
+    # bpo-31705: On kernel older than 4.5, sendto() failed with ENOKEY,
+    # at least on ppc64le architecture
+    @support.requires_linux_version(4, 5)
     def test_sha256(self):
         expected = bytes.fromhex("ba7816bf8f01cfea414140de5dae2223b00361a396"
                                  "177a9cb410ff61f20015ad")

src/greentest/3.7/test_ssl.py

@@ -850,7 +850,7 @@ class BasicSocketTests(unittest.TestCase):
         self.cert_time_ok("Jan  5 09:34:61 2018 GMT", 1515144901)
         self.cert_time_fail("Jan  5 09:34:62 2018 GMT")  # invalid seconds
 
-        # no special treatement for the special value:
+        # no special treatment for the special value:
         #   99991231235959Z (rfc 5280)
         self.cert_time_ok("Dec 31 23:59:59 9999 GMT", 253402300799.0)
 

src/greentest/3.7/test_subprocess.py

@@ -370,7 +370,7 @@ class ProcessTestCase(BaseTestCase):
         # is relative.
         python_dir, python_base = self._split_python_path()
         rel_python = os.path.join(os.curdir, python_base)
-        with support.temp_cwd('test_cwd_with_relative_arg', quiet=True) as wrong_dir: # gevent: use distinct name, avoid Travis CI failure
+        with support.temp_cwd('test_cwd_with_relative_arg', quiet=True) as wrong_dir: # gevent: use distinct name, avoid Travis CI failure)
             # Before calling with the correct cwd, confirm that the call fails
             # without cwd and with the wrong cwd.
             self.assertRaises(FileNotFoundError, subprocess.Popen,
@@ -2290,11 +2290,11 @@ class POSIXProcessTestCase(BaseTestCase):
         fds_to_keep = set(open_fds.pop() for _ in range(8))
         p = subprocess.Popen([sys.executable, fd_status],
                              stdout=subprocess.PIPE, close_fds=True,
-                             pass_fds=())
+                             pass_fds=fds_to_keep)
         output, ignored = p.communicate()
         remaining_fds = set(map(int, output.split(b',')))
 
-        self.assertFalse(remaining_fds & fds_to_keep & open_fds,
+        self.assertFalse((remaining_fds - fds_to_keep) & open_fds,
                          "Some fds not in pass_fds were left open")
         self.assertIn(1, remaining_fds, "Subprocess failed")
 
@@ -2743,11 +2743,6 @@ class Win32ProcessTestCase(BaseTestCase):
                           [sys.executable, "-c",
                            "import sys; sys.exit(47)"],
                           preexec_fn=lambda: 1)
-        self.assertRaises(ValueError, subprocess.call,
-                          [sys.executable, "-c",
-                           "import sys; sys.exit(47)"],
-                          stdout=subprocess.PIPE,
-                          close_fds=True)
 
     @support.cpython_only
     def test_issue31471(self):
@@ -2765,6 +2760,67 @@ class Win32ProcessTestCase(BaseTestCase):
                               close_fds=True)
         self.assertEqual(rc, 47)
 
+    def test_close_fds_with_stdio(self):
+        import msvcrt
+
+        fds = os.pipe()
+        self.addCleanup(os.close, fds[0])
+        self.addCleanup(os.close, fds[1])
+
+        handles = []
+        for fd in fds:
+            os.set_inheritable(fd, True)
+            handles.append(msvcrt.get_osfhandle(fd))
+
+        p = subprocess.Popen([sys.executable, "-c",
+                              "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
+                             stdout=subprocess.PIPE, close_fds=False)
+        stdout, stderr = p.communicate()
+        self.assertEqual(p.returncode, 0)
+        int(stdout.strip())  # Check that stdout is an integer
+
+        p = subprocess.Popen([sys.executable, "-c",
+                              "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
+                             stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
+        stdout, stderr = p.communicate()
+        self.assertEqual(p.returncode, 1)
+        self.assertIn(b"OSError", stderr)
+
+        # The same as the previous call, but with an empty handle_list
+        handle_list = []
+        startupinfo = subprocess.STARTUPINFO()
+        startupinfo.lpAttributeList = {"handle_list": handle_list}
+        p = subprocess.Popen([sys.executable, "-c",
+                              "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
+                             stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                             startupinfo=startupinfo, close_fds=True)
+        stdout, stderr = p.communicate()
+        self.assertEqual(p.returncode, 1)
+        self.assertIn(b"OSError", stderr)
+
+        # Check for a warning due to using handle_list and close_fds=False
+        with support.check_warnings((".*overriding close_fds", RuntimeWarning)):
+            startupinfo = subprocess.STARTUPINFO()
+            startupinfo.lpAttributeList = {"handle_list": handles[:]}
+            p = subprocess.Popen([sys.executable, "-c",
+                                  "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
+                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                 startupinfo=startupinfo, close_fds=False)
+            stdout, stderr = p.communicate()
+            self.assertEqual(p.returncode, 0)
+
+    def test_empty_attribute_list(self):
+        startupinfo = subprocess.STARTUPINFO()
+        startupinfo.lpAttributeList = {}
+        subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
+                        startupinfo=startupinfo)
+
+    def test_empty_handle_list(self):
+        startupinfo = subprocess.STARTUPINFO()
+        startupinfo.lpAttributeList = {"handle_list": []}
+        subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
+                        startupinfo=startupinfo)
+
     def test_shell_sequence(self):
         # Run command through the shell (sequence)
         newenv = os.environ.copy()

src/greentest/3.7/test_threading.py

@@ -17,7 +17,7 @@ import weakref
 import os
 import subprocess
 
-from test import lock_tests
+import lock_tests # gevent: use our local copy
 from test import support
 
 
@@ -132,10 +132,10 @@ class ThreadTests(BaseTestCase):
         # Kill the "immortal" _DummyThread
         del threading._active[ident[0]]
 
-    # run with a small(ish) thread stack size (256kB)
+    # run with a small(ish) thread stack size (256 KiB)
     def test_various_ops_small_stack(self):
         if verbose:
-            print('with 256kB thread stack size...')
+            print('with 256 KiB thread stack size...')
         try:
             threading.stack_size(262144)
         except _thread.error:
@@ -144,10 +144,10 @@ class ThreadTests(BaseTestCase):
         self.test_various_ops()
         threading.stack_size(0)
 
-    # run with a large thread stack size (1MB)
+    # run with a large thread stack size (1 MiB)
     def test_various_ops_large_stack(self):
         if verbose:
-            print('with 1MB thread stack size...')
+            print('with 1 MiB thread stack size...')
         try:
             threading.stack_size(0x100000)
         except _thread.error:
@@ -427,7 +427,7 @@ class ThreadTests(BaseTestCase):
         t.daemon = True
         self.assertIn('daemon', repr(t))
 
-    def test_deamon_param(self):
+    def test_daemon_param(self):
         t = threading.Thread()
         self.assertFalse(t.daemon)
         t = threading.Thread(daemon=False)
@@ -1138,14 +1138,6 @@ class TimerTests(BaseTestCase):
 class LockTests(lock_tests.LockTests):
     locktype = staticmethod(threading.Lock)
 
-    @unittest.skip("not on gevent")
-    def test_locked_repr(self):
-        pass
-
-    @unittest.skip("not on gevent")
-    def test_repr(self):
-        pass
-
 class PyRLockTests(lock_tests.RLockTests):
     locktype = staticmethod(threading._PyRLock)
 
@@ -1156,11 +1148,6 @@ class CRLockTests(lock_tests.RLockTests):
 class EventTests(lock_tests.EventTests):
     eventtype = staticmethod(threading.Event)
 
-    @unittest.skip("not on gevent")
-    def test_reset_internal_locks(self):
-        # xxx: gevent: This uses an internal _cond attribute we don't have
-        pass
-
 class ConditionAsRLockTests(lock_tests.RLockTests):
     # Condition uses an RLock by default and exports its API.
     locktype = staticmethod(threading.Condition)

src/greentest/3.7/version

-3.7.0a3
+3.7.0a4

src/greentest/greentest/__init__.py

@@ -33,6 +33,7 @@ from greentest.sysinfo import LIBUV
 from greentest.sysinfo import CFFI_BACKEND
 from greentest.sysinfo import DEBUG
 from greentest.sysinfo import RUN_LEAKCHECKS
+from greentest.sysinfo import RUN_COVERAGE
 
 from greentest.sysinfo import PY2
 from greentest.sysinfo import PY3

src/greentest/greentest/testcase.py

@@ -44,7 +44,7 @@ class TimeAssertMixin(object):
 
     if sysinfo.EXPECT_POOR_TIMER_RESOLUTION:
         # pylint:disable=unused-argument
-        def assertTimeWithinRange(self, delay, min_time, max_time):
+        def assertTimeWithinRange(self, time_taken, min_time, max_time):
             return
     else:
         def assertTimeWithinRange(self, time_taken, min_time, max_time):

src/greentest/lock_tests.py

@@ -423,7 +423,8 @@ class ConditionTests(BaseTestCase):
         Bunch(f, N).wait_for_finished()
         self.assertEqual(len(results), 5)
         for dt in results:
-            self.assertTrue(dt >= 0.19, dt) # XXX: libuv sometimes produces 0.19958
+            # XXX: libuv sometimes produces 0.19958
+            self.assertTimeWithinRange(dt, 0.2, 2.0)
 
 
 class BaseSemaphoreTests(BaseTestCase):
@@ -568,5 +569,198 @@ class BoundedSemaphoreTests(BaseSemaphoreTests):
         sem.release()
         self.assertRaises(ValueError, sem.release)
 
+class BarrierTests(BaseTestCase):
+    """
+    Tests for Barrier objects.
+    """
+    N = 5
+    defaultTimeout = 2.0
+
+    def setUp(self):
+        self.barrier = self.barriertype(self.N, timeout=self.defaultTimeout)
+    def tearDown(self):
+        self.barrier.abort()
+
+    def run_threads(self, f):
+        b = Bunch(f, self.N-1)
+        f()
+        b.wait_for_finished()
+
+    def multipass(self, results, n):
+        m = self.barrier.parties
+        self.assertEqual(m, self.N)
+        for i in range(n):
+            results[0].append(True)
+            self.assertEqual(len(results[1]), i * m)
+            self.barrier.wait()
+            results[1].append(True)
+            self.assertEqual(len(results[0]), (i + 1) * m)
+            self.barrier.wait()
+        self.assertEqual(self.barrier.n_waiting, 0)
+        self.assertFalse(self.barrier.broken)
+
+    def test_barrier(self, passes=1):
+        """
+        Test that a barrier is passed in lockstep
+        """
+        results = [[],[]]
+        def f():
+            self.multipass(results, passes)
+        self.run_threads(f)
+
+    def test_barrier_10(self):
+        """
+        Test that a barrier works for 10 consecutive runs
+        """
+        return self.test_barrier(10)
+
+    def test_wait_return(self):
+        """
+        test the return value from barrier.wait
+        """
+        results = []
+        def f():
+            r = self.barrier.wait()
+            results.append(r)
+
+        self.run_threads(f)
+        self.assertEqual(sum(results), sum(range(self.N)))
+
+    def test_action(self):
+        """
+        Test the 'action' callback
+        """
+        results = []
+        def action():
+            results.append(True)
+        barrier = self.barriertype(self.N, action)
+        def f():
+            barrier.wait()
+            self.assertEqual(len(results), 1)
+
+        self.run_threads(f)
+
+    def test_abort(self):
+        """
+        Test that an abort will put the barrier in a broken state
+        """
+        results1 = []
+        results2 = []
+        def f():
+            try:
+                i = self.barrier.wait()
+                if i == self.N//2:
+                    raise RuntimeError
+                self.barrier.wait()
+                results1.append(True)
+            except threading.BrokenBarrierError:
+                results2.append(True)
+            except RuntimeError:
+                self.barrier.abort()
+                pass
+
+        self.run_threads(f)
+        self.assertEqual(len(results1), 0)
+        self.assertEqual(len(results2), self.N-1)
+        self.assertTrue(self.barrier.broken)
+
+    def test_reset(self):
+        """
+        Test that a 'reset' on a barrier frees the waiting threads
+        """
+        results1 = []
+        results2 = []
+        results3 = []
+        def f():
+            i = self.barrier.wait()
+            if i == self.N//2:
+                # Wait until the other threads are all in the barrier.
+                while self.barrier.n_waiting < self.N-1:
+                    time.sleep(0.001)
+                self.barrier.reset()
+            else:
+                try:
+                    self.barrier.wait()
+                    results1.append(True)
+                except threading.BrokenBarrierError:
+                    results2.append(True)
+            # Now, pass the barrier again
+            self.barrier.wait()
+            results3.append(True)
+
+        self.run_threads(f)
+        self.assertEqual(len(results1), 0)
+        self.assertEqual(len(results2), self.N-1)
+        self.assertEqual(len(results3), self.N)
+
+
+    def test_abort_and_reset(self):
+        """
+        Test that a barrier can be reset after being broken.
+        """
+        results1 = []
+        results2 = []
+        results3 = []
+        barrier2 = self.barriertype(self.N)
+        def f():
+            try:
+                i = self.barrier.wait()
+                if i == self.N//2:
+                    raise RuntimeError
+                self.barrier.wait()
+                results1.append(True)
+            except threading.BrokenBarrierError:
+                results2.append(True)
+            except RuntimeError:
+                self.barrier.abort()
+                pass
+            # Synchronize and reset the barrier.  Must synchronize first so
+            # that everyone has left it when we reset, and after so that no
+            # one enters it before the reset.
+            if barrier2.wait() == self.N//2:
+                self.barrier.reset()
+            barrier2.wait()
+            self.barrier.wait()
+            results3.append(True)
+
+        self.run_threads(f)
+        self.assertEqual(len(results1), 0)
+        self.assertEqual(len(results2), self.N-1)
+        self.assertEqual(len(results3), self.N)
+
+    def test_timeout(self):
+        """
+        Test wait(timeout)
+        """
+        def f():
+            i = self.barrier.wait()
+            if i == self.N // 2:
+                # One thread is late!
+                time.sleep(1.0)
+            # Default timeout is 2.0, so this is shorter.
+            self.assertRaises(threading.BrokenBarrierError,
+                              self.barrier.wait, 0.5)
+        self.run_threads(f)
+
+    def test_default_timeout(self):
+        """
+        Test the barrier's default timeout
+        """
+        # create a barrier with a low default timeout
+        barrier = self.barriertype(self.N, timeout=0.3)
+        def f():
+            i = barrier.wait()
+            if i == self.N // 2:
+                # One thread is later than the default timeout of 0.3s.
+                time.sleep(1.0)
+            self.assertRaises(threading.BrokenBarrierError, barrier.wait)
+        self.run_threads(f)
+
+    def test_single_thread(self):
+        b = self.barriertype(1)
+        b.wait()
+        b.wait()
+
+
 if __name__ == '__main__':
     print("This module contains no tests; it is used by other test cases like test_threading_2")

src/greentest/test___monkey_patching.py

@@ -32,11 +32,11 @@ def TESTRUNNER(tests=None):
     if preferred_version != version:
         util.log('WARNING: The tests in %s/ are from version %s and your Python is %s', directory, preferred_version, version)
 
+    version_tests = glob.glob('%s/test_*.py' % full_directory)
+    version_tests = sorted(version_tests)
     if not tests:
         tests = glob.glob('%s/test_*.py' % directory)
-        version_tests = glob.glob('%s/test_*.py' % full_directory)
         tests = sorted(tests)
-        version_tests = sorted(version_tests)
 
     PYTHONPATH = (os.getcwd() + os.pathsep + get_absolute_pythonpath()).rstrip(':')
 

src/greentest/test__threadpool.py

@@ -74,12 +74,8 @@ class PoolBasicTests(TestCase):
         def raiser():
             raise ExpectedException()
 
-        try:
+        with self.assertRaises(ExpectedException):
             pool.apply(raiser)
-        except ExpectedException:
-            pass
-        else:
-            self.fail("Should have raised ExpectedException")
     # Don't let the metaclass automatically force any error
     # that reaches the hub from a spawned greenlet to become
     # fatal; that defeats the point of the test.
@@ -97,8 +93,8 @@ class PoolBasicTests(TestCase):
 
 class TimingWrapper(object):
 
-    def __init__(self, func):
-        self.func = func
+    def __init__(self, the_func):
+        self.func = the_func
         self.elapsed = None
 
     def __call__(self, *args, **kwds):
@@ -147,10 +143,10 @@ class _AbstractPoolTest(TestCase):
 SMALL_RANGE = 10
 LARGE_RANGE = 1000
 
-if greentest.PYPY and greentest.WIN:
-    # PyPy 5.9 is *really* slow at spawning or switching between threads on Windows
-    # Tests that happen instantaneously on other platforms
-    # time out due to the overhead
+if greentest.PYPY and (greentest.WIN or greentest.RUN_COVERAGE):
+    # PyPy 5.10 is *really* slow at spawning or switching between
+    # threads (especially on Windows or when coverage is enabled) Tests that happen
+    # instantaneously on other platforms time out due to the overhead
     LARGE_RANGE = 50
 
 class TestPool(_AbstractPoolTest):
@@ -168,7 +164,7 @@ class TestPool(_AbstractPoolTest):
 
     def test_async_callback(self):
         result = []
-        res = self.pool.apply_async(sqr, (7, TIMEOUT1,), callback=lambda x: result.append(x))
+        res = self.pool.apply_async(sqr, (7, TIMEOUT1,), callback=result.append)
         get = TimingWrapper(res.get)
         self.assertEqual(get(), 49)
         self.assertTimeoutAlmostEqual(get.elapsed, TIMEOUT1, 1)
@@ -208,10 +204,11 @@ class TestPool(_AbstractPoolTest):
     def test_imap_unordered_gc(self):
         it = self.pool.imap_unordered(sqr, range(SMALL_RANGE))
         result = []
-        for i in range(SMALL_RANGE):
+        for _ in range(SMALL_RANGE):
             result.append(six.advance_iterator(it))
             gc.collect()
-        self.assertRaises(StopIteration, lambda: six.advance_iterator(it))
+        with self.assertRaises(StopIteration):
+            six.advance_iterator(it)
         self.assertEqual(sorted(result), [x * x for x in range(SMALL_RANGE)])
 
     def test_imap_random(self):
@@ -439,9 +436,10 @@ class TestRef(TestCase):
             if PYPY:
                 gc.collect()
                 gc.collect()
-            for index, r in enumerate(refs):
-                assert r() is None, (index, r(), greentest.getrefcount(r()), refs)
-            assert len(refs) == 4, refs
+            for r in refs:
+                self.assertIsNone(r())
+
+            self.assertEqual(4, len(refs))
 
 
 class Object(object):
@@ -450,13 +448,15 @@ class Object(object):
 
 class SomeClass(object):
 
+    refs = None
+
     def func(self, arg1, kwarg1=None):
         result = Object()
         self.refs.extend([weakref.ref(x) for x in [arg1, kwarg1, result]])
         return result
 
 
-def func():
+def noop():
     pass
 
 
@@ -464,7 +464,7 @@ class TestRefCount(TestCase):
 
     def test(self):
         pool = ThreadPool(1)
-        pool.spawn(func)
+        pool.spawn(noop)
         gevent.sleep(0)
         pool.kill()
 
@@ -499,7 +499,7 @@ if hasattr(gevent.threadpool, 'ThreadPoolExecutor'):
 
             def callback(future):
                 future.calledback += 1
-                raise Exception("Expected, ignored")
+                raise greentest.ExpectedException("Expected, ignored")
 
             future = pool.submit(fn)
             future.calledback = 0