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             attributes = ('read', 'expt', 'write', 'closed', 'error_handled')
 119
 120             expected = (
 121                 (select.POLLIN, 'read'),
 122                 (select.POLLPRI, 'expt'),
 123                 (select.POLLOUT, 'write'),
 124                 (select.POLLERR, 'closed'),
 125                 (select.POLLHUP, 'closed'),
 126                 (select.POLLNVAL, 'closed'),
 127                 )
 128
 129             class testobj:
 130                 def __init__(self):
 131                     self.read = False
 132                     self.write = False
 133                     self.closed = False
 134                     self.expt = False
 135                     self.error_handled = False
 136
 137                 def handle_read_event(self):
 138                     self.read = True
 139
 140                 def handle_write_event(self):
 141                     self.write = True
 142
 143                 def handle_close(self):
 144                     self.closed = True
 145
 146                 def handle_expt_event(self):
 147                     self.expt = True
 148
 149                 def handle_error(self):
 150                     self.error_handled = True
 151
 152             for flag, expectedattr in expected:
 153                 tobj = testobj()
 154                 self.assertEqual(getattr(tobj, expectedattr), False)
 155                 asyncore.readwrite(tobj, flag)
 156
 157                 # Only the attribute modified by the routine we expect to be
 158                 # called should be True.
 159                 for attr in attributes:
 160                     self.assertEqual(getattr(tobj, attr), attr==expectedattr)
 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, flag)
 166
 167                 # check that an exception other than ExitNow in the object handler
 168                 # method causes the handle_error method to get called
 169                 tr2 = crashingdummy()
 170                 self.assertEqual(tr2.error_handled, False)
 171                 asyncore.readwrite(tr2, flag)
 172                 self.assertEqual(tr2.error_handled, True)
 173
 174     def test_closeall(self):
 175         self.closeall_check(False)
 176
 177     def test_closeall_default(self):
 178         self.closeall_check(True)
 179
 180     def closeall_check(self, usedefault):
 181         # Check that close_all() closes everything in a given map
 182
 183         l = []
 184         testmap = {}
 185         for i in range(10):
 186             c = dummychannel()
 187             l.append(c)
 188             self.assertEqual(c.socket.closed, False)
 189             testmap[i] = c
 190
 191         if usedefault:
 192             socketmap = asyncore.socket_map
 193             try:
 194                 asyncore.socket_map = testmap
 195                 asyncore.close_all()
 196             finally:
 197                 testmap, asyncore.socket_map = asyncore.socket_map, socketmap
 198         else:
 199             asyncore.close_all(testmap)
 200
 201         self.assertEqual(len(testmap), 0)
 202
 203         for c in l:
 204             self.assertEqual(c.socket.closed, True)
 205
 206     def test_compact_traceback(self):
 207         try:
 208             raise Exception("I don't like spam!")
 209         except:
 210             real_t, real_v, real_tb = sys.exc_info()
 211             r = asyncore.compact_traceback()
 212         else:
 213             self.fail("Expected exception")
 214
 215         (f, function, line), t, v, info = r
 216         self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')
 217         self.assertEqual(function, 'test_compact_traceback')
 218         self.assertEqual(t, real_t)
 219         self.assertEqual(v, real_v)
 220         self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))
 221
 222
 223 class DispatcherTests(unittest.TestCase):
 224     def setUp(self):
 225         pass
 226
 227     def tearDown(self):
 228         asyncore.close_all()
 229
 230     def test_basic(self):
 231         d = asyncore.dispatcher()
 232         self.assertEqual(d.readable(), True)
 233         self.assertEqual(d.writable(), True)
 234
 235     def test_repr(self):
 236         d = asyncore.dispatcher()
 237         self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))
 238
 239     def test_log(self):
 240         d = asyncore.dispatcher()
 241
 242         # capture output of dispatcher.log() (to stderr)
 243         fp = StringIO()
 244         stderr = sys.stderr
 245         l1 = "Lovely spam! Wonderful spam!"
 246         l2 = "I don't like spam!"
 247         try:
 248             sys.stderr = fp
 249             d.log(l1)
 250             d.log(l2)
 251         finally:
 252             sys.stderr = stderr
 253
 254         lines = fp.getvalue().splitlines()
 255         self.assertEquals(lines, ['log: %s' % l1, 'log: %s' % l2])
 256
 257     def test_log_info(self):
 258         d = asyncore.dispatcher()
 259
 260         # capture output of dispatcher.log_info() (to stdout via print)
 261         fp = StringIO()
 262         stdout = sys.stdout
 263         l1 = "Have you got anything without spam?"
 264         l2 = "Why can't she have egg bacon spam and sausage?"
 265         l3 = "THAT'S got spam in it!"
 266         try:
 267             sys.stdout = fp
 268             d.log_info(l1, 'EGGS')
 269             d.log_info(l2)
 270             d.log_info(l3, 'SPAM')
 271         finally:
 272             sys.stdout = stdout
 273
 274         lines = fp.getvalue().splitlines()
 275         expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]
 276
 277         self.assertEquals(lines, expected)
 278
 279     def test_unhandled(self):
 280         d = asyncore.dispatcher()
 281         d.ignore_log_types = ()
 282
 283         # capture output of dispatcher.log_info() (to stdout via print)
 284         fp = StringIO()
 285         stdout = sys.stdout
 286         try:
 287             sys.stdout = fp
 288             d.handle_expt()
 289             d.handle_read()
 290             d.handle_write()
 291             d.handle_connect()
 292             d.handle_accept()
 293         finally:
 294             sys.stdout = stdout
 295
 296         lines = fp.getvalue().splitlines()
 297         expected = ['warning: unhandled incoming priority event',
 298                     'warning: unhandled read event',
 299                     'warning: unhandled write event',
 300                     'warning: unhandled connect event',
 301                     'warning: unhandled accept event']
 302         self.assertEquals(lines, expected)
 303
 304
 305
 306 class dispatcherwithsend_noread(asyncore.dispatcher_with_send):
 307     def readable(self):
 308         return False
 309
 310     def handle_connect(self):
 311         pass
 312
 313 class DispatcherWithSendTests(unittest.TestCase):
 314     usepoll = False
 315
 316     def setUp(self):
 317         pass
 318
 319     def tearDown(self):
 320         asyncore.close_all()
 321
 322     @test_support.reap_threads
 323     def test_send(self):
 324         evt = threading.Event()
 325         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 326         sock.settimeout(3)
 327         port = test_support.bind_port(sock)
 328
 329         cap = StringIO()
 330         args = (evt, cap, sock)
 331         t = threading.Thread(target=capture_server, args=args)
 332         t.start()
 333         try:
 334             # wait a little longer for the server to initialize (it sometimes
 335             # refuses connections on slow machines without this wait)
 336             time.sleep(0.2)
 337
 338             data = "Suppose there isn't a 16-ton weight?"
 339             d = dispatcherwithsend_noread()
 340             d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
 341             d.connect((HOST, port))
 342
 343             # give time for socket to connect
 344             time.sleep(0.1)
 345
 346             d.send(data)
 347             d.send(data)
 348             d.send('\n')
 349
 350             n = 1000
 351             while d.out_buffer and n > 0:
 352                 asyncore.poll()
 353                 n -= 1
 354
 355             evt.wait()
 356
 357             self.assertEqual(cap.getvalue(), data*2)
 358         finally:
 359             t.join()
 360
 361
 362 class DispatcherWithSendTests_UsePoll(DispatcherWithSendTests):
 363     usepoll = True
 364
 365 if hasattr(asyncore, 'file_wrapper'):
 366     class FileWrapperTest(unittest.TestCase):
 367         def setUp(self):
 368             self.d = "It's not dead, it's sleeping!"
 369             file(TESTFN, 'w').write(self.d)
 370
 371         def tearDown(self):
 372             unlink(TESTFN)
 373
 374         def test_recv(self):
 375             fd = os.open(TESTFN, os.O_RDONLY)
 376             w = asyncore.file_wrapper(fd)
 377             os.close(fd)
 378
 379             self.assertNotEqual(w.fd, fd)
 380             self.assertNotEqual(w.fileno(), fd)
 381             self.assertEqual(w.recv(13), "It's not dead")
 382             self.assertEqual(w.read(6), ", it's")
 383             w.close()
 384             self.assertRaises(OSError, w.read, 1)
 385
 386         def test_send(self):
 387             d1 = "Come again?"
 388             d2 = "I want to buy some cheese."
 389             fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
 390             w = asyncore.file_wrapper(fd)
 391             os.close(fd)
 392
 393             w.write(d1)
 394             w.send(d2)
 395             w.close()
 396             self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
 397
 398
 399 def test_main():
 400     tests = [HelperFunctionTests, DispatcherTests, DispatcherWithSendTests,
 401              DispatcherWithSendTests_UsePoll]
 402     if hasattr(asyncore, 'file_wrapper'):
 403         tests.append(FileWrapperTest)
 404
 405     run_unittest(*tests)
 406
 407 if __name__ == "__main__":
 408     test_main()