Drop support for python 2.6.

On travis, start a build with 2.7.8 so that _ssl2.py gets
tested (previously tested by Python 2.6). Do so with the test_ssl.py
from that version (because the changes are large), but keep using the
2.7.11 tests with just a few removals otherwise.

.pep8

 [pep8]
 ignore=E702,E265,E402,E731,E266,E261,W503,E129
 max_line_length=160
-exclude=.runtimes,.eggs,.tox,.git,build,2.6,2.7,2.7pypy,3.3,3.5,test_support.py,test_queue.py,patched_tests_setup.py,test_threading_2.py,lock_tests.py,_sslgte279.py,3.4
+exclude=.runtimes,.eggs,.tox,.git,build,2.6,2.7,2.7.8,2.7pypy,3.3,3.5,test_support.py,test_queue.py,patched_tests_setup.py,test_threading_2.py,lock_tests.py,_sslgte279.py,3.4

.travis.yml

@@ -12,7 +12,7 @@ env:
       - TASK=test-pypy
       - TASK=lint-py27
       - TASK=test-py35
-      - TASK=test-py26
+      - TASK=test-py278
       - TASK=test-py27
       - TASK=test-py33
       - TASK=test-py34

Makefile

@@ -62,7 +62,7 @@ test_prelim:
 	make bench
 
 toxtest: test_prelim
-	cd greentest && GEVENT_RESOLVER=thread python testrunner.py --config ../known_failures.py
+	cd greentest && GEVENT_RESOLVER=thread ${PYTHON} testrunner.py --config ../known_failures.py
 
 fulltoxtest: test_prelim
 	cd greentest && GEVENT_RESOLVER=thread ${PYTHON} testrunner.py --config ../known_failures.py
@@ -92,7 +92,7 @@ travis_test_linters:
 
 BUILD_RUNTIMES?=$(PWD)/.runtimes
 
-PY26=$(BUILD_RUNTIMES)/snakepit/python2.6
+PY278=$(BUILD_RUNTIMES)/snakepit/python2.7.8
 PY27=$(BUILD_RUNTIMES)/snakepit/python2.7
 PY33=$(BUILD_RUNTIMES)/snakepit/python3.3
 PY34=$(BUILD_RUNTIMES)/snakepit/python3.4
@@ -109,8 +109,8 @@ TOOL_PYTHON=$(TOOL_VIRTUALENV)/bin/python
 TOOL_PIP=$(TOOL_VIRTUALENV)/bin/pip
 TOOL_INSTALL=$(TOOL_PIP) install --upgrade
 
-$(PY26):
-	scripts/install.sh 2.6
+$(PY278):
+	scripts/install.sh 2.7.8
 
 $(PY27):
 	scripts/install.sh 2.7
@@ -130,7 +130,8 @@ $(PYPY):
 PIP?=$(BUILD_RUNTIMES)/versions/$(PYTHON)/bin/pip
 
 develop:
-	echo $(PIP) $(PYTHON)
+	echo pip is at `which $(PIP)`
+	echo python is at `which $(PYTHON)`
 # First install a newer pip so that it can use the wheel cache
 # (only needed until travis upgrades pip to 7.x; note that the 3.5
 # environment uses pip 7.1 by default)
@@ -138,9 +139,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.
-	${PIP} install -U wheel
-	${PIP} install -U tox cython cffi greenlet pep8 pyflakes "coverage>=4.0" "coveralls>=1.0"
-	${PYTHON} setup.py develop
+	${PIP} install -U -r dev-requirements.txt
 
 lint-py27: $(PY27)
 	PYTHON=python2.7 PATH=$(BUILD_RUNTIMES)/versions/python2.7/bin:$(PATH) make develop travis_test_linters
@@ -148,8 +147,9 @@ lint-py27: $(PY27)
 test-py27: $(PY27)
 	PYTHON=python2.7 PATH=$(BUILD_RUNTIMES)/versions/python2.7/bin:$(PATH) make develop fulltoxtest
 
-test-py26: $(PY26)
-	PYTHON=python2.6 PATH=$(BUILD_RUNTIMES)/versions/python2.6/bin:$(PATH) make develop fulltoxtest
+test-py278: $(PY278)
+	ls $(BUILD_RUNTIMES)/versions/python2.7.8/bin/
+	PYTHON=python2.7.8 PATH=$(BUILD_RUNTIMES)/versions/python2.7.8/bin:$(PATH) make develop fulltoxtest
 
 test-py33: $(PY33)
 	PYTHON=python3.3 PATH=$(BUILD_RUNTIMES)/versions/python3.3/bin:$(PATH) make develop fulltoxtest

