[Python-checkins] r47090 - in sandbox/trunk/pdb: README.txt mconnection.py mpdb.py test/test_mconnection.py test/test_mpdb.py

Sat Jun 24 17:52:00 CEST 2006

Author: matt.fleming
Date: Sat Jun 24 17:51:59 2006
New Revision: 47090
Modified:
 sandbox/trunk/pdb/README.txt
 sandbox/trunk/pdb/mconnection.py
 sandbox/trunk/pdb/mpdb.py
 sandbox/trunk/pdb/test/test_mconnection.py
 sandbox/trunk/pdb/test/test_mpdb.py
Log:
Started to fix unit tests and introduce some more to test the top-level
routines. Fixed and unclear statement in README.txt
Modified: sandbox/trunk/pdb/README.txt
==============================================================================

--- sandbox/trunk/pdb/README.txt	(original)
+++ sandbox/trunk/pdb/README.txt	Sat Jun 24 17:51:59 2006
@@ -25,10 +25,9 @@
 * Lightweight and heavyweight mechanism for setting up threads
 - The reason we want a lightweight mechanism is so that we can
 	place mpdb.set_trace() inside a script so that we can debug
-	any threads created with the threading module. It has to be
-	lighweight because the programmer might not want all the features
-	of an MPdb instance, if for example they only care about debugging
-	this one thread instance.
+	the script. It has to be lighweight because the programmer
+	might not want all the features of an MPdb instance, if for example
+	they only care about debugging this one thread instance.
 	- We need a heavyweight mechanism to allow a programmer to inspect
 	and control all threads.
 * Provide a proper top-level methods including, set_trace(), post_mortem(),
Modified: sandbox/trunk/pdb/mconnection.py
==============================================================================
--- sandbox/trunk/pdb/mconnection.py	(original)
+++ sandbox/trunk/pdb/mconnection.py	Sat Jun 24 17:51:59 2006
@@ -155,7 +155,6 @@
 """Connect to the server. 'input' reads data from the
 server. 'output' writes data to the server. Specify the
 address of the server (e.g. host:2020). """
-
 h, p = addr.split(':')
 self.host = h
 self.port = int(p)
Modified: sandbox/trunk/pdb/mpdb.py
==============================================================================
--- sandbox/trunk/pdb/mpdb.py	(original)
+++ sandbox/trunk/pdb/mpdb.py	Sat Jun 24 17:51:59 2006
@@ -53,11 +53,13 @@
 self.waiter = threading.Event()
 self.tracers = []
 self.threads = []
+ self._info_cmds.append('target')
 
 def _rebind_input(self, new_input):
 """ This method rebinds the debugger's input to the object specified
 by 'new_input'.
 """
+ self.stdin.flush()
 self.use_rawinput = False
 self.stdin = new_input
 
@@ -84,7 +86,7 @@
 # The output from the command that we've just sent to the server
 # is returned along with the prompt of that server. So we keep reading
 # until we find our prompt.
- while self.prompt not in ret:
+ while self.local_prompt not in ret:
 ret += self.connection.readline()
 self.msg_nocr(ret)
 return
@@ -110,31 +112,32 @@
 return
 
 args = arg.split()
- if 'target'.startswith(args[0]):
+ if 'target'.startswith(args[0]) and len(args[0]) > 2:
 self.msg("target is %s" % self.target)
- else:
- pydb.Pdb.do_info(self, arg)
-
- def do_thread(self, arg):
- """ Enable thread debugging. """
- # XXX Rocky, how are we subclassing info/set commands? This will do
- # for now.
- if arg == 'info':
+ elif 'thread'.startswith(args[0]) and len(args[0])> 2:
 if not self.debug_thread:
 self.msg('Thread debugging is not on.')
 return
- # XXX We need to remove old threads once the script has finished
- # and currently, we don't.
+ # We need some way to remove old thread instances
 self.msg(self.threads)
 return
- if arg == 'debug':
- # We do not continue with the main thread for as long as there
- # is another thread running. (Will change later so you can choose
- # between threads).
+ else:
+ pydb.Pdb.do_info(self, arg)
+
+
+ def do_set(self, arg):
+ """ Extends pydb do_set() to allow setting thread debugging. """
+ if not arg:
+ pydb.Pdb.do_set(self, arg)
+ return
+
+ args = arg.split()
+ if 'thread'.startswith(args[0]):
 threading.settrace(self.thread_trace_dispatch)
- self.msg('Thread debugging on.')
+ self.msg('Thread debugging on')
 self.debug_thread = True
 return
+ 
 
 def thread_trace_dispatch(self, frame, event, arg):
 """ Create an MTracer object so trace the thread. """
@@ -197,14 +200,12 @@
 try:
 from mconnection import (MConnectionClientTCP,
 ConnectionFailed)
- # Matt - Where are the connection parameters? 
 self.connection = MConnectionClientTCP()
 
 except ImportError:
 self.msg('Could not import MConnectionClientTCP')
 return
 elif target == 'serial':
