1 # test asynchat -- requires threading

     2 

     3 import thread # If this fails, we can't test this module

     4 import asyncore, asynchat, socket, threading, time

     5 import unittest

     6 from test import test_support

     7 

     8 HOST = "127.0.0.1"

     9 PORT = 54322

    10 

    11 class echo_server(threading.Thread):

    12 

    13     def run(self):

    14         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    15         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    16         global PORT

    17         PORT = test_support.bind_port(sock, HOST, PORT)

    18         sock.listen(1)

    19         conn, client = sock.accept()

    20         buffer = ""

    21         while "\n" not in buffer:

    22             data = conn.recv(1)

    23             if not data:

    24                 break

    25             buffer = buffer + data

    26         while buffer:

    27             n = conn.send(buffer)

    28             buffer = buffer[n:]

    29         conn.close()

    30         sock.close()

    31 

    32 class echo_client(asynchat.async_chat):

    33 

    34     def __init__(self, terminator):

    35         asynchat.async_chat.__init__(self)

    36         self.contents = None

    37         self.create_socket(socket.AF_INET, socket.SOCK_STREAM)

    38         self.connect((HOST, PORT))

    39         self.set_terminator(terminator)

    40         self.buffer = ""

    41 

    42     def handle_connect(self):

    43         pass

    44         ##print "Connected"

    45 

    46     def collect_incoming_data(self, data):

    47         self.buffer = self.buffer + data

    48 

    49     def found_terminator(self):

    50         #print "Received:", repr(self.buffer)

    51         self.contents = self.buffer

    52         self.buffer = ""

    53         self.close()

    54 

    55 

    56 class TestAsynchat(unittest.TestCase):

    57     def setUp (self):

    58         pass

    59 

    60     def tearDown (self):

    61         pass

    62 

    63     def test_line_terminator(self):

    64         s = echo_server()

    65         s.start()

    66         time.sleep(1) # Give server time to initialize

    67         c = echo_client('\n')

    68         c.push("hello ")

    69         c.push("world\n")

    70         asyncore.loop()

    71         s.join()

    72 

    73         self.assertEqual(c.contents, 'hello world')

    74 

    75     def test_numeric_terminator(self):

    76         # Try reading a fixed number of bytes

    77         s = echo_server()

    78         s.start()

    79         time.sleep(1) # Give server time to initialize

    80         c = echo_client(6L)

    81         c.push("hello ")

    82         c.push("world\n")

    83         asyncore.loop()

    84         s.join()

    85 

    86         self.assertEqual(c.contents, 'hello ')

    87 

    88 

    89 def test_main(verbose=None):

    90     test_support.run_unittest(TestAsynchat)

    91 

    92 if __name__ == "__main__":

    93     test_main(verbose=True)