summaryrefslogtreecommitdiff
path: root/Lib/test/test_socket.py
diff options
context:
space:
mode:
authorDave Cole <djc@object-craft.com.au>2004-08-09 04:51:41 +0000
committerDave Cole <djc@object-craft.com.au>2004-08-09 04:51:41 +0000
commit06a005077f8d30d16de1c4cd4428195aef8c745f (patch)
tree0687eabb3c05e850610a88608d68b2ebede69d5f /Lib/test/test_socket.py
parent41c9559665ff8ffce667dba2390f161b476f4040 (diff)
downloadcpython-06a005077f8d30d16de1c4cd4428195aef8c745f.tar.gz
Patch #1003700: Add socketpair function to socket module.
Diffstat (limited to 'Lib/test/test_socket.py')
-rw-r--r--Lib/test/test_socket.py43
1 files changed, 43 insertions, 0 deletions
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index adeca569e4..2dc34aa7b5 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -187,6 +187,28 @@ class SocketConnectedTest(ThreadedTCPSocketTest):
self.serv_conn = None
ThreadedTCPSocketTest.clientTearDown(self)
+class SocketPairTest(unittest.TestCase, ThreadableTest):
+
+ def __init__(self, methodName='runTest'):
+ unittest.TestCase.__init__(self, methodName=methodName)
+ ThreadableTest.__init__(self)
+
+ def setUp(self):
+ self.serv, self.cli = socket.socketpair()
+
+ def tearDown(self):
+ self.serv.close()
+ self.serv = None
+
+ def clientSetUp(self):
+ pass
+
+ def clientTearDown(self):
+ self.cli.close()
+ self.cli = None
+ ThreadableTest.clientTearDown(self)
+
+
#######################################################################
## Begin Tests
@@ -541,6 +563,25 @@ class BasicUDPTest(ThreadedUDPSocketTest):
def _testRecvFrom(self):
self.cli.sendto(MSG, 0, (HOST, PORT))
+class BasicSocketPairTest(SocketPairTest):
+
+ def __init__(self, methodName='runTest'):
+ SocketPairTest.__init__(self, methodName=methodName)
+
+ def testRecv(self):
+ msg = self.serv.recv(1024)
+ self.assertEqual(msg, MSG)
+
+ def _testRecv(self):
+ self.cli.send(MSG)
+
+ def testSend(self):
+ self.serv.send(MSG)
+
+ def _testSend(self):
+ msg = self.cli.recv(1024)
+ self.assertEqual(msg, MSG)
+
class NonBlockingTCPTests(ThreadedTCPSocketTest):
def __init__(self, methodName='runTest'):
@@ -786,6 +827,8 @@ def test_main():
LineBufferedFileObjectClassTestCase,
SmallBufferedFileObjectClassTestCase
])
+ if hasattr(socket, "socketpair"):
+ tests.append(BasicSocketPairTest)
test_support.run_unittest(*tests)
if __name__ == "__main__":