]> git.ipfire.org Git - thirdparty/Python/cpython.git/commitdiff
attempt to fix asyncore buildbot failure
authorGiampaolo Rodola' <g.rodola@gmail.com>
Fri, 23 Mar 2012 14:09:58 +0000 (15:09 +0100)
committerGiampaolo Rodola' <g.rodola@gmail.com>
Fri, 23 Mar 2012 14:09:58 +0000 (15:09 +0100)
1  2 
Lib/test/test_asyncore.py

index 0bd09fd737ad3628fd7f4fea4d37670b18e6812e,5f55df89f51cf802d9bcbae37a843ac55c8b3986..1123c251ebd4ad29e9e031ed29db36a8936f458b
@@@ -783,40 -735,23 +783,44 @@@ class BaseTestAPI(unittest.TestCase)
      @support.reap_threads
      def test_quick_connect(self):
          # see: http://bugs.python.org/issue10340
 -        server = TCPServer()
 -        t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1, count=500))
 -        t.start()
 +        if self.family in (socket.AF_INET, getattr(socket, "AF_INET6", object())):
 +            server = BaseServer(self.family, self.addr)
 +            t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
 +                                                              count=500))
 +            t.start()
 +
-             for x in range(20):
-                 s = socket.socket(self.family, socket.SOCK_STREAM)
-                 s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
-                              struct.pack('ii', 1, 0))
 -        for x in range(20):
 -            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
++            s = socket.socket(self.family, socket.SOCK_STREAM)
+             s.settimeout(.2)
+             s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+                          struct.pack('ii', 1, 0))
+             try:
                  s.connect(server.address)
+             except socket.error:
+                 pass
+             finally:
                  s.close()
  
 -class TestAPI_UseSelect(BaseTestAPI):
 +class TestAPI_UseIPv4Sockets(BaseTestAPI):
 +    family = socket.AF_INET
 +    addr = (HOST, 0)
 +
 +@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 support required')
 +class TestAPI_UseIPv6Sockets(BaseTestAPI):
 +    family = socket.AF_INET6
 +    addr = ('::1', 0)
 +
 +@unittest.skipUnless(HAS_UNIX_SOCKETS, 'Unix sockets required')
 +class TestAPI_UseUnixSockets(BaseTestAPI):
 +    if HAS_UNIX_SOCKETS:
 +        family = socket.AF_UNIX
 +    addr = support.TESTFN
 +
 +    def tearDown(self):
 +        unlink(self.addr)
 +        BaseTestAPI.tearDown(self)
 +
 +class TestAPI_UseIPv4Select(TestAPI_UseIPv4Sockets):
      use_poll = False
  
  @unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')