path: root/deprecated/dejagnu/gdb_dejagnu.py

#!/usr/bin/python2

# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""The gdb dejagnu test wrapper."""
import optparse
import os
from os import path
import re
import shutil
import stat
import sys
import tempfile
import time

from cros_utils import command_executer
from cros_utils import logger
from cros_utils import misc

from run_dejagnu import TryAcquireMachine

_VALID_TEST_RESULTS = ['FAIL', 'UNRESOLVED', 'XPASS', 'ERROR', 'UNSUPPORTED',
                       'PASS']


def ProcessArguments(argv):
  """Processing/validating script arguments."""
  parser = optparse.OptionParser(description=(
      'Launches gdb dejagnu test in chroot for chromeos toolchain, compares '
      'the test result with a repository baseline and prints out the result.'),
                                 usage='run_dejagnu options')
  parser.add_option('-c',
                    '--chromeos_root',
                    dest='chromeos_root',
                    help='Required. Specify chromeos root')
  parser.add_option('-m',
                    '--mount',
                    dest='mount',
                    help=('Specify gdb source to mount instead of "auto". '
                          'Under "auto" mode, which is the default - gdb is '
                          'checked out and built automatically at default '
                          'directories. Under "mount" mode '
                          '- the gdb_source is set to "$chromeos_'
                          'root/chroot/usr/local/toolchain_root/gdb", which is '
                          'the mount point for this option value.'))
  parser.add_option('-b',
                    '--board',
                    dest='board',
                    help=('Required. Specify board.'))
  parser.add_option('-r',
                    '--remote',
                    dest='remote',
                    help=('Required. Specify addresses/names of the board, '
                          'seperate each address/name using comma(\',\').'))
  parser.add_option('--cleanup',
                    dest='cleanup',
                    default=None,
                    help=('Optional. Values to this option could be '
                          '\'chroot\' (delete chroot) and '
                          '\'chromeos\' (delete the whole chromeos tree).'))

  options, args = parser.parse_args(argv)

  if not options.chromeos_root:
    raise SyntaxError('Missing argument for --chromeos_root.')
  if not options.remote:
    raise SyntaxError('Missing argument for --remote.')
  if not options.board:
    raise SyntaxError('Missing argument for --board.')
  if options.cleanup == 'mount' and not options.mount:
    raise SyntaxError('--cleanup=\'mount\' not valid unless --mount is given.')
  if options.cleanup and not (options.cleanup == 'mount' or
                              options.cleanup == 'chroot' or
                              options.cleanup == 'chromeos'):
    raise SyntaxError('Invalid option value for --cleanup')

  return options


