[Python-checkins] python/dist/src/Lib/test test_socket.py,1.52,1.53
gvanrossum@users.sourceforge.net
gvanrossum@users.sourceforge.net
Wed, 07 Aug 2002 18:00:33 -0700
Update of /cvsroot/python/python/dist/src/Lib/test
In directory usw-pr-cvs1:/tmp/cvs-serv5948/test
Modified Files:
test_socket.py
Log Message:
Replace docstrings on test functions witrh comments -- then unittest
prints function and module names, which is more informative now that
we repeat some tests in slightly modified subclasses.
Add a test for read() until EOF.
Add test suites for line-buffered (bufsize==1) and a small custom
buffer size (bufsize==2).
Restructure testUnbufferedRead() somewhat to avoid a potentially
infinite loop.
Index: test_socket.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/test/test_socket.py,v
retrieving revision 1.52
retrieving revision 1.53
diff -C2 -d -r1.52 -r1.53
*** test_socket.py 7 Aug 2002 19:02:49 -0000 1.52
--- test_socket.py 8 Aug 2002 01:00:28 -0000 1.53
***************
*** 192,196 ****
def testSocketError(self):
! """Testing that socket module exceptions."""
def raise_error(*args, **kwargs):
raise socket.error
--- 192,196 ----
def testSocketError(self):
! # Testing socket module exceptions
def raise_error(*args, **kwargs):
raise socket.error
***************
*** 207,211 ****
def testCrucialConstants(self):
! """Testing for mission critical constants."""
socket.AF_INET
socket.SOCK_STREAM
--- 207,211 ----
def testCrucialConstants(self):
! # Testing for mission critical constants
socket.AF_INET
socket.SOCK_STREAM
***************
*** 218,222 ****
def testHostnameRes(self):
! """Testing hostname resolution mechanisms."""
hostname = socket.gethostname()
ip = socket.gethostbyname(hostname)
--- 218,222 ----
def testHostnameRes(self):
! # Testing hostname resolution mechanisms
hostname = socket.gethostname()
ip = socket.gethostbyname(hostname)
***************
*** 229,233 ****
def testRefCountGetNameInfo(self):
! """Testing reference count for getnameinfo."""
import sys
if hasattr(sys, "getrefcount"):
--- 229,233 ----
def testRefCountGetNameInfo(self):
! # Testing reference count for getnameinfo
import sys
if hasattr(sys, "getrefcount"):
***************
*** 241,245 ****
def testInterpreterCrash(self):
! """Making sure getnameinfo doesn't crash the interpreter."""
try:
# On some versions, this crashes the interpreter.
--- 241,245 ----
def testInterpreterCrash(self):
! # Making sure getnameinfo doesn't crash the interpreter
try:
# On some versions, this crashes the interpreter.
***************
*** 259,263 ****
def testGetServByName(self):
! """Testing getservbyname()."""
# try a few protocols - not everyone has telnet enabled
found = 0
--- 259,263 ----
def testGetServByName(self):
! # Testing getservbyname()
# try a few protocols - not everyone has telnet enabled
found = 0
***************
*** 279,283 ****
def testDefaultTimeout(self):
! """Testing default timeout."""
# The default timeout should initially be None
self.assertEqual(socket.getdefaulttimeout(), None)
--- 279,283 ----
def testDefaultTimeout(self):
! # Testing default timeout
# The default timeout should initially be None
self.assertEqual(socket.getdefaulttimeout(), None)
***************
*** 309,313 ****
def testSockName(self):
! """Testing getsockname()."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("0.0.0.0", PORT+1))
--- 309,313 ----
def testSockName(self):
! # Testing getsockname()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("0.0.0.0", PORT+1))
***************
*** 316,320 ****
def testGetSockOpt(self):
! """Testing getsockopt()."""
# We know a socket should start without reuse==0
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
--- 316,320 ----
def testGetSockOpt(self):
! # Testing getsockopt()
# We know a socket should start without reuse==0
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
***************
*** 323,327 ****
def testSetSockOpt(self):
! """Testing setsockopt()."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
--- 323,327 ----
def testSetSockOpt(self):
! # Testing setsockopt()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
***************
*** 330,334 ****
def testSendAfterClose(self):
! """testing send() after close() with timeout."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
--- 330,334 ----
def testSendAfterClose(self):
! # testing send() after close() with timeout
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
***************
*** 342,346 ****
def testRecv(self):
! """Testing large receive over TCP."""
msg = self.cli_conn.recv(1024)
self.assertEqual(msg, MSG)
--- 342,346 ----
def testRecv(self):
! # Testing large receive over TCP
msg = self.cli_conn.recv(1024)
self.assertEqual(msg, MSG)
***************
*** 350,354 ****
def testOverFlowRecv(self):
! """Testing receive in chunks over TCP."""
seg1 = self.cli_conn.recv(len(MSG) - 3)
seg2 = self.cli_conn.recv(1024)
--- 350,354 ----
def testOverFlowRecv(self):
! # Testing receive in chunks over TCP
seg1 = self.cli_conn.recv(len(MSG) - 3)
seg2 = self.cli_conn.recv(1024)
***************
*** 360,364 ****
def testRecvFrom(self):
! """Testing large recvfrom() over TCP."""
msg, addr = self.cli_conn.recvfrom(1024)
self.assertEqual(msg, MSG)
--- 360,364 ----
def testRecvFrom(self):
! # Testing large recvfrom() over TCP
msg, addr = self.cli_conn.recvfrom(1024)
self.assertEqual(msg, MSG)
***************
*** 368,372 ****
def testOverFlowRecvFrom(self):
! """Testing recvfrom() in chunks over TCP."""
seg1, addr = self.cli_conn.recvfrom(len(MSG)-3)
seg2, addr = self.cli_conn.recvfrom(1024)
--- 368,372 ----
def testOverFlowRecvFrom(self):
! # Testing recvfrom() in chunks over TCP
seg1, addr = self.cli_conn.recvfrom(len(MSG)-3)
seg2, addr = self.cli_conn.recvfrom(1024)
***************
*** 378,382 ****
def testSendAll(self):
! """Testing sendall() with a 2048 byte string over TCP."""
while 1:
read = self.cli_conn.recv(1024)
--- 378,382 ----
def testSendAll(self):
! # Testing sendall() with a 2048 byte string over TCP
while 1:
read = self.cli_conn.recv(1024)
***************
*** 392,396 ****
def testFromFd(self):
! """Testing fromfd()."""
if not hasattr(socket, "fromfd"):
return # On Windows, this doesn't exist
--- 392,396 ----
def testFromFd(self):
! # Testing fromfd()
if not hasattr(socket, "fromfd"):
return # On Windows, this doesn't exist
***************
*** 404,408 ****
def testShutdown(self):
! """Testing shutdown()."""
msg = self.cli_conn.recv(1024)
self.assertEqual(msg, MSG)
--- 404,408 ----
def testShutdown(self):
! # Testing shutdown()
msg = self.cli_conn.recv(1024)
self.assertEqual(msg, MSG)
***************
*** 418,422 ****
def testSendtoAndRecv(self):
! """Testing sendto() and Recv() over UDP."""
msg = self.serv.recv(len(MSG))
self.assertEqual(msg, MSG)
--- 418,422 ----
def testSendtoAndRecv(self):
! # Testing sendto() and Recv() over UDP
msg = self.serv.recv(len(MSG))
self.assertEqual(msg, MSG)
***************
*** 426,430 ****
def testRecvFrom(self):
! """Testing recvfrom() over UDP."""
msg, addr = self.serv.recvfrom(len(MSG))
self.assertEqual(msg, MSG)
--- 426,430 ----
def testRecvFrom(self):
! # Testing recvfrom() over UDP
msg, addr = self.serv.recvfrom(len(MSG))
self.assertEqual(msg, MSG)
***************
*** 439,443 ****
def testSetBlocking(self):
! """Testing whether set blocking works."""
self.serv.setblocking(0)
start = time.time()
--- 439,443 ----
def testSetBlocking(self):
! # Testing whether set blocking works
self.serv.setblocking(0)
start = time.time()
***************
*** 453,457 ****
def testAccept(self):
! """Testing non-blocking accept."""
self.serv.setblocking(0)
try:
--- 453,457 ----
def testAccept(self):
! # Testing non-blocking accept
self.serv.setblocking(0)
try:
***************
*** 472,476 ****
def testConnect(self):
! """Testing non-blocking connect."""
conn, addr = self.serv.accept()
--- 472,476 ----
def testConnect(self):
! # Testing non-blocking connect
conn, addr = self.serv.accept()
***************
*** 480,484 ****
def testRecv(self):
! """Testing non-blocking recv."""
conn, addr = self.serv.accept()
conn.setblocking(0)
--- 480,484 ----
def testRecv(self):
! # Testing non-blocking recv
conn, addr = self.serv.accept()
conn.setblocking(0)
***************
*** 527,531 ****
def testSmallRead(self):
! """Performing small file read test."""
first_seg = self.serv_file.read(len(MSG)-3)
second_seg = self.serv_file.read(3)
--- 527,531 ----
def testSmallRead(self):
! # Performing small file read test
first_seg = self.serv_file.read(len(MSG)-3)
second_seg = self.serv_file.read(3)
***************
*** 537,549 ****
self.cli_file.flush()
def testUnbufferedRead(self):
! """Performing unbuffered file read test."""
buf = ''
while 1:
char = self.serv_file.read(1)
! self.failIf(not char)
! buf += char
! if buf == MSG:
break
def _testUnbufferedRead(self):
--- 537,558 ----
self.cli_file.flush()
+ def testFullRead(self):
+ # read until EOF
+ msg = self.serv_file.read()
+ self.assertEqual(msg, MSG)
+
+ def _testFullRead(self):
+ self.cli_file.write(MSG)
+ self.cli_file.close()
+
def testUnbufferedRead(self):
! # Performing unbuffered file read test
buf = ''
while 1:
char = self.serv_file.read(1)
! if not char:
break
+ buf += char
+ self.assertEqual(buf, MSG)
def _testUnbufferedRead(self):
***************
*** 552,556 ****
def testReadline(self):
! """Performing file readline test."""
line = self.serv_file.readline()
self.assertEqual(line, MSG)
--- 561,565 ----
def testReadline(self):
! # Performing file readline test
line = self.serv_file.readline()
self.assertEqual(line, MSG)
***************
*** 573,577 ****
def testUnbufferedReadline(self):
! """Read a line, create a new file object, read another line with it."""
line = self.serv_file.readline() # first line
self.assertEqual(line, "A. " + MSG) # first line
--- 582,586 ----
def testUnbufferedReadline(self):
! # Read a line, create a new file object, read another line with it
line = self.serv_file.readline() # first line
self.assertEqual(line, "A. " + MSG) # first line
***************
*** 585,588 ****
--- 594,605 ----
self.cli_file.flush()
+ class LineBufferedFileObjectClassTestCase(FileObjectClassTestCase):
+
+ bufsize = 1 # Default-buffered for reading; line-buffered for writing
+
+
+ class SmallBufferedFileObjectClassTestCase(FileObjectClassTestCase):
+
+ bufsize = 2 # Exercise the buffering code
def test_main():
***************
*** 594,597 ****
--- 611,616 ----
suite.addTest(unittest.makeSuite(FileObjectClassTestCase))
suite.addTest(unittest.makeSuite(UnbufferedFileObjectClassTestCase))
+ suite.addTest(unittest.makeSuite(LineBufferedFileObjectClassTestCase))
+ suite.addTest(unittest.makeSuite(SmallBufferedFileObjectClassTestCase))
test_support.run_suite(suite)