[Python-checkins] r47202 - in sandbox/trunk/pdb: mconnection.py test/Makefile test/test_mpdb.py test/test_mthread.py

matt.fleming python-checkins at python.org
Mon Jul 3 01:14:31 CEST 2006


Author: matt.fleming
Date: Mon Jul  3 01:14:30 2006
New Revision: 47202

Added:
   sandbox/trunk/pdb/test/test_mthread.py
Modified:
   sandbox/trunk/pdb/mconnection.py
   sandbox/trunk/pdb/test/Makefile
   sandbox/trunk/pdb/test/test_mpdb.py
Log:
A bug in Python 2.5 meant we had to call shutdown on the socket, but it's 
now been fixed. Also move the tests for the thread code into a separate file.


Modified: sandbox/trunk/pdb/mconnection.py
==============================================================================
--- sandbox/trunk/pdb/mconnection.py	(original)
+++ sandbox/trunk/pdb/mconnection.py	Mon Jul  3 01:14:30 2006
@@ -126,7 +126,7 @@
     def disconnect(self):
         if self.output is None or self._sock is None:
             return
-        self.output.shutdown(socket.SHUT_RDWR)
+        self.output.close()
         self._sock.close()
         self._sock = None
         self.listening = False

Modified: sandbox/trunk/pdb/test/Makefile
==============================================================================
--- sandbox/trunk/pdb/test/Makefile	(original)
+++ sandbox/trunk/pdb/test/Makefile	Mon Jul  3 01:14:30 2006
@@ -8,13 +8,16 @@
 
 PY = python
 
-.PHONY: all test test_mpdb test_mconnection
+.PHONY: all test test_mpdb test_mconnection test_mthread
 all: test
 
-test: test_mpdb test_mconnection
+test: test_mpdb test_mconnection test_mthread
 
 test_mpdb:
 	@$(PY) test_mpdb.py
 
 test_mconnection:
 	@$(PY) test_mconnection.py
+
+test_mthread:
+	@$(PY) test_mthread.py

Modified: sandbox/trunk/pdb/test/test_mpdb.py
==============================================================================
--- sandbox/trunk/pdb/test/test_mpdb.py	(original)
+++ sandbox/trunk/pdb/test/test_mpdb.py	Mon Jul  3 01:14:30 2006
@@ -3,7 +3,9 @@
 import os
 import sys
 import socket
+import time
 import thread
+import threading
 import unittest
 
 from test import test_support
@@ -14,7 +16,7 @@
 MAXTRIES = 100
 
 sys.path.append("..")
-from mpdb import MPdb, pdbserver, target
+from mpdb import MPdb, pdbserver, target, Exit
 from mconnection import (MConnectionClientTCP, MConnectionServerTCP,
                          ConnectionFailed)
 
@@ -25,7 +27,7 @@
     if address is None:
         address = __addr__
     client.connection = MConnectionClientTCP()
-
+    
     while True:
         try:
             client.connection.connect(address)
@@ -44,6 +46,24 @@
 
     def msg_nocr(self, msg):
         self.lines.append(msg)
+
+class Pdbserver(threading.Thread, MPdb):
+    def __init__(self):
+        MPdb.__init__(self)
+        threading.Thread.__init__(self)
+        self.botframe = None
+        script = os.path.abspath('thread_script.py')
+        self._sys_argv = [script]
+
+        
+    def run(self):
+        self.do_pdbserver('tcp localhost:8000')
+        while True:
+            try:
+                self.cmdloop()
+            except Exit:
+                break
+        
         
 class TestRemoteDebugging(unittest.TestCase):
     """ Test Case to make sure debugging remotely works properly. """
@@ -100,6 +120,10 @@
 
         self.assertEquals(errmsg, line)
 
+        server.disconnect()
+        while server._sock != None:
+            time.sleep(0.1)
+
     def testRebindOutput(self):
         """ Test rebinding output. """
         self.server = MPdb()
@@ -128,46 +152,24 @@
         f.close()
         self.assertEquals(line, 'help', 'Could not rebind input.')
 
-    def testThread(self):
-        """ Test the thread command. """
-        server = MConnectionServerTCP()
-
-        thread.start_new_thread(server.connect, (__addr__,))
+    def testRestart(self):
+        """ Test the restart command. """
+        server = Pdbserver()
+        server.start()
 
         self.client1 = MPdbTest()
-        connect_to_target(self.client1)
+        self.client1.do_target('tcp localhost:8000')
 
-        # Turn on thread debugging
-        self.client1.onecmd('set thread')
-        line = self.client1.lines[0]
-        self.assertEquals('Thread debugging on\n', line)
-        
-        # Thread with no commands should return current thread
-        self.client1.onecmd('thread')
-        assert 'MainThread' in self.client1.lines[1]
-
-        # 'thread apply' without thread ID should return an error message
-        self.client1.onecmd('thread apply')
-        line = self.client1.lines[2]
-        errmsg = '*** Please specify a Thread ID\n'
-        self.assertEquals(errmsg, line)
-
-        # Need a command to actually apply to a thread
-        self.client1.onecmd('thread apply 49843')
-        line = self.client1.lines[3]
-        errmsg = '*** Please specify a command following the thread ID\n'
-        self.assertEquals(errmsg, line)
-
-        # We've still not started any threads
-        self.client1.onecmd('thread apply 2 info break')
-        line = self.client1.lines[4]
-        errmsg = '*** Thread ID 2 not known.\n'
-        self.assertEquals(errmsg, line)
+        while 'Failed' in self.client1.lines[0]:
+            self.client1.lines = []
+            self.client1.do_target('tcp localhost:8000')
+
+        server.target = 'remote'
+        self.client1.onecmd('restart')
+        self.client1.connection.write('rquit\n')
 
-        self.client1.onecmd('thread')
-        line = self.client1.lines[5]
-        msg = 'Current thread is 1 (<_MainThread(MainThread, started)>)\n'
-        self.assertEquals(msg, line)
+        while server.connection != None:
+            time.sleep(0.1)
 
 def test_main():
     test_support.run_unittest(TestRemoteDebugging)

Added: sandbox/trunk/pdb/test/test_mthread.py
==============================================================================
--- (empty file)
+++ sandbox/trunk/pdb/test/test_mthread.py	Mon Jul  3 01:14:30 2006
@@ -0,0 +1,24 @@
+#!/usr/bin/env python
+
+# Unit tests for the thread debugging code.
+
+import sys
+import unittest
+from test import test_support
+
+sys.path.append('..')
+import mthread
+
+class TestThreadDebugging(unittest.TestCase):
+    def testMthreadInit(self):
+        """ Test the init method of the mthread file. """
+        m = sys.stdout.write
+        e = sys.stderr.write
+        mthread.init(m, e)
+
+def test_main():
+    test_support.run_unittest(TestThreadDebugging)
+
+if __name__ == '__main__':
+    test_main()
+


More information about the Python-checkins mailing list