README.rst

@@ -27,7 +27,7 @@ Check out the detailed changelog_ for this version.
 Get gevent
 ==========
 
-Gevent runs on Python >= 2.6, Python >= 3.3, or PyPy >= 4.0.1
+Gevent runs on Python >= 2.7, Python >= 3.3, or PyPy >= 4.0.1
 (but not PyPy3) (*Note*: PyPy is not supported in Windows). On all
 platforms, installing setuptools is recommended (this is done
 automatically if working in a virtual environment).

changelog.rst

@@ -9,6 +9,7 @@
 
 - Update libev to version 4.22.
 - Update tblib to 1.3.0.
+- Remove support for Python 2.6.
 
 1.1.0 (Mar 5, 2016)
 ===================

dev-requirements.txt

 setuptools
+wheel
 cython>=0.23.4
 greenlet>=0.4.9
+pep8
+pyflakes
 coverage>=4.0
+coveralls>=1.0
 cffi
 -e .

doc/intro.rst

@@ -22,10 +22,10 @@ Features include:
 Installation and Requirements
 =============================
 
-`gevent 1.1`_ runs on Python 2 and Python 3. Versions 2.6 and 2.7 of
-Python 2 are supported, and versions 3.3, 3.4, and 3.5 of Python 3 are
+`gevent 1.1`_ runs on Python 2 and Python 3. Version 2.7 of
+Python 2 is supported, and versions 3.3, 3.4, and 3.5 of Python 3 are
 supported. (Users of older versions of Python 2 need to install gevent
-1.0.x; Python 3 is not supported by 1.0.) gevent requires the
+1.0.x (2.5) or 1.1.x (2.6); Python 3 is not supported by 1.0.) gevent requires the
 greenlet__ library.
 
 gevent 1.1 also runs on PyPy 2.6.1 and above, although 4.0 or above is

greentest/2.6/sha256.pem

