From: <zy...@us...> - 2009-03-08 04:11:01
|
Revision: 6084 http://jython.svn.sourceforge.net/jython/?rev=6084&view=rev Author: zyasoft Date: 2009-03-08 04:10:37 +0000 (Sun, 08 Mar 2009) Log Message: ----------- Re-enabled test_asynchat for all platforms Modified Paths: -------------- trunk/jython/Lib/test/regrtest.py Removed Paths: ------------- trunk/jython/Lib/test/test_asynchat.py Modified: trunk/jython/Lib/test/regrtest.py =================================================================== --- trunk/jython/Lib/test/regrtest.py 2009-03-07 17:57:57 UTC (rev 6083) +++ trunk/jython/Lib/test/regrtest.py 2009-03-08 04:10:37 UTC (rev 6084) @@ -1535,8 +1535,6 @@ self.expected.add('test_mhlib') import platform os_name = platform.java_ver()[3][0] - if os_name == 'Mac OS X' or 'BSD' in os_name: - self.expected.add('test_asynchat') self.valid = True Deleted: trunk/jython/Lib/test/test_asynchat.py =================================================================== --- trunk/jython/Lib/test/test_asynchat.py 2009-03-07 17:57:57 UTC (rev 6083) +++ trunk/jython/Lib/test/test_asynchat.py 2009-03-08 04:10:37 UTC (rev 6084) @@ -1,100 +0,0 @@ -# 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 - -import platform -os_name = platform.java_ver()[3][0] -if os_name == 'Mac OS X' or os_name == 'SunOS' or 'BSD' in os_name \ - or 'Windows' in os_name: - raise test_support.TestSkipped('test_asynchat deadlocks on Jython: ' - 'http://bugs.jython.org/issue1064') - -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) This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site. |