From: <pj...@us...> - 2008-11-17 20:06:01
|
Revision: 5583 http://jython.svn.sourceforge.net/jython/?rev=5583&view=rev Author: pjenvey Date: 2008-11-17 20:05:55 +0000 (Mon, 17 Nov 2008) Log Message: ----------- o allow Socket{TCP,UDP}Test HOST/PORT to be overriden per class o use a different port than test_socket for test_select as test_socket sockopt tests don't REUSEADDR Modified Paths: -------------- trunk/jython/Lib/test/test_select.py trunk/jython/Lib/test/test_socket.py Modified: trunk/jython/Lib/test/test_select.py =================================================================== --- trunk/jython/Lib/test/test_select.py 2008-11-16 22:52:56 UTC (rev 5582) +++ trunk/jython/Lib/test/test_select.py 2008-11-17 20:05:55 UTC (rev 5583) @@ -1,21 +1,18 @@ """ AMAK: 20050515: This module is the test_select.py from cpython 2.4, ported to jython + unittest """ - -try: - object -except NameError: - class object: pass - import errno +import os import select import socket - -import os import sys -from test import test_support +import test_socket import unittest +from test import test_support +HOST = test_socket.HOST +PORT = test_socket.PORT + 100 + class SelectWrapper: def __init__(self): @@ -140,11 +137,11 @@ # # using the test_socket thread based server/client management, for convenience. # - -import test_socket - class ThreadedPollClientSocket(test_socket.ThreadedTCPSocketTest): + HOST = HOST + PORT = PORT + def testSocketRegisteredBeforeConnected(self): self.cli_conn = self.serv.accept() @@ -158,7 +155,7 @@ self.failIf(self.cli in result_sockets, "Unconnected client socket should not have been selectable") # Now connect the socket, but DO NOT register it again self.cli.setblocking(0) - self.cli.connect( (test_socket.HOST, test_socket.PORT) ) + self.cli.connect( (self.HOST, self.PORT) ) # Now poll again, to check that the poll object has recognised that the socket is now connected result_list = poll_object.poll(timeout) result_sockets = [r[0] for r in result_list] @@ -169,7 +166,7 @@ def _testSocketMustBeNonBlocking(self): self.cli.setblocking(1) - self.cli.connect( (test_socket.HOST, test_socket.PORT) ) + self.cli.connect( (self.HOST, self.PORT) ) timeout = 1000 # milliseconds poll_object = select.poll() try: Modified: trunk/jython/Lib/test/test_socket.py =================================================================== --- trunk/jython/Lib/test/test_socket.py 2008-11-16 22:52:56 UTC (rev 5582) +++ trunk/jython/Lib/test/test_socket.py 2008-11-17 20:05:55 UTC (rev 5583) @@ -28,10 +28,13 @@ class SocketTCPTest(unittest.TestCase): + HOST = HOST + PORT = PORT + def setUp(self): self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - self.serv.bind((HOST, PORT)) + self.serv.bind((self.HOST, self.PORT)) self.serv.listen(1) def tearDown(self): @@ -40,10 +43,13 @@ class SocketUDPTest(unittest.TestCase): + HOST = HOST + PORT = PORT + def setUp(self): self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - self.serv.bind((HOST, PORT)) + self.serv.bind((self.HOST, self.PORT)) def tearDown(self): self.serv.close() @@ -196,7 +202,7 @@ def clientSetUp(self): ThreadedTCPSocketTest.clientSetUp(self) - self.cli.connect((HOST, PORT)) + self.cli.connect((self.HOST, self.PORT)) self.serv_conn = self.cli def clientTearDown(self): @@ -796,7 +802,7 @@ self.assertEqual(msg, MSG) def _testSendtoAndRecv(self): - self.cli.sendto(MSG, 0, (HOST, PORT)) + self.cli.sendto(MSG, 0, (self.HOST, self.PORT)) def testSendtoAndRecvTimeoutMode(self): # Need to test again in timeout mode, which follows @@ -807,7 +813,7 @@ def _testSendtoAndRecvTimeoutMode(self): self.cli.settimeout(10) - self.cli.sendto(MSG, 0, (HOST, PORT)) + self.cli.sendto(MSG, 0, (self.HOST, self.PORT)) def testRecvFrom(self): # Testing recvfrom() over UDP @@ -815,7 +821,7 @@ self.assertEqual(msg, MSG) def _testRecvFrom(self): - self.cli.sendto(MSG, 0, (HOST, PORT)) + self.cli.sendto(MSG, 0, (self.HOST, self.PORT)) def testRecvFromTimeoutMode(self): # Need to test again in timeout mode, which follows @@ -826,7 +832,7 @@ def _testRecvFromTimeoutMode(self): self.cli.settimeout(10) - self.cli.sendto(MSG, 0, (HOST, PORT)) + self.cli.sendto(MSG, 0, (self.HOST, self.PORT)) def testSendtoEightBitSafe(self): # This test is necessary because java only supports signed bytes @@ -834,7 +840,7 @@ self.assertEqual(msg, EIGHT_BIT_MSG) def _testSendtoEightBitSafe(self): - self.cli.sendto(EIGHT_BIT_MSG, 0, (HOST, PORT)) + self.cli.sendto(EIGHT_BIT_MSG, 0, (self.HOST, self.PORT)) def testSendtoEightBitSafeTimeoutMode(self): # Need to test again in timeout mode, which follows @@ -845,7 +851,7 @@ def _testSendtoEightBitSafeTimeoutMode(self): self.cli.settimeout(10) - self.cli.sendto(EIGHT_BIT_MSG, 0, (HOST, PORT)) + self.cli.sendto(EIGHT_BIT_MSG, 0, (self.HOST, self.PORT)) class UDPBroadcastTest(ThreadedUDPSocketTest): @@ -854,13 +860,13 @@ self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) def testBroadcast(self): - self.serv.bind( ("<broadcast>", PORT) ) + self.serv.bind( ("<broadcast>", self.PORT) ) msg = self.serv.recv(len(EIGHT_BIT_MSG)) self.assertEqual(msg, EIGHT_BIT_MSG) def _testBroadcast(self): self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - self.cli.sendto(EIGHT_BIT_MSG, ("<broadcast>", PORT) ) + self.cli.sendto(EIGHT_BIT_MSG, ("<broadcast>", self.PORT) ) class BasicSocketPairTest(SocketPairTest): @@ -927,7 +933,7 @@ def _testAcceptConnection(self): # Make a connection to the server - self.cli.connect((HOST, PORT)) + self.cli.connect((self.HOST, self.PORT)) # # AMAK: 20070311 @@ -942,7 +948,7 @@ def _testBlockingConnect(self): # Testing blocking connect self.cli.settimeout(10) - self.cli.connect((HOST, PORT)) + self.cli.connect((self.HOST, self.PORT)) def testNonBlockingConnect(self): # Testing non-blocking connect @@ -951,7 +957,7 @@ def _testNonBlockingConnect(self): # Testing non-blocking connect self.cli.setblocking(0) - result = self.cli.connect_ex((HOST, PORT)) + result = self.cli.connect_ex((self.HOST, self.PORT)) rfds, wfds, xfds = select.select([], [self.cli], []) self.failUnless(self.cli in wfds) try: @@ -970,13 +976,13 @@ def _testConnectWithLocalBind(self): # Testing blocking connect with local bind - cli_port = PORT - 1 + cli_port = self.PORT - 1 while True: # Keep trying until a local port is available self.cli.settimeout(1) - self.cli.bind( (HOST, cli_port) ) + self.cli.bind( (self.HOST, cli_port) ) try: - self.cli.connect((HOST, PORT)) + self.cli.connect((self.HOST, self.PORT)) break except socket.error, se: # cli_port is in use (maybe in TIME_WAIT state from a @@ -1004,7 +1010,7 @@ self.fail("Non-blocking socket with data should been in read list.") def _testRecvData(self): - self.cli.connect((HOST, PORT)) + self.cli.connect((self.HOST, self.PORT)) self.cli.send(MSG) def testRecvNoData(self): @@ -1019,7 +1025,7 @@ self.fail("Non-blocking recv of no data should have raised socket.error.") def _testRecvNoData(self): - self.cli.connect((HOST, PORT)) + self.cli.connect((self.HOST, self.PORT)) time.sleep(0.1) class NonBlockingUDPTests(ThreadedUDPSocketTest): pass @@ -1067,7 +1073,7 @@ self.cli_file = self.cli.makefile('wb') self.cli_file.close() try: - self.cli.sendto(MSG, 0, (HOST, PORT)) + self.cli.sendto(MSG, 0, (self.HOST, self.PORT)) except Exception, x: self.fail("Closing file wrapper appears to have closed underlying socket: %s" % str(x)) @@ -1079,7 +1085,7 @@ def _testCloseSocketDoesNotCloseFile(self): try: - self.cli.sendto(MSG, 0, (HOST, PORT)) + self.cli.sendto(MSG, 0, (self.HOST, self.PORT)) except Exception, x: self.fail("Closing file wrapper appears to have closed underlying socket: %s" % str(x)) @@ -1488,7 +1494,7 @@ pass def _testUnicodeHostname(self): - self.cli.connect((unicode(HOST), PORT)) + self.cli.connect((unicode(self.HOST), self.PORT)) class TestInvalidUsage(unittest.TestCase): This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site. |