Home | History | Annotate | Download | only in test 
      1 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
      2 # Use of this source code is governed by a BSD-style license that can be
      3 # found in the LICENSE file.
      4 
      5 # This file is based on gdb_rsp.py file from NaCl repository.
      6 
      7 import re
      8 import socket
      9 import time
     10 
     11 
     12 def RspChecksum(data):
     13   checksum = 0
     14   for char in data:
     15     checksum = (checksum + ord(char)) % 0x100
     16   return checksum
     17 
     18 
     19 class GdbRspConnection(object):
     20 
     21   def __init__(self, addr):
     22     self._socket = self._Connect(addr)
     23 
     24   def _Connect(self, addr):
     25     # We have to poll because we do not know when sel_ldr has
     26     # successfully done bind() on the TCP port.  This is inherently
     27     # unreliable.
     28     # TODO(mseaborn): Add a more reliable connection mechanism to
     29     # sel_ldr's debug stub.
     30     timeout_in_seconds = 10
     31     poll_time_in_seconds = 0.1
     32     for i in xrange(int(timeout_in_seconds / poll_time_in_seconds)):
     33       # On Mac OS X, we have to create a new socket FD for each retry.
     34       sock = socket.socket()
     35       try:
     36         sock.connect(addr)
     37       except socket.error:
     38         # Retry after a delay.
     39         time.sleep(poll_time_in_seconds)
     40       else:
     41         return sock
     42     raise Exception('Could not connect to sel_ldr\'s debug stub in %i seconds'
     43                     % timeout_in_seconds)
     44 
     45   def _GetReply(self):
     46     reply = ''
     47     while True:
     48       data = self._socket.recv(1024)
     49       if len(data) == 0:
     50         raise AssertionError('EOF on socket reached with '
     51                              'incomplete reply message: %r' % reply)
     52       reply += data
     53       if '#' in data:
     54         break
     55     match = re.match('\+\$([^#]*)#([0-9a-fA-F]{2})$', reply)
     56     if match is None:
     57       raise AssertionError('Unexpected reply message: %r' % reply)
     58     reply_body = match.group(1)
     59     checksum = match.group(2)
     60     expected_checksum = '%02x' % RspChecksum(reply_body)
     61     if checksum != expected_checksum:
     62       raise AssertionError('Bad RSP checksum: %r != %r' %
     63                            (checksum, expected_checksum))
     64     # Send acknowledgement.
     65     self._socket.send('+')
     66     return reply_body
     67 
     68   # Send an rsp message, but don't wait for or expect a reply.
     69   def RspSendOnly(self, data):
     70     msg = '$%s#%02x' % (data, RspChecksum(data))
     71     return self._socket.send(msg)
     72 
     73   def RspRequest(self, data):
     74     self.RspSendOnly(data)
     75     return self._GetReply()
     76 
     77   def RspInterrupt(self):
     78     self._socket.send('\x03')
     79     return self._GetReply()
     80