------BEGIN CERTIFICATE-----
-MIIFxzCCA6+gAwIBAgIJALnlnf5uzTkIMA0GCSqGSIb3DQEBCwUAMEsxCzAJBgNV
-BAYTAkRFMRcwFQYDVQQKEw5zY2hva29rZWtzLm9yZzEjMCEGCSqGSIb3DQEJARYU
-aGFubm9Ac2Nob2tva2Vrcy5vcmcwHhcNMTAwMTI3MDAyMTI1WhcNMjAwMTI1MDAy
-MTI1WjBLMQswCQYDVQQGEwJERTEXMBUGA1UEChMOc2Nob2tva2Vrcy5vcmcxIzAh
-BgkqhkiG9w0BCQEWFGhhbm5vQHNjaG9rb2tla3Mub3JnMIICIjANBgkqhkiG9w0B
-AQEFAAOCAg8AMIICCgKCAgEApJ4ODPwEooMW35dQPlBqdvcfkEvjhcsA7jmJfFqN
-e/1T34zT44X9+KnMBSG2InacbD7eyFgjfaENFsZ87YkEBDIFZ/SHotLJZORQ8PUj
-YoxPG4mjKN+yL2WthNcYbRyJreTbbDroNMuw6tkTSxeSXyYFQrKMCUfErVbZa/d5
-RvfFVk+Au9dVUFhed/Stn5cv+a0ffvpyA7ygihm1kMFICbvPeI0846tmC2Ph7rM5
-pYQyNBDOVpULODTk5Wu6jiiJJygvJWCZ1FdpsdBs5aKWHWdRhX++quGuflTTjH5d
-qaIka4op9H7XksYphTDXmV+qHnva5jbPogwutDQcVsGBQcJaLmQqhsQK13bf4khE
-iWJvfBLfHn8OOpY25ZwwuigJIwifNCxQeeT1FrLmyuYNhz2phPpzx065kqSUSR+A
-Iw8DPE6e65UqMDKqZnID3dQeiQaFrHEV+Ibo0U/tD0YSBw5p33TMh0Es33IBWMac
-m7x4hIFWdhl8W522u6qOrTswY3s8vB7blNWqMc9n7oWH8ybFf7EgKeDVtEN9AyBE
-0WotXIEZWI+WvDbU1ACJXau9sQhYP/eerg7Zwr3iGUy4IQ5oUJibnjtcE+z8zmDN
-pE6YcMCLJyLjXiQ3iHG9mNXzw7wPnslTbEEEukrfSlHGgW8Dm+VrNyW0JUM1bntx
-vbMCAwEAAaOBrTCBqjAdBgNVHQ4EFgQUCedv7pDTuXtCxm4HTw9hUtrTvsowewYD
-VR0jBHQwcoAUCedv7pDTuXtCxm4HTw9hUtrTvsqhT6RNMEsxCzAJBgNVBAYTAkRF
-MRcwFQYDVQQKEw5zY2hva29rZWtzLm9yZzEjMCEGCSqGSIb3DQEJARYUaGFubm9A
-c2Nob2tva2Vrcy5vcmeCCQC55Z3+bs05CDAMBgNVHRMEBTADAQH/MA0GCSqGSIb3
-DQEBCwUAA4ICAQBHKAxA7WA/MEFjet03K8ouzEOr6Jrk2fZOuRhoDZ+9gr4FtaJB
-P3Hh5D00kuSOvDnwsvCohxeNd1KTMAwVmVoH+NZkHERn3UXniUENlp18koI1ehlr
-CZbXbzzE9Te9BelliSFA63q0cq0yJN1x9GyabU34XkAouCAmOqfSpKNZWZHGBHPF
-bbYnZrHEMcsye6vKeTOcg1GqUHGrQM2WK0QaOwnCQv2RblI9VN+SeRoUJ44qTXdW
-TwIYStsIPesacNcAQTStnHgKqIPx4zCwdx5xo8zONbXJfocqwyFqiAofvb9dN1nW
-g1noVBcXB+oRBZW5CjFw87U88itq39i9+BWl835DWLBW2pVmx1QTLGv0RNgs/xVx
-mWnjH4nNHvrjn6pRmqHZTk/SS0Hkl2qtDsynVxIl8EiMTfWSU3DBTuD2J/RSzuOE
-eKtAbaoXkXE31jCl4FEZLITIZd8UkXacb9rN304tAK92L76JOAV+xOZxFRipmvx4
-+A9qQXgLhtP4VaDajb44V/kCKPSA0Vm3apehke9Wl8dDtagfos1e6MxSu3EVLXRF
-SP2U777V77pdMSd0f/7cerKn5FjrxW1v1FaP1oIGniMk4qQNTgA/jvvhjybsPlVA
-jsfnhWGbh1voJa0RQcMiRMsxpw2P1KNOEu37W2eq/vFghVztZJQUmb5iNw==
------END CERTIFICATE-----

greentest/2.6/test_select.py

-from test import test_support
-import unittest
-import select
-import os
-import sys
-
-class SelectTestCase(unittest.TestCase):
-
-    class Nope:
-        pass
-
-    class Almost:
-        def fileno(self):
-            return 'fileno'
-
-    def test_error_conditions(self):
-        self.assertRaises(TypeError, select.select, 1, 2, 3)
-        self.assertRaises(TypeError, select.select, [self.Nope()], [], [])
-        self.assertRaises(TypeError, select.select, [self.Almost()], [], [])
-        self.assertRaises(TypeError, select.select, [], [], [], "not a number")
-
-    def test_returned_list_identity(self):
-        if sys.platform[:3] in ('win', 'mac', 'os2'):
-            if test_support.verbose:
-                print "can't easily test on this system"
-            return
-        # See issue #8329
-        r, w, x = select.select([], [], [], 1)
-        self.assertFalse(r is w)
-        self.assertFalse(r is x)
-        self.assertFalse(w is x)
-
-    def test_select(self):
-        if sys.platform[:3] in ('win', 'mac', 'os2'):
-            if test_support.verbose:
-                print "can't easily test on this system"
-            return
-        cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 0.1; done'
-        p = os.popen(cmd, 'r')
-        for tout in (0, 0.1, 0.2, 0.4, 0.8, 1.6) + (None,)*10:
-            if test_support.verbose:
-                print 'timeout =', tout
-            rfd, wfd, xfd = select.select([p], [], [], tout)
-            if (rfd, wfd, xfd) == ([], [], []):
-                continue
-            if (rfd, wfd, xfd) == ([p], [], []):
-                line = p.readline()
-                if test_support.verbose:
-                    print repr(line)
-                if not line:
-                    if test_support.verbose:
-                        print 'EOF'
-                    break
-                continue
-            self.fail('Unexpected return values from select():', rfd, wfd, xfd)
-        p.close()
-
-
-def test_main():
-    test_support.run_unittest(SelectTestCase)
-    test_support.reap_children()
-
-if __name__ == "__main__":
-    test_main()

