diff options
author | Andrew M. Kuchling <amk@amk.ca> | 2005-06-09 14:56:31 (GMT) |
---|---|---|
committer | Andrew M. Kuchling <amk@amk.ca> | 2005-06-09 14:56:31 (GMT) |
commit | 5ac2534bbc92b21dec5a69e78f42fe654cbc8b2b (patch) | |
tree | 605b1307587a932a0529f86c86eab38fbf8c5eb8 /Lib/test | |
parent | 01cb47b59c434ff51d272b9399c7806bd6b1c0e9 (diff) | |
download | cpython-5ac2534bbc92b21dec5a69e78f42fe654cbc8b2b.zip cpython-5ac2534bbc92b21dec5a69e78f42fe654cbc8b2b.tar.gz cpython-5ac2534bbc92b21dec5a69e78f42fe654cbc8b2b.tar.bz2 |
Convert asynchat test to unittest; exercise the client using a numeric value as the terminator
Diffstat (limited to 'Lib/test')
-rw-r--r-- | Lib/test/test_asynchat.py | 60 |
1 files changed, 46 insertions, 14 deletions
diff --git a/Lib/test/test_asynchat.py b/Lib/test/test_asynchat.py index e91c572..d10e389 100644 --- a/Lib/test/test_asynchat.py +++ b/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) |