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

Mon Jun 26 15:29:42 CEST 2006

Author: matt.fleming
Date: Mon Jun 26 15:29:41 2006
New Revision: 47102
Modified:
 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:
Raise ConnectionFailed for invalid address to target() and pdbserver(). Fix unit tests.
Modified: sandbox/trunk/pdb/mconnection.py
==============================================================================

--- sandbox/trunk/pdb/mconnection.py	(original)
+++ sandbox/trunk/pdb/mconnection.py	Mon Jun 26 15:29:41 2006
@@ -104,7 +104,10 @@
 """Set to allow a connection from a client. 'addr' specifies
 the hostname and port combination of the server.
 """
- h,p = addr.split(':')
+ try:
+ h,p = addr.split(':')
+ except ValueError:
+ raise ConnectionFailed, 'Invalid address'
 self.host = h
 self.port = int(p)
 if not self.listening:
@@ -155,7 +158,10 @@
 """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(':')
+ try:
+ h, p = addr.split(':')
+ except ValueError:
+ raise ConnectionFailed, 'Invalid address'
 self.host = h
 self.port = int(p)
 self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
Modified: sandbox/trunk/pdb/mpdb.py
==============================================================================
--- sandbox/trunk/pdb/mpdb.py	(original)
+++ sandbox/trunk/pdb/mpdb.py	Mon Jun 26 15:29:41 2006
@@ -59,7 +59,6 @@
 """ 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
 
@@ -116,7 +115,7 @@
 self.msg("target is %s" % self.target)
 elif 'thread'.startswith(args[0]) and len(args[0])> 2:
 if not self.debug_thread:
- self.msg('Thread debugging is not on.')
+ self.errmsg('Thread debugging is not on.')
 return
 # We need some way to remove old thread instances
 self.msg(self.threads)
@@ -147,7 +146,7 @@
 try:
 from mthread import MTracer
 except ImportError:
- self.msg('Could not import mthread.MTracer')
+ self.errmsg('Could not import mthread.MTracer')
 sys.settrace(None)
 return # Thread not being traced
 
@@ -189,11 +188,11 @@
 try:
 target, addr = args.split(' ')
 except ValueError:
- self.msg('Invalid arguments')
+ self.errmsg('Invalid arguments')
 return
 if self.target == 'remote':
- self.msg('Already connected to a remote machine.')
- return
+ self.errmsg('Already connected to a remote machine.')
+ return 
 if target == 'tcp':
 # TODO: need to save state of current debug session
 if self.connection: self.connection.disconnect()
@@ -203,7 +202,7 @@
 self.connection = MConnectionClientTCP()
 
 except ImportError:
- self.msg('Could not import MConnectionClientTCP')
+ self.errmsg('Could not import MConnectionClientTCP')
 return
 elif target == 'serial':
 if self.connection: self.connection.disconnect()
@@ -212,7 +211,7 @@
 ConnectionFailed)
 self.connection = MConnectionSerial()
 except ImportError:
- self.msg('Could not import MConnectionSerial')
+ self.errmsg('Could not import MConnectionSerial')
 return
 else:
 if '.' in target:
@@ -220,18 +219,23 @@
 # We dynamically load the class for the connection
 base = target[:target.rfind('.')]
 cls = target[target.rfind('.')+1:]
- exec 'from ' + base + ' import (' + cls + ', ConnectionFailed)'
+ try:
+ exec 'from ' + base + ' import (' + cls + ', \
+ ConnectionFailed)'
+ except ImportError:
+ self.errmsg('Unknown target type')
+ return 
 else:
 try:
 __import__(target)
 except ImportError:
- self.msg('Unknown target type')
- return
+ self.errmsg('Unknown target type')
+ return 
 self.connection = eval(target+'()')
 try: 
 self.connection.connect(addr)
 except ConnectionFailed, err:
- self.msg("Failed to connect to %s: (%s)" % (addr, err))
+ self.errmsg("Failed to connect to %s: (%s)" % (addr, err))
 return
 # This interpreter no longer interprets commands but sends
 # them straight across this object's connection to a server.
@@ -245,6 +249,7 @@
 self.msg_nocr(line)
 self.onecmd = self.remote_onecmd
 self.target = 'remote'
+ return
 
 def do_detach(self, args):
 """ Detach a process or file previously attached.
@@ -270,12 +275,12 @@
 
 """
 try:
- target, comm, scriptfile_and_args = args.split(' ')
+ target, comm = args.split(' ')
 except ValueError:
- self.msg('Invalid arguments')
+ self.errmsg('Invalid arguments')
 return
 if self.target == 'remote':
- self.msg('Already connected remotely')
+ self.errmsg('Already connected remotely')
 return
 if target == 'tcp':
 try:
@@ -283,7 +288,7 @@
 ConnectionFailed)
 self.connection = MConnectionServerTCP()
 except ImportError:
- self.msg('Could not load MConnectionServerTCP class')
+ self.errmsg('Could not load MConnectionServerTCP class')
 return
 elif target == 'serial':
 try:
@@ -291,21 +296,29 @@
 ConnectionFailed)
 self.connection = MConnectionSerial()
 except ImportError:
- self.msg('Could not load MConnectionSerial class')
+ self.errmsg('Could not load MConnectionSerial class')
 return
 else:
 if '.' in target:
 base = target[:target.rfind('.')]
 target = target[target.rfind('.')+1:]