greentest/2.6/test_socketserver.py

-"""
-Test suite for SocketServer.py.
-"""
-
-import contextlib
-import errno
-import imp
-import os
-import select
-import signal
-import socket
-import tempfile
-import threading
-import time
-import unittest
-import SocketServer
-
-import test.test_support
-from test.test_support import reap_children, verbose, TestSkipped
-from test.test_support import TESTFN as TEST_FILE
-
-test.test_support.requires("network")
-
-TEST_STR = "hello world\n"
-HOST = test.test_support.HOST
-
-HAVE_UNIX_SOCKETS = hasattr(socket, "AF_UNIX")
-HAVE_FORKING = hasattr(os, "fork") and os.name != "os2"
-
-def signal_alarm(n):
-    """Call signal.alarm when it exists (i.e. not on Windows)."""
-    if hasattr(signal, 'alarm'):
-        signal.alarm(n)
-
-def receive(sock, n, timeout=20):
-    r, w, x = select.select([sock], [], [], timeout)
-    if sock in r:
-        return sock.recv(n)
-    else:
-        raise RuntimeError, "timed out on %r" % (sock,)
-
-if HAVE_UNIX_SOCKETS:
-    class ForkingUnixStreamServer(SocketServer.ForkingMixIn,
-                                  SocketServer.UnixStreamServer):
-        pass
-
-    class ForkingUnixDatagramServer(SocketServer.ForkingMixIn,
-                                    SocketServer.UnixDatagramServer):
-        pass
-
-
-@contextlib.contextmanager
-def simple_subprocess(testcase):
-    pid = os.fork()
-    if pid == 0:
-        # Don't throw an exception; it would be caught by the test harness.
-        os._exit(72)
-    yield None
-    pid2, status = os.waitpid(pid, 0)
-    testcase.assertEquals(pid2, pid)
-    testcase.assertEquals(72 << 8, status)
-
-
-class SocketServerTest(unittest.TestCase):
-    """Test all socket servers."""
-
-    def setUp(self):
-        signal_alarm(20)  # Kill deadlocks after 20 seconds.
-        self.port_seed = 0
-        self.test_files = []
-
-    def tearDown(self):
-        signal_alarm(0)  # Didn't deadlock.
-        reap_children()
-
-        for fn in self.test_files:
-            try:
-                os.remove(fn)
-            except os.error:
-                pass
-        self.test_files[:] = []
-
-    def pickaddr(self, proto):
-        if proto == socket.AF_INET:
-            return (HOST, 0)
-        else:
-            # XXX: We need a way to tell AF_UNIX to pick its own name
-            # like AF_INET provides port==0.
-            dir = None
-            if os.name == 'os2':
-                dir = '\socket'
-            fn = tempfile.mktemp(prefix='unix_socket.', dir=dir)
-            if os.name == 'os2':
-                # AF_UNIX socket names on OS/2 require a specific prefix
-                # which can't include a drive letter and must also use
-                # backslashes as directory separators
-                if fn[1] == ':':
-                    fn = fn[2:]
-                if fn[0] in (os.sep, os.altsep):
-                    fn = fn[1:]
-                if os.sep == '/':
-                    fn = fn.replace(os.sep, os.altsep)
-                else:
-                    fn = fn.replace(os.altsep, os.sep)
-            self.test_files.append(fn)
-            return fn
-
-    def make_server(self, addr, svrcls, hdlrbase):
-        class MyServer(svrcls):
-            def handle_error(self, request, client_address):
-                self.close_request(request)
-                self.server_close()
-                raise
-
-        class MyHandler(hdlrbase):
-            def handle(self):
-                line = self.rfile.readline()
-                self.wfile.write(line)
-
-        if verbose: print "creating server"
-        server = MyServer(addr, MyHandler)
-        self.assertEquals(server.server_address, server.socket.getsockname())
-        return server
-
-    def run_server(self, svrcls, hdlrbase, testfunc):
-        server = self.make_server(self.pickaddr(svrcls.address_family),
-                                  svrcls, hdlrbase)
-        # We had the OS pick a port, so pull the real address out of
-        # the server.
-        addr = server.server_address
-        if verbose:
-            print "server created"
-            print "ADDR =", addr
-            print "CLASS =", svrcls
-        t = threading.Thread(
-            name='%s serving' % svrcls,
-            target=server.serve_forever,
-            # Short poll interval to make the test finish quickly.
-            # Time between requests is short enough that we won't wake
-            # up spuriously too many times.
-            kwargs={'poll_interval':0.01})
-        t.daemon = True  # In case this function raises.
-        t.start()
-        if verbose: print "server running"
-        for i in range(3):
-            if verbose: print "test client", i
-            testfunc(svrcls.address_family, addr)
-        if verbose: print "waiting for server"
-        server.shutdown()
-        t.join()
-        if verbose: print "done"
-
-    def stream_examine(self, proto, addr):
-        s = socket.socket(proto, socket.SOCK_STREAM)
-        s.connect(addr)
-        s.sendall(TEST_STR)
-        buf = data = receive(s, 100)
-        while data and '\n' not in buf:
-            data = receive(s, 100)
-            buf += data
-        self.assertEquals(buf, TEST_STR)
-        s.close()
-
-    def dgram_examine(self, proto, addr):
-        s = socket.socket(proto, socket.SOCK_DGRAM)
-        s.sendto(TEST_STR, addr)
-        buf = data = receive(s, 100)
-        while data and '\n' not in buf:
-            data = receive(s, 100)
-            buf += data
-        self.assertEquals(buf, TEST_STR)
-        s.close()
-
-    def test_TCPServer(self):
-        self.run_server(SocketServer.TCPServer,
-                        SocketServer.StreamRequestHandler,
-                        self.stream_examine)
-
-    def test_ThreadingTCPServer(self):
-        self.run_server(SocketServer.ThreadingTCPServer,
-                        SocketServer.StreamRequestHandler,
-                        self.stream_examine)
-
-    if HAVE_FORKING:
-        def test_ForkingTCPServer(self):
-            with simple_subprocess(self):
-                self.run_server(SocketServer.ForkingTCPServer,
-                                SocketServer.StreamRequestHandler,
-                                self.stream_examine)
-
-    if HAVE_UNIX_SOCKETS:
-        def test_UnixStreamServer(self):
-            self.run_server(SocketServer.UnixStreamServer,
-                            SocketServer.StreamRequestHandler,
-                            self.stream_examine)
-
-        def test_ThreadingUnixStreamServer(self):
-            self.run_server(SocketServer.ThreadingUnixStreamServer,
-                            SocketServer.StreamRequestHandler,
-                            self.stream_examine)
-
-        if HAVE_FORKING:
-            def test_ForkingUnixStreamServer(self):
-                with simple_subprocess(self):
-                    self.run_server(ForkingUnixStreamServer,
-                                    SocketServer.StreamRequestHandler,
-                                    self.stream_examine)
-
-    def test_UDPServer(self):
-        self.run_server(SocketServer.UDPServer,
-                        SocketServer.DatagramRequestHandler,
-                        self.dgram_examine)
-
-    def test_ThreadingUDPServer(self):
-        self.run_server(SocketServer.ThreadingUDPServer,
-                        SocketServer.DatagramRequestHandler,
-                        self.dgram_examine)
-
-    if HAVE_FORKING:
-        def test_ForkingUDPServer(self):
-            with simple_subprocess(self):
-                self.run_server(SocketServer.ForkingUDPServer,
-                                SocketServer.DatagramRequestHandler,
-                                self.dgram_examine)
-
-    # Alas, on Linux (at least) recvfrom() doesn't return a meaningful
-    # client address so this cannot work:
-
-    # if HAVE_UNIX_SOCKETS:
-    #     def test_UnixDatagramServer(self):
-    #         self.run_server(SocketServer.UnixDatagramServer,
-    #                         SocketServer.DatagramRequestHandler,
-    #                         self.dgram_examine)
-    #
-    #     def test_ThreadingUnixDatagramServer(self):
-    #         self.run_server(SocketServer.ThreadingUnixDatagramServer,
-    #                         SocketServer.DatagramRequestHandler,
-    #                         self.dgram_examine)
-    #
-    #     if HAVE_FORKING:
-    #         def test_ForkingUnixDatagramServer(self):
-    #             self.run_server(SocketServer.ForkingUnixDatagramServer,
-    #                             SocketServer.DatagramRequestHandler,
-    #                             self.dgram_examine)
-
-    def test_shutdown(self):
-        # Issue #2302: shutdown() should always succeed in making an
-        # other thread leave serve_forever().
-        class MyServer(SocketServer.TCPServer):
-            pass
-
-        class MyHandler(SocketServer.StreamRequestHandler):
-            pass
-
-        threads = []
-        for i in range(20):
-            s = MyServer((HOST, 0), MyHandler)
-            t = threading.Thread(
-                name='MyServer serving',
-                target=s.serve_forever,
-                kwargs={'poll_interval':0.01})
-            t.daemon = True  # In case this function raises.
-            threads.append((t, s))
-        for t, s in threads:
-            t.start()
-            s.shutdown()
-        for t, s in threads:
-            t.join()
-
-
-def test_main():
-    if imp.lock_held():
-        # If the import lock is held, the threads will hang
-        raise TestSkipped("can't run when import lock is held")
-
-    test.test_support.run_unittest(SocketServerTest)
-
-if __name__ == "__main__":
-    test_main()
-    signal_alarm(3)  # Shutdown shouldn't take more than 3 seconds.

