[Python-checkins] r46960 - in sandbox/trunk/pdb: mconnection.py test/tcptest.py

Wed Jun 14 17:52:33 CEST 2006

Author: matt.fleming
Date: Wed Jun 14 17:52:33 2006
New Revision: 46960
Modified:
 sandbox/trunk/pdb/mconnection.py
 sandbox/trunk/pdb/test/tcptest.py
Log:
Move import of socket module nearer to the classes that actually use it in 
mconnection.py. Also change the TCP connection unit test to use 
threading.Timer instead of os.fork.
Modified: sandbox/trunk/pdb/mconnection.py
==============================================================================

--- sandbox/trunk/pdb/mconnection.py	(original)
+++ sandbox/trunk/pdb/mconnection.py	Wed Jun 14 17:52:33 2006
@@ -1,7 +1,6 @@
 """ This file contains all connections that a debugger can
 create.
 """
-import socket
 
 NotImplementedMessage = "This method must be overriden in a subclass"
 
@@ -90,6 +89,8 @@
 MClientConnectionSerial = MServerConnectionSerial
 
 ### This might go in a different file
+import socket
+
 class MServerConnectionTCP(MServerConnectionInterface):
 """This is an implementation of a server class that uses the TCP
 protocol as its means of communication.
Modified: sandbox/trunk/pdb/test/tcptest.py
==============================================================================
--- sandbox/trunk/pdb/test/tcptest.py	(original)
+++ sandbox/trunk/pdb/test/tcptest.py	Wed Jun 14 17:52:33 2006
@@ -2,7 +2,7 @@
 import os
 import sys
 import socket
-import time
+import threading
 import unittest
 
 __addr__ = 'localhost:8000'
@@ -12,26 +12,20 @@
 
 class TestTCPConnections(unittest.TestCase):
 def testClientConnectAndRead(self):
- pid = os.fork()
- if pid == 0:
- # Child
- self.server = MServerConnectionTCP()
- self.server.connect(__addr__)
- self.server.write("good")
- self.server.disconnect()
- else:
- # Parent
- self.client = MClientConnectionTCP()
- for i in range(30):
- try:
- self.client.connect(__addr__)
- break
- except socket.error:
- pass
- time.sleep(5)
- line = self.client.readline()
- self.assertEqual("good", line, "Could not read from server")
- self.client.disconnect()
+ self.server = MServerConnectionTCP()
+ self.client = MClientConnectionTCP()
+ # Create a new thread for the client and delay for 3 seconds
+ t = threading.Timer(3.0, self.client.connect, [__addr__])
+ t.start()
+ # Start server, which waits for a connection. self.client will connect
+ # in a few seconds
+ self.server.connect(__addr__)
+
+ self.server.write("good")
+ line = self.client.readline()
+ self.assertEqual("good", line, "Could not read from server")
+ self.server.disconnect()
+ self.client.disconnect()
 
 if __name__ == '__main__':
 unittest.main()