class DejagnuExecuter(object):
  """The class wrapper for dejagnu test executer."""

  def __init__(self, base_dir, source_dir, chromeos_root, remote, board,
               cleanup):
    self._l = logger.GetLogger()
    self._chromeos_root = chromeos_root
    self._chromeos_chroot = path.join(chromeos_root, 'chroot')

    self._remote = remote
    self._board = board
    ## Compute target from board
    self._target = misc.GetCtargetFromBoard(board, chromeos_root)
    if not self._target:
      raise RuntimeError('Unsupported board "%s"' % board)
    self._executer = command_executer.GetCommandExecuter()
    self._base_dir = base_dir
    self._tmp_abs = None
    self._cleanup = cleanup
    self._sshflag = ('-o StrictHostKeyChecking=no ' + '-o CheckHostIP=no ' +
                     '-o UserKnownHostsFile=$(mktemp) ')

    if source_dir:
      self._source_dir = source_dir
      self._mount_flag = 'USE="mounted_sources"'
      self.MountSource()
    else:
      self._source_dir = None
      self._mount_flag = ''

  def SetupTestingDir(self):
    self._tmp_abs = tempfile.mkdtemp(
        prefix='dejagnu_',
        dir=path.join(self._chromeos_chroot, 'tmp'))
    self._tmp = self._tmp_abs[len(self._chromeos_chroot):]
    self._tmp_testing_rsa = path.join(self._tmp, 'testing_rsa')
    self._tmp_testing_rsa_abs = path.join(self._tmp_abs, 'testing_rsa')

  def PrepareTestingRsaKeys(self):
    if not path.isfile(self._tmp_testing_rsa_abs):
      shutil.copy(
          path.join(self._chromeos_root,
                    'src/scripts/mod_for_test_scripts/ssh_keys/testing_rsa'),
          self._tmp_testing_rsa_abs)
      os.chmod(self._tmp_testing_rsa_abs, stat.S_IRUSR)

  def PrepareTestFiles(self):
    """Prepare site.exp and board exp files."""
    # Create the boards directory.
    os.mkdir('%s/boards' % self._tmp_abs)

    # Generate the chromeos.exp file.
    with open('%s/boards/gdb.exp.in' % self._base_dir, 'r') as template_file:
      content = template_file.read()

    substitutions = dict({
        '__boardname__': self._board,
        '__board_hostname__': self._remote,
        '__tmp_testing_rsa__': self._tmp_testing_rsa,
        '__tmp_dir__': self._tmp
    })
    for pat, sub in substitutions.items():
      content = content.replace(pat, sub)

    board_file_name = '%s/boards/%s.exp' % (self._tmp_abs, self._board)
    with open(board_file_name, 'w') as board_file:
      board_file.write(content)

    # Generate the site file
    with open('%s/site.exp' % self._tmp_abs, 'w') as site_file:
      site_file.write('set target_list "%s"\n' % self._board)

    with open('%s/boards/gdbserver.sh.in' % self._base_dir, 'r') \
        as template_file:
      content = template_file.read()
    substitutions = dict({
        '__board_hostname__': self._remote,
        '__tmp_testing_rsa__': self._tmp_testing_rsa,
        '__tmp_dir__': self._tmp
    })
    for pat, sub in substitutions.items():
      content = content.replace(pat, sub)

    gdbserver_file_name = '%s/boards/gdbserver.sh' % (self._tmp_abs)
    with open(gdbserver_file_name, 'w') as board_file:
      board_file.write(content)

    st = os.stat(gdbserver_file_name)
    os.chmod(gdbserver_file_name, st.st_mode | stat.S_IXGRP | stat.S_IXUSR)

  def PrepareGdb(self):
    self.PrepareGdbDefault()

  def PrepareGdbDefault(self):
    ret = self._executer.ChrootRunCommandWOutput(
        self._chromeos_root, 'equery w cross-%s/gdb' % self._target)[1]
    ret = path.basename(ret.strip())

    matcher = re.match(r'(.*).ebuild', ret)
    if matcher:
      gdb_reversion = matcher.group(1)
    else:
      raise RuntimeError('Failed to get gdb reversion.')
    gdb_version = gdb_reversion.split('-r')[0]
    gdb_portage_dir = '/var/tmp/portage/cross-%s/%s/work' % (self._target,
                                                             gdb_reversion)
    self._gdb_source_dir = path.join(gdb_portage_dir, gdb_version)

    ret = self._executer.ChrootRunCommand(self._chromeos_root, (
        'sudo %s ebuild $(equery w cross-%s/gdb) clean compile' % (
            self._mount_flag, self._target)))
    if ret:
      raise RuntimeError('ebuild gdb failed.')

  def PrepareGdbserver(self):
    self.PrepareGdbserverDefault()

  def PrepareGdbserverDefault(self):
    cmd = ('setup_board --board {0}; '
           '{1} emerge-{0} gdb'.format(self._board, self._mount_flag))
    ret = self._executer.ChrootRunCommand(self._chromeos_root,
                                          cmd,
                                          print_to_console=True)
    if ret:
      raise RuntimeError('ebuild gdbserver failed.')

    cmd = ('scp -i {0}  {1} '
           '/build/{2}/usr/bin/gdbserver root@{3}:/usr/local/bin/'.format(
               self._tmp_testing_rsa, self._sshflag, self._board, self._remote))
    ret = self._executer.ChrootRunCommand(self._chromeos_root,
                                          cmd,
                                          print_to_console=True)
    if ret:
      raise RuntimeError('copy gdbserver failed.')

    if self._mount_flag:
      self.MountSource(unmount=False)

  def Cleanup(self):
    if not self._cleanup:
      return

    if self._cleanup == 'chroot' or self._cleanup == 'chromeos':
      self._l.LogOutput('[Cleanup]: Deleting chroot inside \'{0}\''.format(
          self._chromeos_root))
      command = 'cd %s; cros_sdk --delete' % self._chromeos_root
      rv = self._executer.RunCommand(command)
      if rv:
        self._l.LogWarning('Warning - failed to delete chroot.')
      # Delete .cache - crosbug.com/34956
      command = 'sudo rm -fr %s' % os.path.join(self._chromeos_root, '.cache')
      rv = self._executer.RunCommand(command)
      if rv:
        self._l.LogWarning('Warning - failed to delete \'.cache\'.')

    if self._cleanup == 'chromeos':
      self._l.LogOutput('[Cleanup]: Deleting chromeos tree \'{0}\' ...'.format(
          self._chromeos_root))
      command = 'rm -fr {0}'.format(self._chromeos_root)
      rv = self._executer.RunCommand(command)
      if rv:
        self._l.LogWarning('Warning - failed to remove chromeos tree.')

  def MakeCheck(self):
    cmd = ('ssh -i {0} {1}  root@{2} "reboot && exit"'
           .format(self._tmp_testing_rsa, self._sshflag, self._remote))
    self._executer.ChrootRunCommand(self._chromeos_root, cmd)
    time.sleep(40)

    cmd = ('ssh -i {0} {1}  root@{2} '
           '"iptables -A INPUT -p tcp --dport 1234 -j ACCEPT"'.format(
               self._tmp_testing_rsa, self._sshflag, self._remote))
    self._executer.ChrootRunCommand(self._chromeos_root, cmd)

    cmd = ('cd %s ; '
           'DEJAGNU=%s make check' % (path.join(self._gdb_source_dir, 'gdb'),
                                      path.join(self._tmp, 'site.exp')))
    ret = self._executer.ChrootRunCommand(self._chromeos_root, cmd)
    if ret:
      raise RuntimeError('Make check failed.')

  # This method ensures necessary mount points before executing chroot comamnd.
  def MountSource(self, unmount=False):
    script = os.path.join(self._base_dir, 'build_tc.py')
    if unmount:
      mount = '-u'
    else:
      mount = '-m'
    cmd = ('python {0} --chromeos_root={1} '
           '--gdb_dir={2} --board={3} {4}'.format(script, self._chromeos_root,
                                                  self._source_dir, self._board,
                                                  mount))
    rv = self._executer.RunCommand(cmd)
    if rv:
      raise RuntimeError('Mount source failed.')

  def ResultValidate(self):
    self.PrepareResult()
    result = []
    for key, value in self.base_result.items():
      if 'PASS' not in value:
        continue
      if key not in self.test_result:
        continue
      test_result = self.test_result[key]
      if 'PASS' not in test_result:
        result.append(key)
    return result

  def PrepareResult(self):
    test_output = os.path.join(self._gdb_source_dir, 'gdb', 'testsuite',
                               'gdb.sum')
    test_output = misc.GetOutsideChrootPath(self._chromeos_root, test_output)
    base_output = os.path.join(self._base_dir, 'gdb_baseline', self._target)

    self.test_result = self.ParseResult(test_output)
    self.base_result = self.ParseResult(base_output)

  def ParseResult(self, gdb_sum):
    result = {}
    multi_keys = {}
    with open(gdb_sum) as input_sum:
      for line in input_sum:
        line = line.strip()
        r = line.split(':', 1)
        if r[0] in _VALID_TEST_RESULTS:
          key = r[1]
          if r[1] in result:
            if r[1] in multi_keys:
              multi_keys[r[1]] += 1
            else:
              multi_keys[r[1]] = 2
            key = r[1] + '_____{0}_____'.format(multi_keys[r[1]])
          result[key] = r[0]
    return result


def Main(argv):
  opts = ProcessArguments(argv)
  available_machine = TryAcquireMachine(opts.remote)
  executer = DejagnuExecuter(
      misc.GetRoot(argv[0])[0], opts.mount, opts.chromeos_root,
      available_machine._name, opts.board, opts.cleanup)
  # Return value is a 3- or 4-element tuple
  #   element#1 - exit code
  #   element#2 - stdout
  #   element#3 - stderr
  #   element#4 - exception infor
  # Some other scripts need these detailed information.
  ret = (1, '', '')
  try:
    executer.SetupTestingDir()
    executer.PrepareTestingRsaKeys()
    executer.PrepareTestFiles()
    executer.PrepareGdb()
    executer.PrepareGdbserver()
    executer.MakeCheck()
    result = executer.ResultValidate()
    print result
    if result:
      ret = (1, result, '')
    else:
      ret = (0, '', '')

  except Exception as e:
    # At least log the exception on console.
    print e
    # The #4 element encodes the runtime exception.
    ret = (1, '', '', 'Exception happened during execution: \n' + str(e))
  finally:
    executer.Cleanup()
    return ret


if __name__ == '__main__':
  retval = Main(sys.argv)[0]
  sys.exit(retval)