Changeset 740

Show
Ignore:
Timestamp:
07/07/07 17:53:03 (2 years ago)
Author:
robin
Message:

--

Files:

Legend:

Unmodified
Added
Removed
Modified
Copied
Moved

  • asycamore/trunk/asycamore/wsgistreams.py

    r739 r740  
    9595 
    9696 
    97 def run_tests(): 
    98     test_read() 
    99     if 0: 
    100         test_readlines() 
    101         test_readlines_with_linepending() 
    102         test_readlines_with_split_terminator_pending() 
    103         test_readlines_with_pending() 
    104         test_readlines_with_pending2() 
    10597 
    106  
    107 def test_read(): 
    108     from socket import socketpair 
    109     clientsock, dumbserver = socketpair() 
    110     dumbserver.send('0' * 64) 
    111     dumbserver.close() 
    112     inp = InputStream('', clientsock) 
    113     assert inp.read(64) == '0' * 64 
    114  
    115     clientsock, dumbserver = socketpair() 
    116     dumbserver.send('1' * 16) 
    117     dumbserver.send('2' * 16) 
    118     dumbserver.send('3' * 16) 
    119     dumbserver.send('4' * 16) 
    120     dumbserver.close() 
    121     inp = InputStream('', clientsock) 
    122  
    123     d = inp.read(16); assert d == '1' * 16, d 
    124     d = inp.read(16); assert d == '2' * 16, d 
    125     d = inp.read(16); assert d == '3' * 16, d 
    126     d = inp.read(16); assert d == '4' * 16, d 
    127  
    128  
    129 def test_readlines(): 
    130     from socket import socketpair 
    131     clientsock, dumbserver = socketpair() 
    132     linedata = '\r\n'.join(['1','2','3', '']) 
    133     dumbserver.send(linedata) 
    134     dumbserver.close() 
    135     inp = InputStream('', clientsock) 
    136     for i, ln in enumerate(inp.readlines()): 
    137         expect = '%s\r\n' % str(i+1) 
    138         assert expect == ln, '"%s" != "%s"' % ( 
    139                 expect, ln 
    140                 ) 
    141  
    142  
    143 def test_readlines_with_linepending(): 
    144     from socket import socketpair 
    145     clientsock, dumbserver = socketpair() 
    146     linedata = '\r\n'.join(['1','2','3', '']) 
    147     dumbserver.send(linedata) 
    148     dumbserver.close() 
    149     inp = InputStream('0\r\n', clientsock) 
    150     for i, ln in enumerate(inp.readlines()): 
    151         expect = '%s\r\n' % str(i) 
    152         assert expect == ln, '"%s" != "%s"' % ( 
    153                 expect, ln 
    154                 ) 
    155  
    156  
    157 def test_readlines_with_split_terminator_pending(): 
    158     from socket import socketpair 
    159     clientsock, dumbserver = socketpair() 
    160     linedata = '\n' + '\r\n'.join(['1','2','3', '']) 
    161     dumbserver.send(linedata) 
    162     dumbserver.close() 
    163     inp = InputStream('0\r', clientsock) 
    164     for i, ln in enumerate(inp.readlines()): 
    165         expect = '%s\r\n' % str(i) 
    166         assert expect == ln, '"%s" != "%s"' % ( 
    167                 expect, ln 
    168                 ) 
    169  
    170  
    171 def test_readlines_with_pending(): 
    172     """pending is line content, terminator is first thing off the socket.""" 
    173     from socket import socketpair 
    174     clientsock, dumbserver = socketpair() 
    175     linedata = '\r\n'.join(['','2','3', '']) 
    176     dumbserver.send(linedata) 
    177     dumbserver.close() 
    178     inp = InputStream('1', clientsock) 
    179     for i, ln in enumerate(inp.readlines()): 
    180         expect = '%s\r\n' % str(i+1) 
    181         assert expect == ln, '"%s" != "%s"' % ( 
    182                 expect, ln 
    183                 ) 
    184  
    185  
    186 def test_readlines_with_pending2(): 
    187     """pending is partial line content.""" 
    188     from socket import socketpair 
    189     clientsock, dumbserver = socketpair() 
    190     linedata = '\r\n'.join(['11','2222','3333', '']) 
    191     dumbserver.send(linedata) 
    192     dumbserver.close() 
    193     inp = InputStream('11', clientsock) 
    194     for i, ln in enumerate(inp.readlines()): 
    195         expect = '%s\r\n' % (str(i+1) * 4) 
    196         assert expect == ln, '"%s" != "%s"' % ( 
    197                 expect, ln 
    198                 ) 
    199  
    200 if __name__== '__main__': 
    201     run_tests()