Lib/test/test_asyncore.py

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