greentest/2.6/test_telnetlib.py

-import socket
-import threading
-import telnetlib
-import time
-
-from unittest import TestCase
-from test import test_support
-
-HOST = test_support.HOST
-
-def server(evt, serv):
-    serv.listen(5)
-    evt.set()
-    try:
-        conn, addr = serv.accept()
-    except socket.timeout:
-        pass
-    finally:
-        serv.close()
-        evt.set()
-
-class GeneralTests(TestCase):
-
-    def setUp(self):
-        self.evt = threading.Event()
-        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.sock.settimeout(3)
-        self.port = test_support.bind_port(self.sock)
-        threading.Thread(target=server, args=(self.evt,self.sock)).start()
-        self.evt.wait()
-        self.evt.clear()
-        time.sleep(.1)
-
-    def tearDown(self):
-        self.evt.wait()
-
-    def testBasic(self):
-        # connects
-        telnet = telnetlib.Telnet(HOST, self.port)
-        telnet.sock.close()
-
-    def testTimeoutDefault(self):
-        self.assertTrue(socket.getdefaulttimeout() is None)
-        socket.setdefaulttimeout(30)
-        try:
-            telnet = telnetlib.Telnet("localhost", self.port)
-        finally:
-            socket.setdefaulttimeout(None)
-        self.assertEqual(telnet.sock.gettimeout(), 30)
-        telnet.sock.close()
-
-    def testTimeoutNone(self):
-        # None, having other default
-        self.assertTrue(socket.getdefaulttimeout() is None)
-        socket.setdefaulttimeout(30)
-        try:
-            telnet = telnetlib.Telnet(HOST, self.port, timeout=None)
-        finally:
-            socket.setdefaulttimeout(None)
-        self.assertTrue(telnet.sock.gettimeout() is None)
-        telnet.sock.close()
-
-    def testTimeoutValue(self):
-        telnet = telnetlib.Telnet("localhost", self.port, timeout=30)
-        self.assertEqual(telnet.sock.gettimeout(), 30)
-        telnet.sock.close()
-
-    def testTimeoutOpen(self):
-        telnet = telnetlib.Telnet()
-        telnet.open("localhost", self.port, timeout=30)
-        self.assertEqual(telnet.sock.gettimeout(), 30)
-        telnet.sock.close()
-
-
-def test_main(verbose=None):
-    test_support.run_unittest(GeneralTests)
-
-if __name__ == '__main__':
-    test_main()