- exec 'from ' + base + ' import (' + target + \
+ try:
+ exec 'from ' + base + ' import (' + target + \
 ', ConnectionFailed)'
+ except ImportError:
+ self.errmsg('Unknown protocol')
+ return
 else:
- __import__(target)
+ try:
+ __import__(target)
+ except ImportError:
+ self.errmsg('Unknown protocol')
+ return
 self.connection = eval(target+'()')
 try:
 self.connection.connect(comm)
 except ConnectionFailed, err:
- self.msg("Failed to connect to %s: (%s)" % (comm, err))
+ self.errmsg("Failed to connect to %s: (%s)" % (comm, err))
 return
 self.target = 'remote'
 self._rebind_input(self.connection)
Modified: sandbox/trunk/pdb/test/test_mconnection.py
==============================================================================
--- sandbox/trunk/pdb/test/test_mconnection.py	(original)
+++ sandbox/trunk/pdb/test/test_mconnection.py	Mon Jun 26 15:29:41 2006
@@ -29,7 +29,7 @@
 client.connect(addr)
 # The _sock variable appears when there's a connection
 if client._sock: break
- except socket.error:
+ except ConnectionFailed:
 pass
 
 class TestTCPConnections(unittest.TestCase):
@@ -90,9 +90,8 @@
 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)
-
+ self.assertRaises(ConnectionFailed, 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	Mon Jun 26 15:29:41 2006
@@ -1,5 +1,6 @@
 #!/usr/bin/env python
 
+import os
 import sys
 import socket
 import thread
@@ -9,23 +10,32 @@
 
 # Global vars
 __addr__ = 'localhost:8002'
-script = ""
+script = 'thread_script.py'
 MAXTRIES = 100
 
 sys.path.append("..")
 from mpdb import MPdb, pdbserver, target
+from mconnection import (MConnectionClientTCP, MConnectionServerTCP,
+ ConnectionFailed)
 
 TESTFN = 'tester'
 
+# This provides us with a fine-grain way of connecting to a server
 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)
+ client.connection = MConnectionClientTCP()
 
+ while True:
+ try:
+ client.connection.connect(address)
+ except ConnectionFailed, e:
+ # This is the error message we expect i.e. when the server thread
+ # hasn't started yet. Otherwise it's probably a _real_ error
+ if 'Connection refused' in e: pass
+ else: raise ConnectionFailed, e
+ break
+ 
 class MPdbTest(MPdb):
 def __init__(self):
 MPdb.__init__(self)
@@ -36,14 +46,7 @@
 
 class TestRemoteDebugging(unittest.TestCase):
 """ Test Case to make sure debugging remotely works properly. """
- def setUp(self):
- self.server = MPdb()
- self.client = MPdbTest()
-
 def tearDown(self):
- if self.server.connection:
- self.server.connection.disconnect()
- import os
 if TESTFN in os.listdir('.'):
 os.unlink(TESTFN)
 
@@ -54,25 +57,42 @@
 
 def testPdbserver(self):
 """ Test the pdbserver. """
- thread.start_new_thread(connect_to_target, (self.client,))
- self.server.do_pdbserver('tcp '+__addr__+' '+script)
- self.assertEquals('remote', self.server.target)
+ client = MPdbTest()
+ thread.start_new_thread(connect_to_target, (client,))
+ 
+ self.server1 = MPdb()
+ self.server1.do_pdbserver('tcp '+__addr__)
 
+ self.server1.connection.disconnect()
+
+ self.server2 = MPdbTest()
+ self.server2.do_pdbserver('unknown_protocol unknownhost')
+ line = self.server2.lines[0]
+ self.assertEquals('*** Unknown protocol\n', line)
+
+ 
 def testTarget(self):
 """ Test the target command. """
- addr = 'tcp '+__addr__+' '+script
- thread.start_new_thread(self.client.do_pdbserver, (addr,))
+ server = MConnectionServerTCP()
+ thread.start_new_thread(server.connect, (__addr__,))
+
+ self.client1 = MPdbTest()
+ connect_to_target(self.client1)
 
- # 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.')
+ self.client2 = MPdbTest()
 
+ self.client2.do_target('dlkdlksldkslkd')
+ line = self.client2.lines[0]
+ self.assertEquals('*** Invalid arguments\n', line)
+
+
+ self.client3 = MPdbTest()
+
+ server.disconnect()
 
 def testRebindOutput(self):
 """ Test rebinding output. """
+ self.server = MPdb()
 f = open(TESTFN, 'w+')
 self.server._rebind_output(f)
 self.server.msg('some text')
@@ -83,8 +103,11 @@
 f.close()
 self.assertEquals('some text\n', line, 'Could not rebind output')
 
+ 
 def testRebindInput(self):
 """ Test rebinding input. """
+ self.server = MPdb()
+ 
 f = open(TESTFN, 'w+')
 f.write('help')
 f.close()
@@ -97,8 +120,8 @@
 
 def testTargetRoutine(self):
 """ Test that the top-level target routine works properly. """
- invalid_address = 'tcp ::::::'
- self.assertRaises(ValueError, target, invalid_address)
+ #invalid_address = 'tcp ::::::'
+ pass
 
 
 def test_main():
