1import asyncore
2import unittest
3import select
4import os
5import socket
6import sys
7import time
8import errno
9import struct
10
11from test import support
12from io import BytesIO
13
14if support.PGO:
15    raise unittest.SkipTest("test is not helpful for PGO")
16
17try:
18    import threading
19except ImportError:
20    threading = None
21
22TIMEOUT = 3
23HAS_UNIX_SOCKETS = hasattr(socket, 'AF_UNIX')
24
25class dummysocket:
26    def __init__(self):
27        self.closed = False
28
29    def close(self):
30        self.closed = True
31
32    def fileno(self):
33        return 42
34
35class dummychannel:
36    def __init__(self):
37        self.socket = dummysocket()
38
39    def close(self):
40        self.socket.close()
41
42class exitingdummy:
43    def __init__(self):
44        pass
45
46    def handle_read_event(self):
47        raise asyncore.ExitNow()
48
49    handle_write_event = handle_read_event
50    handle_close = handle_read_event
51    handle_expt_event = handle_read_event
52
53class crashingdummy:
54    def __init__(self):
55        self.error_handled = False
56
57    def handle_read_event(self):
58        raise Exception()
59
60    handle_write_event = handle_read_event
61    handle_close = handle_read_event
62    handle_expt_event = handle_read_event
63
64    def handle_error(self):
65        self.error_handled = True
66
67# used when testing senders; just collects what it gets until newline is sent
68def capture_server(evt, buf, serv):
69    try:
70        serv.listen()
71        conn, addr = serv.accept()
72    except socket.timeout:
73        pass
74    else:
75        n = 200
76        start = time.time()
77        while n > 0 and time.time() - start < 3.0:
78            r, w, e = select.select([conn], [], [], 0.1)
79            if r:
80                n -= 1
81                data = conn.recv(10)
82                # keep everything except for the newline terminator
83                buf.write(data.replace(b'\n', b''))
84                if b'\n' in data:
85                    break
86            time.sleep(0.01)
87
88        conn.close()
89    finally:
90        serv.close()
91        evt.set()
92
93def bind_af_aware(sock, addr):
94    """Helper function to bind a socket according to its family."""
95    if HAS_UNIX_SOCKETS and sock.family == socket.AF_UNIX:
96        # Make sure the path doesn't exist.
97        support.unlink(addr)
98        support.bind_unix_socket(sock, addr)
99    else:
100        sock.bind(addr)
101
102
103class HelperFunctionTests(unittest.TestCase):
104    def test_readwriteexc(self):
105        # Check exception handling behavior of read, write and _exception
106
107        # check that ExitNow exceptions in the object handler method
108        # bubbles all the way up through asyncore read/write/_exception calls
109        tr1 = exitingdummy()
110        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
111        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
112        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)
113
114        # check that an exception other than ExitNow in the object handler
115        # method causes the handle_error method to get called
116        tr2 = crashingdummy()
117        asyncore.read(tr2)
118        self.assertEqual(tr2.error_handled, True)
119
120        tr2 = crashingdummy()
121        asyncore.write(tr2)
122        self.assertEqual(tr2.error_handled, True)
123
124        tr2 = crashingdummy()
125        asyncore._exception(tr2)
126        self.assertEqual(tr2.error_handled, True)
127
128    # asyncore.readwrite uses constants in the select module that
129    # are not present in Windows systems (see this thread:
130    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
131    # These constants should be present as long as poll is available
132
133    @unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
134    def test_readwrite(self):
135        # Check that correct methods are called by readwrite()
136
137        attributes = ('read', 'expt', 'write', 'closed', 'error_handled')
138
139        expected = (
140            (select.POLLIN, 'read'),
141            (select.POLLPRI, 'expt'),
142            (select.POLLOUT, 'write'),
143            (select.POLLERR, 'closed'),
144            (select.POLLHUP, 'closed'),
145            (select.POLLNVAL, 'closed'),
146            )
147
148        class testobj:
149            def __init__(self):
150                self.read = False
151                self.write = False
152                self.closed = False
153                self.expt = False
154                self.error_handled = False
155
156            def handle_read_event(self):
157                self.read = True
158
159            def handle_write_event(self):
160                self.write = True
161
162            def handle_close(self):
163                self.closed = True
164
165            def handle_expt_event(self):
166                self.expt = True
167
168            def handle_error(self):
169                self.error_handled = True
170
171        for flag, expectedattr in expected:
172            tobj = testobj()
173            self.assertEqual(getattr(tobj, expectedattr), False)
174            asyncore.readwrite(tobj, flag)
175
176            # Only the attribute modified by the routine we expect to be
177            # called should be True.
178            for attr in attributes:
179                self.assertEqual(getattr(tobj, attr), attr==expectedattr)
180
181            # check that ExitNow exceptions in the object handler method
182            # bubbles all the way up through asyncore readwrite call
183            tr1 = exitingdummy()
184            self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
185
186            # check that an exception other than ExitNow in the object handler
187            # method causes the handle_error method to get called
188            tr2 = crashingdummy()
189            self.assertEqual(tr2.error_handled, False)
190            asyncore.readwrite(tr2, flag)
191            self.assertEqual(tr2.error_handled, True)
192
193    def test_closeall(self):
194        self.closeall_check(False)
195
196    def test_closeall_default(self):
197        self.closeall_check(True)
198
199    def closeall_check(self, usedefault):
200        # Check that close_all() closes everything in a given map
201
202        l = []
203        testmap = {}
204        for i in range(10):
205            c = dummychannel()
206            l.append(c)
207            self.assertEqual(c.socket.closed, False)
208            testmap[i] = c
209
210        if usedefault:
211            socketmap = asyncore.socket_map
212            try:
213                asyncore.socket_map = testmap
214                asyncore.close_all()
215            finally:
216                testmap, asyncore.socket_map = asyncore.socket_map, socketmap
217        else:
218            asyncore.close_all(testmap)
219
220        self.assertEqual(len(testmap), 0)
221
222        for c in l:
223            self.assertEqual(c.socket.closed, True)
224
225    def test_compact_traceback(self):
226        try:
227            raise Exception("I don't like spam!")
228        except:
229            real_t, real_v, real_tb = sys.exc_info()
230            r = asyncore.compact_traceback()
231        else:
232            self.fail("Expected exception")
233
234        (f, function, line), t, v, info = r
235        self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')
236        self.assertEqual(function, 'test_compact_traceback')
237        self.assertEqual(t, real_t)
238        self.assertEqual(v, real_v)
239        self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))
240
241
242class DispatcherTests(unittest.TestCase):
243    def setUp(self):
244        pass
245
246    def tearDown(self):
247        asyncore.close_all()
248
249    def test_basic(self):
250        d = asyncore.dispatcher()
251        self.assertEqual(d.readable(), True)
252        self.assertEqual(d.writable(), True)
253
254    def test_repr(self):
255        d = asyncore.dispatcher()
256        self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))
257
258    def test_log(self):
259        d = asyncore.dispatcher()
260
261        # capture output of dispatcher.log() (to stderr)
262        l1 = "Lovely spam! Wonderful spam!"
263        l2 = "I don't like spam!"
264        with support.captured_stderr() as stderr:
265            d.log(l1)
266            d.log(l2)
267
268        lines = stderr.getvalue().splitlines()
269        self.assertEqual(lines, ['log: %s' % l1, 'log: %s' % l2])
270
271    def test_log_info(self):
272        d = asyncore.dispatcher()
273
274        # capture output of dispatcher.log_info() (to stdout via print)
275        l1 = "Have you got anything without spam?"
276        l2 = "Why can't she have egg bacon spam and sausage?"
277        l3 = "THAT'S got spam in it!"
278        with support.captured_stdout() as stdout:
279            d.log_info(l1, 'EGGS')
280            d.log_info(l2)
281            d.log_info(l3, 'SPAM')
282
283        lines = stdout.getvalue().splitlines()
284        expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]
285        self.assertEqual(lines, expected)
286
287    def test_unhandled(self):
288        d = asyncore.dispatcher()
289        d.ignore_log_types = ()
290
291        # capture output of dispatcher.log_info() (to stdout via print)
292        with support.captured_stdout() as stdout:
293            d.handle_expt()
294            d.handle_read()
295            d.handle_write()
296            d.handle_connect()
297
298        lines = stdout.getvalue().splitlines()
299        expected = ['warning: unhandled incoming priority event',
300                    'warning: unhandled read event',
301                    'warning: unhandled write event',
302                    'warning: unhandled connect event']
303        self.assertEqual(lines, expected)
304
305    def test_strerror(self):
306        # refers to bug #8573
307        err = asyncore._strerror(errno.EPERM)
308        if hasattr(os, 'strerror'):
309            self.assertEqual(err, os.strerror(errno.EPERM))
310        err = asyncore._strerror(-1)
311        self.assertTrue(err != "")
312
313
314class dispatcherwithsend_noread(asyncore.dispatcher_with_send):
315    def readable(self):
316        return False
317
318    def handle_connect(self):
319        pass
320
321
322class DispatcherWithSendTests(unittest.TestCase):
323    def setUp(self):
324        pass
325
326    def tearDown(self):
327        asyncore.close_all()
328
329    @unittest.skipUnless(threading, 'Threading required for this test.')
330    @support.reap_threads
331    def test_send(self):
332        evt = threading.Event()
333        sock = socket.socket()
334        sock.settimeout(3)
335        port = support.bind_port(sock)
336
337        cap = BytesIO()
338        args = (evt, cap, sock)
339        t = threading.Thread(target=capture_server, args=args)
340        t.start()
341        try:
342            # wait a little longer for the server to initialize (it sometimes
343            # refuses connections on slow machines without this wait)
344            time.sleep(0.2)
345
346            data = b"Suppose there isn't a 16-ton weight?"
347            d = dispatcherwithsend_noread()
348            d.create_socket()
349            d.connect((support.HOST, port))
350
351            # give time for socket to connect
352            time.sleep(0.1)
353
354            d.send(data)
355            d.send(data)
356            d.send(b'\n')
357
358            n = 1000
359            while d.out_buffer and n > 0:
360                asyncore.poll()
361                n -= 1
362
363            evt.wait()
364
365            self.assertEqual(cap.getvalue(), data*2)
366        finally:
367            t.join(timeout=TIMEOUT)
368            if t.is_alive():
369                self.fail("join() timed out")
370
371
372@unittest.skipUnless(hasattr(asyncore, 'file_wrapper'),
373                     'asyncore.file_wrapper required')
374class FileWrapperTest(unittest.TestCase):
375    def setUp(self):
376        self.d = b"It's not dead, it's sleeping!"
377        with open(support.TESTFN, 'wb') as file:
378            file.write(self.d)
379
380    def tearDown(self):
381        support.unlink(support.TESTFN)
382
383    def test_recv(self):
384        fd = os.open(support.TESTFN, os.O_RDONLY)
385        w = asyncore.file_wrapper(fd)
386        os.close(fd)
387
388        self.assertNotEqual(w.fd, fd)
389        self.assertNotEqual(w.fileno(), fd)
390        self.assertEqual(w.recv(13), b"It's not dead")
391        self.assertEqual(w.read(6), b", it's")
392        w.close()
393        self.assertRaises(OSError, w.read, 1)
394
395    def test_send(self):
396        d1 = b"Come again?"
397        d2 = b"I want to buy some cheese."
398        fd = os.open(support.TESTFN, os.O_WRONLY | os.O_APPEND)
399        w = asyncore.file_wrapper(fd)
400        os.close(fd)
401
402        w.write(d1)
403        w.send(d2)
404        w.close()
405        with open(support.TESTFN, 'rb') as file:
406            self.assertEqual(file.read(), self.d + d1 + d2)
407
408    @unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),
409                         'asyncore.file_dispatcher required')
410    def test_dispatcher(self):
411        fd = os.open(support.TESTFN, os.O_RDONLY)
412        data = []
413        class FileDispatcher(asyncore.file_dispatcher):
414            def handle_read(self):
415                data.append(self.recv(29))
416        s = FileDispatcher(fd)
417        os.close(fd)
418        asyncore.loop(timeout=0.01, use_poll=True, count=2)
419        self.assertEqual(b"".join(data), self.d)
420
421    def test_resource_warning(self):
422        # Issue #11453
423        fd = os.open(support.TESTFN, os.O_RDONLY)
424        f = asyncore.file_wrapper(fd)
425
426        os.close(fd)
427        with support.check_warnings(('', ResourceWarning)):
428            f = None
429            support.gc_collect()
430
431    def test_close_twice(self):
432        fd = os.open(support.TESTFN, os.O_RDONLY)
433        f = asyncore.file_wrapper(fd)
434        os.close(fd)
435
436        os.close(f.fd)  # file_wrapper dupped fd
437        with self.assertRaises(OSError):
438            f.close()
439
440        self.assertEqual(f.fd, -1)
441        # calling close twice should not fail
442        f.close()
443
444
445class BaseTestHandler(asyncore.dispatcher):
446
447    def __init__(self, sock=None):
448        asyncore.dispatcher.__init__(self, sock)
449        self.flag = False
450
451    def handle_accept(self):
452        raise Exception("handle_accept not supposed to be called")
453
454    def handle_accepted(self):
455        raise Exception("handle_accepted not supposed to be called")
456
457    def handle_connect(self):
458        raise Exception("handle_connect not supposed to be called")
459
460    def handle_expt(self):
461        raise Exception("handle_expt not supposed to be called")
462
463    def handle_close(self):
464        raise Exception("handle_close not supposed to be called")
465
466    def handle_error(self):
467        raise
468
469
470class BaseServer(asyncore.dispatcher):
471    """A server which listens on an address and dispatches the
472    connection to a handler.
473    """
474
475    def __init__(self, family, addr, handler=BaseTestHandler):
476        asyncore.dispatcher.__init__(self)
477        self.create_socket(family)
478        self.set_reuse_addr()
479        bind_af_aware(self.socket, addr)
480        self.listen(5)
481        self.handler = handler
482
483    @property
484    def address(self):
485        return self.socket.getsockname()
486
487    def handle_accepted(self, sock, addr):
488        self.handler(sock)
489
490    def handle_error(self):
491        raise
492
493
494class BaseClient(BaseTestHandler):
495
496    def __init__(self, family, address):
497        BaseTestHandler.__init__(self)
498        self.create_socket(family)
499        self.connect(address)
500
501    def handle_connect(self):
502        pass
503
504
505class BaseTestAPI:
506
507    def tearDown(self):
508        asyncore.close_all(ignore_all=True)
509
510    def loop_waiting_for_flag(self, instance, timeout=5):
511        timeout = float(timeout) / 100
512        count = 100
513        while asyncore.socket_map and count > 0:
514            asyncore.loop(timeout=0.01, count=1, use_poll=self.use_poll)
515            if instance.flag:
516                return
517            count -= 1
518            time.sleep(timeout)
519        self.fail("flag not set")
520
521    def test_handle_connect(self):
522        # make sure handle_connect is called on connect()
523
524        class TestClient(BaseClient):
525            def handle_connect(self):
526                self.flag = True
527
528        server = BaseServer(self.family, self.addr)
529        client = TestClient(self.family, server.address)
530        self.loop_waiting_for_flag(client)
531
532    def test_handle_accept(self):
533        # make sure handle_accept() is called when a client connects
534
535        class TestListener(BaseTestHandler):
536
537            def __init__(self, family, addr):
538                BaseTestHandler.__init__(self)
539                self.create_socket(family)
540                bind_af_aware(self.socket, addr)
541                self.listen(5)
542                self.address = self.socket.getsockname()
543
544            def handle_accept(self):
545                self.flag = True
546
547        server = TestListener(self.family, self.addr)
548        client = BaseClient(self.family, server.address)
549        self.loop_waiting_for_flag(server)
550
551    def test_handle_accepted(self):
552        # make sure handle_accepted() is called when a client connects
553
554        class TestListener(BaseTestHandler):
555
556            def __init__(self, family, addr):
557                BaseTestHandler.__init__(self)
558                self.create_socket(family)
559                bind_af_aware(self.socket, addr)
560                self.listen(5)
561                self.address = self.socket.getsockname()
562
563            def handle_accept(self):
564                asyncore.dispatcher.handle_accept(self)
565
566            def handle_accepted(self, sock, addr):
567                sock.close()
568                self.flag = True
569
570        server = TestListener(self.family, self.addr)
571        client = BaseClient(self.family, server.address)
572        self.loop_waiting_for_flag(server)
573
574
575    def test_handle_read(self):
576        # make sure handle_read is called on data received
577
578        class TestClient(BaseClient):
579            def handle_read(self):
580                self.flag = True
581
582        class TestHandler(BaseTestHandler):
583            def __init__(self, conn):
584                BaseTestHandler.__init__(self, conn)
585                self.send(b'x' * 1024)
586
587        server = BaseServer(self.family, self.addr, TestHandler)
588        client = TestClient(self.family, server.address)
589        self.loop_waiting_for_flag(client)
590
591    def test_handle_write(self):
592        # make sure handle_write is called
593
594        class TestClient(BaseClient):
595            def handle_write(self):
596                self.flag = True
597
598        server = BaseServer(self.family, self.addr)
599        client = TestClient(self.family, server.address)
600        self.loop_waiting_for_flag(client)
601
602    def test_handle_close(self):
603        # make sure handle_close is called when the other end closes
604        # the connection
605
606        class TestClient(BaseClient):
607
608            def handle_read(self):
609                # in order to make handle_close be called we are supposed
610                # to make at least one recv() call
611                self.recv(1024)
612
613            def handle_close(self):
614                self.flag = True
615                self.close()
616
617        class TestHandler(BaseTestHandler):
618            def __init__(self, conn):
619                BaseTestHandler.__init__(self, conn)
620                self.close()
621
622        server = BaseServer(self.family, self.addr, TestHandler)
623        client = TestClient(self.family, server.address)
624        self.loop_waiting_for_flag(client)
625
626    def test_handle_close_after_conn_broken(self):
627        # Check that ECONNRESET/EPIPE is correctly handled (issues #5661 and
628        # #11265).
629
630        data = b'\0' * 128
631
632        class TestClient(BaseClient):
633
634            def handle_write(self):
635                self.send(data)
636
637            def handle_close(self):
638                self.flag = True
639                self.close()
640
641            def handle_expt(self):
642                self.flag = True
643                self.close()
644
645        class TestHandler(BaseTestHandler):
646
647            def handle_read(self):
648                self.recv(len(data))
649                self.close()
650
651            def writable(self):
652                return False
653
654        server = BaseServer(self.family, self.addr, TestHandler)
655        client = TestClient(self.family, server.address)
656        self.loop_waiting_for_flag(client)
657
658    @unittest.skipIf(sys.platform.startswith("sunos"),
659                     "OOB support is broken on Solaris")
660    def test_handle_expt(self):
661        # Make sure handle_expt is called on OOB data received.
662        # Note: this might fail on some platforms as OOB data is
663        # tenuously supported and rarely used.
664        if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
665            self.skipTest("Not applicable to AF_UNIX sockets.")
666
667        if sys.platform == "darwin" and self.use_poll:
668            self.skipTest("poll may fail on macOS; see issue #28087")
669
670        class TestClient(BaseClient):
671            def handle_expt(self):
672                self.socket.recv(1024, socket.MSG_OOB)
673                self.flag = True
674
675        class TestHandler(BaseTestHandler):
676            def __init__(self, conn):
677                BaseTestHandler.__init__(self, conn)
678                self.socket.send(bytes(chr(244), 'latin-1'), socket.MSG_OOB)
679
680        server = BaseServer(self.family, self.addr, TestHandler)
681        client = TestClient(self.family, server.address)
682        self.loop_waiting_for_flag(client)
683
684    def test_handle_error(self):
685
686        class TestClient(BaseClient):
687            def handle_write(self):
688                1.0 / 0
689            def handle_error(self):
690                self.flag = True
691                try:
692                    raise
693                except ZeroDivisionError:
694                    pass
695                else:
696                    raise Exception("exception not raised")
697
698        server = BaseServer(self.family, self.addr)
699        client = TestClient(self.family, server.address)
700        self.loop_waiting_for_flag(client)
701
702    def test_connection_attributes(self):
703        server = BaseServer(self.family, self.addr)
704        client = BaseClient(self.family, server.address)
705
706        # we start disconnected
707        self.assertFalse(server.connected)
708        self.assertTrue(server.accepting)
709        # this can't be taken for granted across all platforms
710        #self.assertFalse(client.connected)
711        self.assertFalse(client.accepting)
712
713        # execute some loops so that client connects to server
714        asyncore.loop(timeout=0.01, use_poll=self.use_poll, count=100)
715        self.assertFalse(server.connected)
716        self.assertTrue(server.accepting)
717        self.assertTrue(client.connected)
718        self.assertFalse(client.accepting)
719
720        # disconnect the client
721        client.close()
722        self.assertFalse(server.connected)
723        self.assertTrue(server.accepting)
724        self.assertFalse(client.connected)
725        self.assertFalse(client.accepting)
726
727        # stop serving
728        server.close()
729        self.assertFalse(server.connected)
730        self.assertFalse(server.accepting)
731
732    def test_create_socket(self):
733        s = asyncore.dispatcher()
734        s.create_socket(self.family)
735        self.assertEqual(s.socket.family, self.family)
736        SOCK_NONBLOCK = getattr(socket, 'SOCK_NONBLOCK', 0)
737        sock_type = socket.SOCK_STREAM | SOCK_NONBLOCK
738        if hasattr(socket, 'SOCK_CLOEXEC'):
739            self.assertIn(s.socket.type,
740                          (sock_type | socket.SOCK_CLOEXEC, sock_type))
741        else:
742            self.assertEqual(s.socket.type, sock_type)
743
744    def test_bind(self):
745        if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
746            self.skipTest("Not applicable to AF_UNIX sockets.")
747        s1 = asyncore.dispatcher()
748        s1.create_socket(self.family)
749        s1.bind(self.addr)
750        s1.listen(5)
751        port = s1.socket.getsockname()[1]
752
753        s2 = asyncore.dispatcher()
754        s2.create_socket(self.family)
755        # EADDRINUSE indicates the socket was correctly bound
756        self.assertRaises(OSError, s2.bind, (self.addr[0], port))
757
758    def test_set_reuse_addr(self):
759        if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
760            self.skipTest("Not applicable to AF_UNIX sockets.")
761
762        with socket.socket(self.family) as sock:
763            try:
764                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
765            except OSError:
766                unittest.skip("SO_REUSEADDR not supported on this platform")
767            else:
768                # if SO_REUSEADDR succeeded for sock we expect asyncore
769                # to do the same
770                s = asyncore.dispatcher(socket.socket(self.family))
771                self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET,
772                                                     socket.SO_REUSEADDR))
773                s.socket.close()
774                s.create_socket(self.family)
775                s.set_reuse_addr()
776                self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET,
777                                                     socket.SO_REUSEADDR))
778
779    @unittest.skipUnless(threading, 'Threading required for this test.')
780    @support.reap_threads
781    def test_quick_connect(self):
782        # see: http://bugs.python.org/issue10340
783        if self.family not in (socket.AF_INET, getattr(socket, "AF_INET6", object())):
784            self.skipTest("test specific to AF_INET and AF_INET6")
785
786        server = BaseServer(self.family, self.addr)
787        # run the thread 500 ms: the socket should be connected in 200 ms
788        t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
789                                                          count=5))
790        t.start()
791        try:
792            with socket.socket(self.family, socket.SOCK_STREAM) as s:
793                s.settimeout(.2)
794                s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
795                             struct.pack('ii', 1, 0))
796
797                try:
798                    s.connect(server.address)
799                except OSError:
800                    pass
801        finally:
802            t.join(timeout=TIMEOUT)
803            if t.is_alive():
804                self.fail("join() timed out")
805
806class TestAPI_UseIPv4Sockets(BaseTestAPI):
807    family = socket.AF_INET
808    addr = (support.HOST, 0)
809
810@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 support required')
811class TestAPI_UseIPv6Sockets(BaseTestAPI):
812    family = socket.AF_INET6
813    addr = (support.HOSTv6, 0)
814
815@unittest.skipUnless(HAS_UNIX_SOCKETS, 'Unix sockets required')
816class TestAPI_UseUnixSockets(BaseTestAPI):
817    if HAS_UNIX_SOCKETS:
818        family = socket.AF_UNIX
819    addr = support.TESTFN
820
821    def tearDown(self):
822        support.unlink(self.addr)
823        BaseTestAPI.tearDown(self)
824
825class TestAPI_UseIPv4Select(TestAPI_UseIPv4Sockets, unittest.TestCase):
826    use_poll = False
827
828@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
829class TestAPI_UseIPv4Poll(TestAPI_UseIPv4Sockets, unittest.TestCase):
830    use_poll = True
831
832class TestAPI_UseIPv6Select(TestAPI_UseIPv6Sockets, unittest.TestCase):
833    use_poll = False
834
835@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
836class TestAPI_UseIPv6Poll(TestAPI_UseIPv6Sockets, unittest.TestCase):
837    use_poll = True
838
839class TestAPI_UseUnixSocketsSelect(TestAPI_UseUnixSockets, unittest.TestCase):
840    use_poll = False
841
842@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
843class TestAPI_UseUnixSocketsPoll(TestAPI_UseUnixSockets, unittest.TestCase):
844    use_poll = True
845
846if __name__ == "__main__":
847    unittest.main()
848