[Python-checkins] CVS: python/dist/src/Lib/test test_asynchat.py,NONE,1.1

[Guido van Rossum]

Update of /cvsroot/python/python/dist/src/Lib/test
In directory usw-pr-cvs1:/tmp/cvs-serv4880

Added Files:
	test_asynchat.py 
Log Message:
Add test for asynchat.  This also tests asyncore.

--- NEW FILE: test_asynchat.py ---
# test asynchat -- requires threading

import asyncore, asynchat, socket, threading

HOST = "127.0.0.1"
PORT = 54321

class echo_server(threading.Thread):

    def run(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind((HOST, PORT))
        sock.listen(1)
        conn, client = sock.accept()
        buffer = ""
        while "\n" not in buffer:
            data = conn.recv(10)
            if not data:
                break
            buffer = buffer + data
        while buffer:
            n = conn.send(buffer)
            buffer = buffer[n:]
        conn.close()
        sock.close()

class echo_client(asynchat.async_chat):

    def __init__(self):
        asynchat.async_chat.__init__(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect((HOST, PORT))
        self.set_terminator("\n")
        self.buffer = ""
        self.send("hello ")
        self.send("world\n")

    def handle_connect(self):
        print "Connected"

    def collect_incoming_data(self, data):
        self.buffer = self.buffer + data

    def found_terminator(self):
        print "Received:", `self.buffer`
        self.buffer = ""
        self.close()

def main():
    s = echo_server()
    s.start()
    c = echo_client()
    asyncore.loop()

main()