greentest/2.6/test_threading_local.py

-import unittest
-from doctest import DocTestSuite
-from test import test_support
-import threading
-import weakref
-import gc
-
-class Weak(object):
-    pass
-
-def target(local, weaklist):
-    weak = Weak()
-    local.weak = weak
-    weaklist.append(weakref.ref(weak))
-
-class ThreadingLocalTest(unittest.TestCase):
-
-    def test_local_refs(self):
-        self._local_refs(20)
-        self._local_refs(50)
-        self._local_refs(100)
-
-    def _local_refs(self, n):
-        local = threading.local()
-        weaklist = []
-        for i in range(n):
-            t = threading.Thread(target=target, args=(local, weaklist))
-            t.start()
-            t.join()
-        del t
-
-        gc.collect()
-        self.assertEqual(len(weaklist), n)
-
-        # XXX threading.local keeps the local of the last stopped thread alive.
-        deadlist = [weak for weak in weaklist if weak() is None]
-        self.assertEqual(len(deadlist), n-1)
-
-        # Assignment to the same thread local frees it sometimes (!)
-        local.someothervar = None
-        gc.collect()
-        deadlist = [weak for weak in weaklist if weak() is None]
-        self.assert_(len(deadlist) in (n-1, n), (n, len(deadlist)))
-
-    def test_derived(self):
-        # Issue 3088: if there is a threads switch inside the __init__
-        # of a threading.local derived class, the per-thread dictionary
-        # is created but not correctly set on the object.
-        # The first member set may be bogus.
-        import time
-        class Local(threading.local):
-            def __init__(self):
-                time.sleep(0.01)
-        local = Local()
-
-        def f(i):
-            local.x = i
-            # Simply check that the variable is correctly set
-            self.assertEqual(local.x, i)
-
-        threads= []
-        for i in range(10):
-            t = threading.Thread(target=f, args=(i,))
-            t.start()
-            threads.append(t)
-
-        for t in threads:
-            t.join()
-
-    def test_derived_cycle_dealloc(self):
-        # http://bugs.python.org/issue6990
-        class Local(threading.local):
-            pass
-        locals = None
-        passed = [False]
-        e1 = threading.Event()
-        e2 = threading.Event()
-
-        def f():
-            # 1) Involve Local in a cycle
-            cycle = [Local()]
-            cycle.append(cycle)
-            cycle[0].foo = 'bar'
-
-            # 2) GC the cycle (triggers threadmodule.c::local_clear
-            # before local_dealloc)
-            del cycle
-            gc.collect()
-            e1.set()
-            e2.wait()
-
-            # 4) New Locals should be empty
-            passed[0] = all(not hasattr(local, 'foo') for local in locals)
-
-        t = threading.Thread(target=f)
-        t.start()
-        e1.wait()
-
-        # 3) New Locals should recycle the original's address. Creating
-        # them in the thread overwrites the thread state and avoids the
-        # bug
-        locals = [Local() for i in range(10)]
-        e2.set()
-        t.join()
-
-        self.assertTrue(passed[0])
-
-
-def test_main():
-    suite = unittest.TestSuite()
-    suite.addTest(DocTestSuite('_threading_local'))
-    suite.addTest(unittest.makeSuite(ThreadingLocalTest))
-
-    try:
-        from thread import _local
-    except ImportError:
-        pass
-    else:
-        import _threading_local
-        local_orig = _threading_local.local
-        def setUp(test):
-            _threading_local.local = _local
-        def tearDown(test):
-            _threading_local.local = local_orig
-        suite.addTest(DocTestSuite('_threading_local',
-                                   setUp=setUp, tearDown=tearDown)
-                      )
-
-    test_support.run_unittest(suite)
-
-if __name__ == '__main__':
-    test_main()

greentest/2.6/version

-2.6.8

greentest/2.7/lock_tests.py

@@ -307,7 +307,7 @@ class EventTests(BaseTestCase):
 
     def test_reset_internal_locks(self):
         evt = self.eventtype()
-        if not hasattr(evt, '_Event__cond'):
+        if not hasattr(evt, '_Event__cond') or sys.version_info[:3] <= (2, 7, 8):
             self.skipTest("gevent: internal impl difference")
         old_lock = evt._Event__cond._Condition__lock
         evt._reset_internal_locks()

greentest/2.7/test_smtplib.py

@@ -294,7 +294,8 @@ class BadHELOServerTests(unittest.TestCase):
 
 @unittest.skipUnless(threading, 'Threading required for this test.')
 class TooLongLineTests(unittest.TestCase):
-    respdata = '250 OK' + ('.' * smtplib._MAXLINE * 2) + '\n'
+    # gevent: run on 2.7.8
+    respdata = '250 OK' + ('.' * getattr(smtplib, '_MAXLINE', 1) * 2) + '\n'
 
     def setUp(self):
         self.old_stdout = sys.stdout