--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/python-2.5.2/win32/Lib/test/test_asynchat.py	Fri Apr 03 17:19:34 2009 +0100
@@ -0,0 +1,93 @@
+# test asynchat -- requires threading
+
+import thread # If this fails, we can't test this module
+import asyncore, asynchat, socket, threading, time
+import unittest
+from test import test_support
+
+HOST = "127.0.0.1"
+PORT = 54322
+
+class echo_server(threading.Thread):
+
+    def run(self):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        global PORT
+        PORT = test_support.bind_port(sock, HOST, PORT)
+        sock.listen(1)
+        conn, client = sock.accept()
+        buffer = ""
+        while "\n" not in buffer:
+            data = conn.recv(1)
+            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, terminator):
+        asynchat.async_chat.__init__(self)
+        self.contents = None
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.connect((HOST, PORT))
+        self.set_terminator(terminator)
+        self.buffer = ""
+
+    def handle_connect(self):
+        pass
+        ##print "Connected"
+
+    def collect_incoming_data(self, data):
+        self.buffer = self.buffer + data
+
+    def found_terminator(self):
+        #print "Received:", repr(self.buffer)
+        self.contents = self.buffer
+        self.buffer = ""
+        self.close()
+
+
+class TestAsynchat(unittest.TestCase):
+    def setUp (self):
+        pass
+
+    def tearDown (self):
+        pass
+
+    def test_line_terminator(self):
+        s = echo_server()
+        s.start()
+        time.sleep(1) # Give server time to initialize
+        c = echo_client('\n')
+        c.push("hello ")
+        c.push("world\n")
+        asyncore.loop()
+        s.join()
+
+        self.assertEqual(c.contents, 'hello world')
+
+    def test_numeric_terminator(self):
+        # Try reading a fixed number of bytes
+        s = echo_server()
+        s.start()
+        time.sleep(1) # Give server time to initialize
+        c = echo_client(6L)
+        c.push("hello ")
+        c.push("world\n")
+        asyncore.loop()
+        s.join()
+
+        self.assertEqual(c.contents, 'hello ')
+
+
+def test_main(verbose=None):
+    test_support.run_unittest(TestAsynchat)
+
+if __name__ == "__main__":
+    test_main(verbose=True)