Lib/test/test_asyncore.py

   1 import asyncore
   2 import unittest
   3 import select
   4 import os
   5 import socket
   6 import threading
   7 import sys
   8 import time
   9
  10 from test import test_support
  11 from test.test_support import TESTFN, run_unittest, unlink
  12 from StringIO import StringIO
  13
  14 HOST = test_support.HOST
  15
  16 class dummysocket:
  17     def __init__(self):
  18         self.closed = False
  19
  20     def close(self):
  21         self.closed = True
  22
  23     def fileno(self):
  24         return 42
  25
  26 class dummychannel:
  27     def __init__(self):
  28         self.socket = dummysocket()
  29
  30     def close(self):
  31         self.socket.close()
  32
  33 class exitingdummy:
  34     def __init__(self):
  35         pass
  36
  37     def handle_read_event(self):
  38         raise asyncore.ExitNow()
  39
  40     handle_write_event = handle_read_event
  41     handle_close = handle_read_event
  42     handle_expt_event = handle_read_event
  43
  44 class crashingdummy:
  45     def __init__(self):
  46         self.error_handled = False
  47
  48     def handle_read_event(self):
  49         raise Exception()
  50
  51     handle_write_event = handle_read_event
  52     handle_close = handle_read_event
  53     handle_expt_event = handle_read_event
  54
  55     def handle_error(self):
  56         self.error_handled = True
  57
  58 # used when testing senders; just collects what it gets until newline is sent
  59 def capture_server(evt, buf, serv):
  60     try:
  61         serv.listen(5)
  62         conn, addr = serv.accept()
  63     except socket.timeout:
  64         pass
  65     else:
  66         n = 200
  67         while n > 0:
  68             r, w, e = select.select([conn], [], [])
  69             if r:
  70                 data = conn.recv(10)
  71                 # keep everything except for the newline terminator
  72                 buf.write(data.replace('\n', ''))
  73                 if '\n' in data:
  74                     break
  75             n -= 1
  76             time.sleep(0.01)
  77
  78         conn.close()
  79     finally:
  80         serv.close()
  81         evt.set()
  82
  83
  84 class HelperFunctionTests(unittest.TestCase):
  85     def test_readwriteexc(self):
  86         # Check exception handling behavior of read, write and _exception
  87
  88         # check that ExitNow exceptions in the object handler method
  89         # bubbles all the way up through asyncore read/write/_exception calls
  90         tr1 = exitingdummy()
  91         self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
  92         self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
  93         self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)
  94
  95         # check that an exception other than ExitNow in the object handler
  96         # method causes the handle_error method to get called
  97         tr2 = crashingdummy()
  98         asyncore.read(tr2)
  99         self.assertEqual(tr2.error_handled, True)
 100
 101         tr2 = crashingdummy()
 102         asyncore.write(tr2)
 103         self.assertEqual(tr2.error_handled, True)
 104
 105         tr2 = crashingdummy()
 106         asyncore._exception(tr2)
 107         self.assertEqual(tr2.error_handled, True)
 108
 109     # asyncore.readwrite uses constants in the select module that
 110     # are not present in Windows systems (see this thread:
 111     # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
 112     # These constants should be present as long as poll is available
 113
 114     if hasattr(select, 'poll'):
 115         def test_readwrite(self):
 116             # Check that correct methods are called by readwrite()
 117
 118             class testobj:
 119                 def __init__(self):
 120                     self.read = False
 121                     self.write = False
 122                     self.closed = False
 123                     self.expt = False
 124
 125                 def handle_read_event(self):
 126                     self.read = True
 127
 128                 def handle_write_event(self):
 129                     self.write = True
 130
 131                 def handle_close(self):
 132                     self.closed = True
 133
 134                 def handle_expt_event(self):
 135                     self.expt = True
 136
 137                 def handle_error(self):
 138                     self.error_handled = True
 139
 140             for flag in (select.POLLIN, select.POLLPRI):
 141                 tobj = testobj()
 142                 self.assertEqual(tobj.read, False)
 143                 asyncore.readwrite(tobj, flag)
 144                 self.assertEqual(tobj.read, True)
 145
 146                 # check that ExitNow exceptions in the object handler method
 147                 # bubbles all the way up through asyncore readwrite call
 148                 tr1 = exitingdummy()
 149                 self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
 150
 151                 # check that an exception other than ExitNow in the object handler
 152                 # method causes the handle_error method to get called
 153                 tr2 = crashingdummy()
 154                 asyncore.readwrite(tr2, flag)
 155                 self.assertEqual(tr2.error_handled, True)
 156
 157             tobj = testobj()
 158             self.assertEqual(tobj.write, False)
 159             asyncore.readwrite(tobj, select.POLLOUT)
 160             self.assertEqual(tobj.write, True)
 161
 162             # check that ExitNow exceptions in the object handler method
 163             # bubbles all the way up through asyncore readwrite call
 164             tr1 = exitingdummy()
 165             self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1,
 166                               select.POLLOUT)
 167
 168             # check that an exception other than ExitNow in the object handler
 169             # method causes the handle_error method to get called
 170             tr2 = crashingdummy()
 171             asyncore.readwrite(tr2, select.POLLOUT)
 172             self.assertEqual(tr2.error_handled, True)
 173
 174             for flag in (select.POLLERR, select.POLLHUP, select.POLLNVAL):
 175                 tobj = testobj()
 176                 self.assertEqual((tobj.expt, tobj.closed)[flag == select.POLLHUP], False)
 177                 asyncore.readwrite(tobj, flag)
 178                 self.assertEqual((tobj.expt, tobj.closed)[flag == select.POLLHUP], True)
 179
 180                 # check that ExitNow exceptions in the object handler method
 181                 # bubbles all the way up through asyncore readwrite calls
 182                 tr1 = exitingdummy()
 183                 self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
 184
 185                 # check that an exception other than ExitNow in the object handler
 186                 # method causes the handle_error method to get called
 187                 tr2 = crashingdummy()
 188                 asyncore.readwrite(tr2, flag)
 189                 self.assertEqual(tr2.error_handled, True)
 190
 191     def test_closeall(self):
 192         self.closeall_check(False)
 193
 194     def test_closeall_default(self):
 195         self.closeall_check(True)
 196
 197     def closeall_check(self, usedefault):
 198         # Check that close_all() closes everything in a given map
 199
 200         l = []
 201         testmap = {}
 202         for i in range(10):
 203             c = dummychannel()
 204             l.append(c)
 205             self.assertEqual(c.socket.closed, False)
 206             testmap[i] = c
 207
 208         if usedefault:
 209             socketmap = asyncore.socket_map
 210             try:
 211                 asyncore.socket_map = testmap
 212                 asyncore.close_all()
 213             finally:
 214                 testmap, asyncore.socket_map = asyncore.socket_map, socketmap
 215         else:
 216             asyncore.close_all(testmap)
 217
 218         self.assertEqual(len(testmap), 0)
 219
 220         for c in l:
 221             self.assertEqual(c.socket.closed, True)
 222
 223     def test_compact_traceback(self):
 224         try:
 225             raise Exception("I don't like spam!")
 226         except:
 227             real_t, real_v, real_tb = sys.exc_info()
 228             r = asyncore.compact_traceback()
 229         else:
 230             self.fail("Expected exception")
 231
 232         (f, function, line), t, v, info = r
 233         self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')
 234         self.assertEqual(function, 'test_compact_traceback')
 235         self.assertEqual(t, real_t)
 236         self.assertEqual(v, real_v)
 237         self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))
 238
 239
 240 class DispatcherTests(unittest.TestCase):
 241     def setUp(self):
 242         pass
 243
 244     def tearDown(self):
 245         asyncore.close_all()
 246
 247     def test_basic(self):
 248         d = asyncore.dispatcher()
 249         self.assertEqual(d.readable(), True)
 250         self.assertEqual(d.writable(), True)
 251
 252     def test_repr(self):
 253         d = asyncore.dispatcher()
 254         self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))
 255
 256     def test_log(self):
 257         d = asyncore.dispatcher()
 258
 259         # capture output of dispatcher.log() (to stderr)
 260         fp = StringIO()
 261         stderr = sys.stderr
 262         l1 = "Lovely spam! Wonderful spam!"
 263         l2 = "I don't like spam!"
 264         try:
 265             sys.stderr = fp
 266             d.log(l1)
 267             d.log(l2)
 268         finally:
 269             sys.stderr = stderr
 270
 271         lines = fp.getvalue().splitlines()
 272         self.assertEquals(lines, ['log: %s' % l1, 'log: %s' % l2])
 273
 274     def test_log_info(self):
 275         d = asyncore.dispatcher()
 276
 277         # capture output of dispatcher.log_info() (to stdout via print)
 278         fp = StringIO()
 279         stdout = sys.stdout
 280         l1 = "Have you got anything without spam?"
 281         l2 = "Why can't she have egg bacon spam and sausage?"
 282         l3 = "THAT'S got spam in it!"
 283         try:
 284             sys.stdout = fp
 285             d.log_info(l1, 'EGGS')
 286             d.log_info(l2)
 287             d.log_info(l3, 'SPAM')
 288         finally:
 289             sys.stdout = stdout
 290
 291         lines = fp.getvalue().splitlines()
 292         if __debug__:
 293             expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]
 294         else:
 295             expected = ['EGGS: %s' % l1, 'SPAM: %s' % l3]
 296
 297         self.assertEquals(lines, expected)
 298
 299     def test_unhandled(self):
 300         d = asyncore.dispatcher()
 301
 302         # capture output of dispatcher.log_info() (to stdout via print)
 303         fp = StringIO()
 304         stdout = sys.stdout
 305         try:
 306             sys.stdout = fp
 307             d.handle_expt()
 308             d.handle_read()
 309             d.handle_write()
 310             d.handle_connect()
 311             d.handle_accept()
 312         finally:
 313             sys.stdout = stdout
 314
 315         lines = fp.getvalue().splitlines()
 316         expected = ['warning: unhandled exception',
 317                     'warning: unhandled read event',
 318                     'warning: unhandled write event',
 319                     'warning: unhandled connect event',
 320                     'warning: unhandled accept event']
 321         self.assertEquals(lines, expected)
 322
 323
 324
 325 class dispatcherwithsend_noread(asyncore.dispatcher_with_send):
 326     def readable(self):
 327         return False
 328
 329     def handle_connect(self):
 330         pass
 331
 332 class DispatcherWithSendTests(unittest.TestCase):
 333     usepoll = False
 334
 335     def setUp(self):
 336         pass
 337
 338     def tearDown(self):
 339         asyncore.close_all()
 340
 341     def test_send(self):
 342         self.evt = threading.Event()
 343         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 344         self.sock.settimeout(3)
 345         self.port = test_support.bind_port(self.sock)
 346
 347         cap = StringIO()
 348         args = (self.evt, cap, self.sock)
 349         threading.Thread(target=capture_server, args=args).start()
 350
 351         # wait a little longer for the server to initialize (it sometimes
 352         # refuses connections on slow machines without this wait)
 353         time.sleep(0.2)
 354
 355         data = "Suppose there isn't a 16-ton weight?"
 356         d = dispatcherwithsend_noread()
 357         d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
 358         d.connect((HOST, self.port))
 359
 360         # give time for socket to connect
 361         time.sleep(0.1)
 362
 363         d.send(data)
 364         d.send(data)
 365         d.send('\n')
 366
 367         n = 1000
 368         while d.out_buffer and n > 0:
 369             asyncore.poll()
 370             n -= 1
 371
 372         self.evt.wait()
 373
 374         self.assertEqual(cap.getvalue(), data*2)
 375
 376
 377 class DispatcherWithSendTests_UsePoll(DispatcherWithSendTests):
 378     usepoll = True
 379
 380 if hasattr(asyncore, 'file_wrapper'):
 381     class FileWrapperTest(unittest.TestCase):
 382         def setUp(self):
 383             self.d = "It's not dead, it's sleeping!"
 384             file(TESTFN, 'w').write(self.d)
 385
 386         def tearDown(self):
 387             unlink(TESTFN)
 388
 389         def test_recv(self):
 390             fd = os.open(TESTFN, os.O_RDONLY)
 391             w = asyncore.file_wrapper(fd)
 392             os.close(fd)
 393
 394             self.assertNotEqual(w.fd, fd)
 395             self.assertNotEqual(w.fileno(), fd)
 396             self.assertEqual(w.recv(13), "It's not dead")
 397             self.assertEqual(w.read(6), ", it's")
 398             w.close()
 399             self.assertRaises(OSError, w.read, 1)
 400
 401         def test_send(self):
 402             d1 = "Come again?"
 403             d2 = "I want to buy some cheese."
 404             fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
 405             w = asyncore.file_wrapper(fd)
 406             os.close(fd)
 407
 408             w.write(d1)
 409             w.send(d2)
 410             w.close()
 411             self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
 412
 413
 414 def test_main():
 415     tests = [HelperFunctionTests, DispatcherTests, DispatcherWithSendTests,
 416              DispatcherWithSendTests_UsePoll]
 417     if hasattr(asyncore, 'file_wrapper'):
 418         tests.append(FileWrapperTest)
 419
 420     run_unittest(*tests)
 421
 422 if __name__ == "__main__":
 423     test_main()