[Python-checkins] r47090 - in sandbox/trunk/pdb: README.txt mconnection.py mpdb.py test/test_mconnection.py test/test_mpdb.py
matt.fleming
python-checkins at python.org
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)
More information about the Python-checkins
mailing list