blob: e811348b18b26f3dd6371a9f15bfe0d2ce72b354 [file] [log] [blame]
Guido van Rossum24e4af82002-06-12 19:18:08 +00001#!/usr/bin/env python
Barry Warsawcf3d4b51997-01-03 20:03:32 +00002
Guido van Rossum24e4af82002-06-12 19:18:08 +00003import unittest
Barry Warsaw04f357c2002-07-23 19:04:11 +00004from test import test_support
Barry Warsawcf3d4b51997-01-03 20:03:32 +00005
Barry Warsawcf3d4b51997-01-03 20:03:32 +00006import socket
Guido van Rossum24e4af82002-06-12 19:18:08 +00007import select
Barry Warsawcf3d4b51997-01-03 20:03:32 +00008import time
Guido van Rossum24e4af82002-06-12 19:18:08 +00009import thread, threading
10import Queue
Barry Warsawcf3d4b51997-01-03 20:03:32 +000011
Guido van Rossum24e4af82002-06-12 19:18:08 +000012PORT = 50007
13HOST = 'localhost'
14MSG = 'Michael Gilfix was here\n'
Barry Warsawcf3d4b51997-01-03 20:03:32 +000015
Guido van Rossum24e4af82002-06-12 19:18:08 +000016class SocketTCPTest(unittest.TestCase):
Barry Warsawcf3d4b51997-01-03 20:03:32 +000017
Guido van Rossum24e4af82002-06-12 19:18:08 +000018 def setUp(self):
19 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
20 self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
21 self.serv.bind((HOST, PORT))
22 self.serv.listen(1)
Barry Warsawcf3d4b51997-01-03 20:03:32 +000023
Guido van Rossum24e4af82002-06-12 19:18:08 +000024 def tearDown(self):
25 self.serv.close()
26 self.serv = None
Barry Warsawcf3d4b51997-01-03 20:03:32 +000027
Guido van Rossum24e4af82002-06-12 19:18:08 +000028class SocketUDPTest(unittest.TestCase):
29
30 def setUp(self):
31 self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
32 self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
33 self.serv.bind((HOST, PORT))
34
35 def tearDown(self):
36 self.serv.close()
37 self.serv = None
38
39class ThreadableTest:
Guido van Rossum83ccb4e2002-06-18 18:35:13 +000040 """Threadable Test class
41
42 The ThreadableTest class makes it easy to create a threaded
43 client/server pair from an existing unit test. To create a
44 new threaded class from an existing unit test, use multiple
45 inheritance:
46
47 class NewClass (OldClass, ThreadableTest):
48 pass
49
50 This class defines two new fixture functions with obvious
51 purposes for overriding:
52
53 clientSetUp ()
54 clientTearDown ()
55
56 Any new test functions within the class must then define
57 tests in pairs, where the test name is preceeded with a
58 '_' to indicate the client portion of the test. Ex:
59
60 def testFoo(self):
61 # Server portion
62
63 def _testFoo(self):
64 # Client portion
65
66 Any exceptions raised by the clients during their tests
67 are caught and transferred to the main thread to alert
68 the testing framework.
69
70 Note, the server setup function cannot call any blocking
71 functions that rely on the client thread during setup,
72 unless serverExplicityReady() is called just before
73 the blocking call (such as in setting up a client/server
74 connection and performing the accept() in setUp().
75 """
Guido van Rossum24e4af82002-06-12 19:18:08 +000076
77 def __init__(self):
78 # Swap the true setup function
79 self.__setUp = self.setUp
80 self.__tearDown = self.tearDown
81 self.setUp = self._setUp
82 self.tearDown = self._tearDown
83
Guido van Rossum83ccb4e2002-06-18 18:35:13 +000084 def serverExplicitReady(self):
85 """This method allows the server to explicitly indicate that
86 it wants the client thread to proceed. This is useful if the
87 server is about to execute a blocking routine that is
88 dependent upon the client thread during its setup routine."""
89 self.server_ready.set()
90
Guido van Rossum24e4af82002-06-12 19:18:08 +000091 def _setUp(self):
Guido van Rossum83ccb4e2002-06-18 18:35:13 +000092 self.server_ready = threading.Event()
93 self.client_ready = threading.Event()
Guido van Rossum24e4af82002-06-12 19:18:08 +000094 self.done = threading.Event()
95 self.queue = Queue.Queue(1)
96
97 # Do some munging to start the client test.
Guido van Rossum11ba0942002-06-13 15:07:44 +000098 methodname = self.id()
99 i = methodname.rfind('.')
100 methodname = methodname[i+1:]
101 test_method = getattr(self, '_' + methodname)
Guido van Rossumab659962002-06-12 21:29:43 +0000102 self.client_thread = thread.start_new_thread(
103 self.clientRun, (test_method,))
Guido van Rossum24e4af82002-06-12 19:18:08 +0000104
105 self.__setUp()
Guido van Rossum83ccb4e2002-06-18 18:35:13 +0000106 if not self.server_ready.isSet():
107 self.server_ready.set()
108 self.client_ready.wait()
Guido van Rossum24e4af82002-06-12 19:18:08 +0000109
110 def _tearDown(self):
111 self.__tearDown()
112 self.done.wait()
113
114 if not self.queue.empty():
115 msg = self.queue.get()
116 self.fail(msg)
117
118 def clientRun(self, test_func):
Guido van Rossum83ccb4e2002-06-18 18:35:13 +0000119 self.server_ready.wait()
120 self.client_ready.set()
Guido van Rossum24e4af82002-06-12 19:18:08 +0000121 self.clientSetUp()
122 if not callable(test_func):
123 raise TypeError, "test_func must be a callable function"
124 try:
125 test_func()
126 except Exception, strerror:
127 self.queue.put(strerror)
128 self.clientTearDown()
129
130 def clientSetUp(self):
131 raise NotImplementedError, "clientSetUp must be implemented."
132
133 def clientTearDown(self):
134 self.done.set()
135 thread.exit()
136
137class ThreadedTCPSocketTest(SocketTCPTest, ThreadableTest):
138
139 def __init__(self, methodName='runTest'):
140 SocketTCPTest.__init__(self, methodName=methodName)
141 ThreadableTest.__init__(self)
142
143 def clientSetUp(self):
144 self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
145
146 def clientTearDown(self):
147 self.cli.close()
148 self.cli = None
149 ThreadableTest.clientTearDown(self)
150
151class ThreadedUDPSocketTest(SocketUDPTest, ThreadableTest):
152
153 def __init__(self, methodName='runTest'):
154 SocketUDPTest.__init__(self, methodName=methodName)
155 ThreadableTest.__init__(self)
156
157 def clientSetUp(self):
158 self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
159
160class SocketConnectedTest(ThreadedTCPSocketTest):
161
162 def __init__(self, methodName='runTest'):
163 ThreadedTCPSocketTest.__init__(self, methodName=methodName)
164
165 def setUp(self):
166 ThreadedTCPSocketTest.setUp(self)
Guido van Rossum83ccb4e2002-06-18 18:35:13 +0000167 # Indicate explicitly we're ready for the client thread to
168 # proceed and then perform the blocking call to accept
169 self.serverExplicitReady()
Guido van Rossum24e4af82002-06-12 19:18:08 +0000170 conn, addr = self.serv.accept()
171 self.cli_conn = conn
172
173 def tearDown(self):
174 self.cli_conn.close()
175 self.cli_conn = None
176 ThreadedTCPSocketTest.tearDown(self)
177
178 def clientSetUp(self):
179 ThreadedTCPSocketTest.clientSetUp(self)
180 self.cli.connect((HOST, PORT))
181 self.serv_conn = self.cli
182
183 def clientTearDown(self):
184 self.serv_conn.close()
185 self.serv_conn = None
186 ThreadedTCPSocketTest.clientTearDown(self)
187
188#######################################################################
189## Begin Tests
190
191class GeneralModuleTests(unittest.TestCase):
192
193 def testSocketError(self):
194 """Testing that socket module exceptions."""
195 def raise_error(*args, **kwargs):
196 raise socket.error
197 def raise_herror(*args, **kwargs):
198 raise socket.herror
199 def raise_gaierror(*args, **kwargs):
200 raise socket.gaierror
201 self.failUnlessRaises(socket.error, raise_error,
202 "Error raising socket exception.")
203 self.failUnlessRaises(socket.error, raise_herror,
204 "Error raising socket exception.")
205 self.failUnlessRaises(socket.error, raise_gaierror,
206 "Error raising socket exception.")
207
208 def testCrucialConstants(self):
209 """Testing for mission critical constants."""
210 socket.AF_INET
211 socket.SOCK_STREAM
212 socket.SOCK_DGRAM
213 socket.SOCK_RAW
214 socket.SOCK_RDM
215 socket.SOCK_SEQPACKET
216 socket.SOL_SOCKET
217 socket.SO_REUSEADDR
218
Guido van Rossum654c11e2002-06-13 20:24:17 +0000219 def testHostnameRes(self):
220 """Testing hostname resolution mechanisms."""
221 hostname = socket.gethostname()
222 ip = socket.gethostbyname(hostname)
223 self.assert_(ip.find('.') >= 0, "Error resolving host to ip.")
224 hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
225 all_host_names = [hname] + aliases
226 fqhn = socket.getfqdn()
227 if not fqhn in all_host_names:
228 self.fail("Error testing host resolution mechanisms.")
Barry Warsawcf3d4b51997-01-03 20:03:32 +0000229
Guido van Rossum284a2cf2002-06-12 21:19:40 +0000230 def testRefCountGetNameInfo(self):
Guido van Rossum24e4af82002-06-12 19:18:08 +0000231 """Testing reference count for getnameinfo."""
232 import sys
Guido van Rossum284a2cf2002-06-12 21:19:40 +0000233 if hasattr(sys, "getrefcount"):
Guido van Rossum24e4af82002-06-12 19:18:08 +0000234 try:
235 # On some versions, this loses a reference
236 orig = sys.getrefcount(__name__)
237 socket.getnameinfo(__name__,0)
238 except SystemError:
239 if sys.getrefcount(__name__) <> orig:
240 self.fail("socket.getnameinfo loses a reference")
Barry Warsawcf3d4b51997-01-03 20:03:32 +0000241
Guido van Rossum24e4af82002-06-12 19:18:08 +0000242 def testInterpreterCrash(self):
243 """Making sure getnameinfo doesn't crash the interpreter."""
244 try:
245 # On some versions, this crashes the interpreter.
246 socket.getnameinfo(('x', 0, 0, 0), 0)
247 except socket.error:
248 pass
Barry Warsawcf3d4b51997-01-03 20:03:32 +0000249
Jeremy Hyltonc075e192002-07-25 16:01:12 +0000250 def testNtoH(self):
Jeremy Hyltoncbd5b892002-07-31 15:57:39 +0000251 for func in socket.htonl, socket.ntohl:
252 for i in (0, 1, 0xffff0000, 2L):
253 self.assertEqual(i, func(func(i)))
254
255 biglong = 2**32L - 1
256 swapped = func(biglong)
257 self.assert_(swapped == biglong or swapped == -1)
258 self.assertRaises(OverflowError, func, 2L**34)
Jeremy Hyltonc075e192002-07-25 16:01:12 +0000259
Guido van Rossum24e4af82002-06-12 19:18:08 +0000260 def testGetServByName(self):
Guido van Rossum1c938012002-06-12 21:17:20 +0000261 """Testing getservbyname()."""
Guido van Rossum24e4af82002-06-12 19:18:08 +0000262 if hasattr(socket, 'getservbyname'):
263 socket.getservbyname('telnet', 'tcp')
264 try:
265 socket.getservbyname('telnet', 'udp')
266 except socket.error:
267 pass
268
Guido van Rossum9d0c8ce2002-07-18 17:08:35 +0000269 def testDefaultTimeout(self):
270 """Testing default timeout."""
271 # The default timeout should initially be None
272 self.assertEqual(socket.getdefaulttimeout(), None)
273 s = socket.socket()
274 self.assertEqual(s.gettimeout(), None)
275 s.close()
276
277 # Set the default timeout to 10, and see if it propagates
278 socket.setdefaulttimeout(10)
279 self.assertEqual(socket.getdefaulttimeout(), 10)
280 s = socket.socket()
281 self.assertEqual(s.gettimeout(), 10)
282 s.close()
283
284 # Reset the default timeout to None, and see if it propagates
285 socket.setdefaulttimeout(None)
286 self.assertEqual(socket.getdefaulttimeout(), None)
287 s = socket.socket()
288 self.assertEqual(s.gettimeout(), None)
289 s.close()
290
291 # Check that setting it to an invalid value raises ValueError
292 self.assertRaises(ValueError, socket.setdefaulttimeout, -1)
293
294 # Check that setting it to an invalid type raises TypeError
295 self.assertRaises(TypeError, socket.setdefaulttimeout, "spam")
296
Guido van Rossumb6cc7d22002-07-19 12:46:46 +0000297 # XXX The following don't test module-level functionality...
Guido van Rossum9d0c8ce2002-07-18 17:08:35 +0000298
Guido van Rossum24e4af82002-06-12 19:18:08 +0000299 def testSockName(self):
300 """Testing getsockname()."""
301 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
Guido van Rossum1c938012002-06-12 21:17:20 +0000302 sock.bind(("0.0.0.0", PORT+1))
Guido van Rossum24e4af82002-06-12 19:18:08 +0000303 name = sock.getsockname()
Guido van Rossum1c938012002-06-12 21:17:20 +0000304 self.assertEqual(name, ("0.0.0.0", PORT+1))
Guido van Rossum24e4af82002-06-12 19:18:08 +0000305
306 def testGetSockOpt(self):
307 """Testing getsockopt()."""
308 # We know a socket should start without reuse==0
309 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
310 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
Guido van Rossum733632a2002-06-12 20:46:49 +0000311 self.failIf(reuse != 0, "initial mode is reuse")
Guido van Rossum24e4af82002-06-12 19:18:08 +0000312
313 def testSetSockOpt(self):
314 """Testing setsockopt()."""
315 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
316 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
317 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
Guido van Rossum733632a2002-06-12 20:46:49 +0000318 self.failIf(reuse == 0, "failed to set reuse mode")
Guido van Rossum24e4af82002-06-12 19:18:08 +0000319
Guido van Rossumb6cc7d22002-07-19 12:46:46 +0000320 def testSendAfterClose(self):
321 """testing send() after close() with timeout."""
322 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
323 sock.settimeout(1)
324 sock.close()
325 self.assertRaises(socket.error, sock.send, "spam")
326
Guido van Rossum24e4af82002-06-12 19:18:08 +0000327class BasicTCPTest(SocketConnectedTest):
328
329 def __init__(self, methodName='runTest'):
330 SocketConnectedTest.__init__(self, methodName=methodName)
331
332 def testRecv(self):
333 """Testing large receive over TCP."""
334 msg = self.cli_conn.recv(1024)
Guido van Rossum76489682002-06-12 20:38:30 +0000335 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000336
337 def _testRecv(self):
338 self.serv_conn.send(MSG)
339
340 def testOverFlowRecv(self):
341 """Testing receive in chunks over TCP."""
342 seg1 = self.cli_conn.recv(len(MSG) - 3)
343 seg2 = self.cli_conn.recv(1024)
Guido van Rossumab659962002-06-12 21:29:43 +0000344 msg = seg1 + seg2
Guido van Rossum76489682002-06-12 20:38:30 +0000345 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000346
347 def _testOverFlowRecv(self):
348 self.serv_conn.send(MSG)
349
350 def testRecvFrom(self):
351 """Testing large recvfrom() over TCP."""
352 msg, addr = self.cli_conn.recvfrom(1024)
Guido van Rossum76489682002-06-12 20:38:30 +0000353 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000354
355 def _testRecvFrom(self):
356 self.serv_conn.send(MSG)
357
358 def testOverFlowRecvFrom(self):
359 """Testing recvfrom() in chunks over TCP."""
360 seg1, addr = self.cli_conn.recvfrom(len(MSG)-3)
361 seg2, addr = self.cli_conn.recvfrom(1024)
Guido van Rossumab659962002-06-12 21:29:43 +0000362 msg = seg1 + seg2
Guido van Rossum76489682002-06-12 20:38:30 +0000363 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000364
365 def _testOverFlowRecvFrom(self):
366 self.serv_conn.send(MSG)
367
368 def testSendAll(self):
369 """Testing sendall() with a 2048 byte string over TCP."""
370 while 1:
371 read = self.cli_conn.recv(1024)
372 if not read:
373 break
374 self.assert_(len(read) == 1024, "Error performing sendall.")
375 read = filter(lambda x: x == 'f', read)
376 self.assert_(len(read) == 1024, "Error performing sendall.")
377
378 def _testSendAll(self):
Guido van Rossumab659962002-06-12 21:29:43 +0000379 big_chunk = 'f' * 2048
Guido van Rossum24e4af82002-06-12 19:18:08 +0000380 self.serv_conn.sendall(big_chunk)
381
382 def testFromFd(self):
383 """Testing fromfd()."""
Guido van Rossum8e95ca82002-06-12 20:55:17 +0000384 if not hasattr(socket, "fromfd"):
Guido van Rossum6fb3d5e2002-06-12 20:48:59 +0000385 return # On Windows, this doesn't exist
Guido van Rossum24e4af82002-06-12 19:18:08 +0000386 fd = self.cli_conn.fileno()
387 sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
388 msg = sock.recv(1024)
Guido van Rossum76489682002-06-12 20:38:30 +0000389 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000390
391 def _testFromFd(self):
392 self.serv_conn.send(MSG)
393
394 def testShutdown(self):
395 """Testing shutdown()."""
396 msg = self.cli_conn.recv(1024)
Guido van Rossum76489682002-06-12 20:38:30 +0000397 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000398
399 def _testShutdown(self):
400 self.serv_conn.send(MSG)
401 self.serv_conn.shutdown(2)
402
403class BasicUDPTest(ThreadedUDPSocketTest):
404
405 def __init__(self, methodName='runTest'):
406 ThreadedUDPSocketTest.__init__(self, methodName=methodName)
407
408 def testSendtoAndRecv(self):
409 """Testing sendto() and Recv() over UDP."""
410 msg = self.serv.recv(len(MSG))
Guido van Rossum76489682002-06-12 20:38:30 +0000411 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000412
413 def _testSendtoAndRecv(self):
414 self.cli.sendto(MSG, 0, (HOST, PORT))
415
Guido van Rossum1c938012002-06-12 21:17:20 +0000416 def testRecvFrom(self):
Guido van Rossumdfad1a92002-06-13 15:03:01 +0000417 """Testing recvfrom() over UDP."""
Guido van Rossum24e4af82002-06-12 19:18:08 +0000418 msg, addr = self.serv.recvfrom(len(MSG))
Guido van Rossum76489682002-06-12 20:38:30 +0000419 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000420
Guido van Rossum1c938012002-06-12 21:17:20 +0000421 def _testRecvFrom(self):
Guido van Rossum24e4af82002-06-12 19:18:08 +0000422 self.cli.sendto(MSG, 0, (HOST, PORT))
423
424class NonBlockingTCPTests(ThreadedTCPSocketTest):
425
426 def __init__(self, methodName='runTest'):
427 ThreadedTCPSocketTest.__init__(self, methodName=methodName)
428
429 def testSetBlocking(self):
430 """Testing whether set blocking works."""
431 self.serv.setblocking(0)
432 start = time.time()
433 try:
434 self.serv.accept()
435 except socket.error:
436 pass
437 end = time.time()
438 self.assert_((end - start) < 1.0, "Error setting non-blocking mode.")
439
440 def _testSetBlocking(self):
Barry Warsaw6870bba2001-03-23 17:40:16 +0000441 pass
Barry Warsawcf3d4b51997-01-03 20:03:32 +0000442
Guido van Rossum24e4af82002-06-12 19:18:08 +0000443 def testAccept(self):
444 """Testing non-blocking accept."""
445 self.serv.setblocking(0)
Guido van Rossum41360a41998-03-26 19:42:58 +0000446 try:
Guido van Rossum24e4af82002-06-12 19:18:08 +0000447 conn, addr = self.serv.accept()
448 except socket.error:
449 pass
450 else:
451 self.fail("Error trying to do non-blocking accept.")
452 read, write, err = select.select([self.serv], [], [])
453 if self.serv in read:
454 conn, addr = self.serv.accept()
455 else:
456 self.fail("Error trying to do accept after select.")
Guido van Rossum67f7a382002-06-06 21:08:16 +0000457
Guido van Rossum24e4af82002-06-12 19:18:08 +0000458 def _testAccept(self):
Guido van Rossumb6cc7d22002-07-19 12:46:46 +0000459 time.sleep(0.1)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000460 self.cli.connect((HOST, PORT))
461
462 def testConnect(self):
463 """Testing non-blocking connect."""
Guido van Rossum24e4af82002-06-12 19:18:08 +0000464 conn, addr = self.serv.accept()
465
466 def _testConnect(self):
Guido van Rossum7b8bac12002-06-13 16:07:04 +0000467 self.cli.settimeout(10)
468 self.cli.connect((HOST, PORT))
Guido van Rossum24e4af82002-06-12 19:18:08 +0000469
470 def testRecv(self):
471 """Testing non-blocking recv."""
472 conn, addr = self.serv.accept()
473 conn.setblocking(0)
474 try:
475 msg = conn.recv(len(MSG))
476 except socket.error:
477 pass
478 else:
479 self.fail("Error trying to do non-blocking recv.")
480 read, write, err = select.select([conn], [], [])
481 if conn in read:
482 msg = conn.recv(len(MSG))
Guido van Rossum76489682002-06-12 20:38:30 +0000483 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000484 else:
485 self.fail("Error during select call to non-blocking socket.")
486
487 def _testRecv(self):
488 self.cli.connect((HOST, PORT))
Guido van Rossumb6cc7d22002-07-19 12:46:46 +0000489 time.sleep(0.1)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000490 self.cli.send(MSG)
491
492class FileObjectClassTestCase(SocketConnectedTest):
493
494 def __init__(self, methodName='runTest'):
495 SocketConnectedTest.__init__(self, methodName=methodName)
496
497 def setUp(self):
498 SocketConnectedTest.setUp(self)
499 self.serv_file = socket._fileobject(self.cli_conn, 'rb', 8192)
500
501 def tearDown(self):
502 self.serv_file.close()
503 self.serv_file = None
504 SocketConnectedTest.tearDown(self)
505
506 def clientSetUp(self):
507 SocketConnectedTest.clientSetUp(self)
508 self.cli_file = socket._fileobject(self.serv_conn, 'rb', 8192)
509
510 def clientTearDown(self):
511 self.cli_file.close()
512 self.cli_file = None
513 SocketConnectedTest.clientTearDown(self)
514
515 def testSmallRead(self):
516 """Performing small file read test."""
517 first_seg = self.serv_file.read(len(MSG)-3)
518 second_seg = self.serv_file.read(3)
Guido van Rossumab659962002-06-12 21:29:43 +0000519 msg = first_seg + second_seg
Guido van Rossum76489682002-06-12 20:38:30 +0000520 self.assertEqual(msg, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000521
522 def _testSmallRead(self):
523 self.cli_file.write(MSG)
524 self.cli_file.flush()
525
526 def testUnbufferedRead(self):
527 """Performing unbuffered file read test."""
528 buf = ''
529 while 1:
530 char = self.serv_file.read(1)
Guido van Rossum76489682002-06-12 20:38:30 +0000531 self.failIf(not char)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000532 buf += char
533 if buf == MSG:
534 break
535
536 def _testUnbufferedRead(self):
537 self.cli_file.write(MSG)
538 self.cli_file.flush()
539
540 def testReadline(self):
541 """Performing file readline test."""
542 line = self.serv_file.readline()
Guido van Rossum76489682002-06-12 20:38:30 +0000543 self.assertEqual(line, MSG)
Guido van Rossum24e4af82002-06-12 19:18:08 +0000544
545 def _testReadline(self):
546 self.cli_file.write(MSG)
547 self.cli_file.flush()
548
Guido van Rossumb995eb72002-07-31 16:08:40 +0000549def test_main():
Guido van Rossum24e4af82002-06-12 19:18:08 +0000550 suite = unittest.TestSuite()
551 suite.addTest(unittest.makeSuite(GeneralModuleTests))
552 suite.addTest(unittest.makeSuite(BasicTCPTest))
553 suite.addTest(unittest.makeSuite(BasicUDPTest))
554 suite.addTest(unittest.makeSuite(NonBlockingTCPTests))
555 suite.addTest(unittest.makeSuite(FileObjectClassTestCase))
556 test_support.run_suite(suite)
557
558if __name__ == "__main__":
Guido van Rossumb995eb72002-07-31 16:08:40 +0000559 test_main()