#! /usr/bin/python
# 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)