Convert asynchat test to unittest; exercise the client using a numeric value as the terminator

Lib/test/test_asynchat.py

@@ -2,6 +2,8 @@
 
 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 = 54321
@@ -16,7 +18,7 @@ class echo_server(threading.Thread):
         conn, client = sock.accept()
         buffer = ""
         while "\n" not in buffer:
-            data = conn.recv(10)
+            data = conn.recv(1)
             if not data:
                 break
             buffer = buffer + data
@@ -28,31 +30,61 @@ class echo_server(threading.Thread):
 
 class echo_client(asynchat.async_chat):
 
-    def __init__(self):
+    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("\n")
+        self.set_terminator(terminator)
         self.buffer = ""
 
     def handle_connect(self):
-        print "Connected"
+        pass
+        ##print "Connected"
 
     def collect_incoming_data(self, data):
         self.buffer = self.buffer + data
 
     def found_terminator(self):
-        print "Received:", repr(self.buffer)
+        #print "Received:", repr(self.buffer)
+        self.contents = self.buffer
         self.buffer = ""
         self.close()
 
-def main():
-    s = echo_server()
-    s.start()
-    time.sleep(1) # Give server time to initialize
-    c = echo_client()
-    c.push("hello ")
-    c.push("world\n")
-    asyncore.loop()
 
-main()
+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()
+
+        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()
+
+        self.assertEqual(c.contents, 'hello ')
+
+
+def test_main(verbose=None):
+    test_support.run_unittest(TestAsynchat)
+
+if __name__ == "__main__":
+    test_main(verbose=True)