- # Matt - Where are the connection parameters? 
 if self.connection: self.connection.disconnect()
 try:
 from mconnection import (MConnectionSerial,
@@ -234,10 +235,14 @@
 return
 # This interpreter no longer interprets commands but sends
 # them straight across this object's connection to a server.
- self.prompt = "" # Get our prompt from the server now
+ # XXX: In the remote_onecmd method we use the local_prompt string
+ # to find where the end of the message from the server is. We
+ # really need a way to get the prompt from the server for checking
+ # in remote_onecmd, because it may be different to this client's.
+ self.local_prompt = self.prompt
+ self.prompt = ""
 line = self.connection.readline()
 self.msg_nocr(line)
- self._rebind_output(self.connection)
 self.onecmd = self.remote_onecmd
 self.target = 'remote'
 
@@ -306,39 +311,52 @@
 self._rebind_input(self.connection)
 self._rebind_output(self.connection)
 
-def pdbserver(protocol, address, filename):
+def pdbserver(addr):
 """ This method sets up a pdbserver debugger that allows debuggers
 to connect to 'address' using 'protocol'. The argument 'filename'
 is the name of the file that is being debugged.
 """
- pass
- 
-
-def target(protocol, address):
- """ Connect to a pdbserver at 'address' using 'protocol'. """
- pass
-
+ m = MPdb()
+ position = addr.rfind(' ')
+ mainpyfile = addr[position+1:]
+ m.mainpyfile = mainpyfile
+ m.do_pdbserver(addr)
+ m._runscript(mainpyfile)
+ sys.exit()
+
+def target(addr):
+ """ Connect this debugger to a pdbserver at 'addr'. 'addr' is
+ a protocol-specific address. i.e.
+ tcp = 'tcp mydomainname.com:9876'
+ serial = '/dev/ttyC0'
+ """
+ m = MPdb()
+ # Look Ma, no script!
+ m.do_target(addr)
+ while True:
+ try:
+ m.cmdloop()
+ except:
+ sys.exit()
 
 def main():
 """ Main entry point to this module. """
 opts, args = parse_opts()
- if not opts.scriptname:
- if not args[0]:
- print 'Error: mpdb.py must be called with a script name!'
- sys.exit(1)
- else:
- mainpyfile = args[0]
- if not os.path.exists(mainpyfile):
- print 'Error:', mainpyfile, 'does not exist'
- sys.exit(1)
- if opts.remote:
- if not opts.protocol:
- print 'Protocol must be specified for remote debugging'
+ if opts.target:
+ target(opts.target)
+ elif opts.pdbserver:
+ pdbserver(opts.pdbserver)
+ else:
+ if not opts.scriptname:
+ if not args:
+ print 'Error: mpdb.py must be called with a script name if ' \
+ + '-p or -t switches are not specified.'
+ sys.exit(1)
+ else:
+ mainpyfile = args[0]
+ if not os.path.exists(mainpyfile):
+ print 'Error:', mainpyfile, 'does not exist'
 sys.exit(1)
- if not opts.debugger:
- pdbserver(opts.protocol, opts.address, mainpyfile)
- else:
- target(opts.protocol, opts.address)
 mpdb = MPdb()
 while 1:
 try:
@@ -368,23 +386,14 @@
 parser = OptionParser()
 parser.add_option("-s", "--script", dest="scriptname",
 help="The script to debug")
- parser.add_option("-l", "--local-debugee", dest="local",
- action="store_true",
- help="This script is to be debugged locally, from " + \
- "another process")
- parser.add_option("-p", "--protocol", dest="protocol",
- help="The protocol to use for remote communication")
- parser.add_option("-r", "--remote-debugee", dest="remote",
- action="store_true",
- help="This script is to be debugged by a remote " + \
- "debugger")
- parser.add_option("-a", "--address", dest="address", 
- help="The protocol-specific address of this debugger")
- parser.add_option("-d", "--debugger", dest="debugger",
- action="store_true",
- help="Invoke the debugger.")
+ parser.add_option("-t", "--target", dest="target",
+ help="Specify a target to connect to. The arguments" \
+ + " should be of form, 'protocol address'.")
+ parser.add_option("-p", "--pdbserver", dest="pdbserver",
+ help="Start the debugger and execute the pdbserver " \
+ + "command. The arguments should be of the form," \
+ + " 'protocol address scriptname'.")
 (options, args) = parser.parse_args()
- # We don't currently support any arguments
 return (options,args)
 
 if __name__ == '__main__':
Modified: sandbox/trunk/pdb/test/test_mconnection.py
==============================================================================
--- sandbox/trunk/pdb/test/test_mconnection.py	(original)
+++ sandbox/trunk/pdb/test/test_mconnection.py	Sat Jun 24 17:51:59 2006
@@ -87,6 +87,12 @@
 """(tcp) Test connection refused error. """
 self.assertRaises(ConnectionFailed, self.client.connect, __addr__)
 
+ def testInvalidAddressPortPair(self):
+ """(tcp) Test invald hostname, port pair. """
+ addr = 'localhost 8000'
+ # Rocky: Should this be a ValueError or some other sort of exception?
+ self.assertRaises(ValueError, self.server.connect, addr)
+
 def tearDown(self):
 self.server.disconnect()
 self.client.disconnect()
Modified: sandbox/trunk/pdb/test/test_mpdb.py
==============================================================================
--- sandbox/trunk/pdb/test/test_mpdb.py	(original)
+++ sandbox/trunk/pdb/test/test_mpdb.py	Sat Jun 24 17:51:59 2006
@@ -10,28 +10,22 @@
 # Global vars
 __addr__ = 'localhost:8002'
 script = ""
-g_server = None
-g_client = None
-CONNECTED = False
-# Commands to execute on the server
-cmds = ['info target', 'help', 'quit']
+MAXTRIES = 100
 
 sys.path.append("..")
-from mpdb import MPdb
+from mpdb import MPdb, pdbserver, target
+
+TESTFN = 'tester'
+
+def connect_to_target(client, address=None):
+ if address is None:
+ address = __addr__
+ client.do_target('tcp '+address)
+
+ while 'Failed' in client.lines[0]:
+ client.lines = []
+ client.do_target('tcp '+address)
 
-def doTargetConnect(cmds=None):
- global g_client
- while True:
- try:
- g_client.do_target('tcp '+__addr__)
- if CONNECTED:
- break
- except socket.error:
- pass
- if cmds:
- for c in cmds:
- g_client.onecmd(c)
- 
 class MPdbTest(MPdb):
 def __init__(self):
 MPdb.__init__(self)
@@ -44,36 +38,68 @@
 """ Test Case to make sure debugging remotely works properly. """
 def setUp(self):
 self.server = MPdb()
- global g_client
- g_client = MPdbTest()
+ self.client = MPdbTest()
 
 def tearDown(self):
- global CONNECTED
- self.server.connection.disconnect()
- CONNETED = False
- 
- def testPdbserver(self):
- """ Test the pdbserver command. """
- global CONNECTED
+ if self.server.connection:
+ self.server.connection.disconnect()
+ import os
+ if TESTFN in os.listdir('.'):
+ os.unlink(TESTFN)
+
+# Whilst a lot of the tests below seem to duplicate tests from
+# test_mconnection, we need to make sure that the methods that mpdb provides
+# are not having side effects, and causing tests that should pass to fail
+# and vice versa.
 
- self.server_tid = thread.start_new_thread(doTargetConnect, ())
+ def testPdbserver(self):
+ """ Test the pdbserver. """
+ thread.start_new_thread(connect_to_target, (self.client,))
 self.server.do_pdbserver('tcp '+__addr__+' '+script)
- CONNECTED = True
+ self.assertEquals('remote', self.server.target)
 
- def testCommandsOnServer(self):
- """ Test all supported commands on the pdbserver.
- """
- global CONNECTED, g_client
- 
- self.server_tid = thread.start_new_thread(doTargetConnect, (cmds,))
- self.server.do_pdbserver('tcp '+__addr__+' '+script)
- CONNECTED = True
- 
- # XXX mpdb needs a bottom frame before it exits
- self.server.botframe = None
- self.server.cmdloop()
+ def testTarget(self):
+ """ Test the target command. """
+ addr = 'tcp '+__addr__+' '+script
+ thread.start_new_thread(self.client.do_pdbserver, (addr,))
+
+ # There was a problem with earlier unit tests that they were
+ # working "by coincidence". This ensures that we don't "assume"
+ # the connection is ready until the target has changed from "local"
+ # to "remote".
+ connect_to_target(self.client, __addr__)
+ self.assertEquals('remote', self.client.target, 'Target is wrong.')
+
+
+ def testRebindOutput(self):
+ """ Test rebinding output. """
+ f = open(TESTFN, 'w+')
+ self.server._rebind_output(f)
+ self.server.msg('some text')
+ f.close()
+
+ f = open(TESTFN, 'r')
+ line = f.readline()
+ f.close()
+ self.assertEquals('some text\n', line, 'Could not rebind output')
+
+ def testRebindInput(self):
+ """ Test rebinding input. """
+ f = open(TESTFN, 'w+')
+ f.write('help')
+ f.close()
+
+ f = open(TESTFN, 'r')
+ self.server._rebind_input(f)
+ line = self.server.stdin.readline()
+
+ self.assertEquals(line, 'help', 'Could not rebind input.')
+
+ def testTargetRoutine(self):
+ """ Test that the top-level target routine works properly. """
+ invalid_address = 'tcp ::::::'
+ self.assertRaises(ValueError, target, invalid_address)
 
- self.assertEquals('target is remote\n', g_client.lines[1])
 
 def test_main():
 test_support.run_unittest(TestRemoteDebugging)
