Add ctest that calls the au test harness.

Change-Id: Ief14528b7e98d792da4be39976ff0fe48219f93b

BUG=6913
TEST=Ran image_to_vm, cros_au_test_harness, and ctest using internal urls.

Review URL: http://codereview.chromium.org/3606008

bin/cros_make_image_bootable

@@ -100,6 +100,9 @@ DEFINE_string espfs_mountpoint "/tmp/espfs" \
 DEFINE_boolean use_dev_keys ${FLAGS_FALSE} \
   "Use developer keys for signing. (Default: false)"
 
+# TODO(sosa):  Remove once known images no longer use this in their config.
+DEFINE_string arm_extra_bootargs "" "DEPRECATED FLAG.  Do not use."
+
 # Parse the boot.desc and any overrides
 eval set -- "${BOOT_DESC} ${FLAG_OVERRIDES}"
 FLAGS "${@}" || exit 1

bin/ctest

@@ -0,0 +1 @@
+ctest.py

bin/ctest.py

@@ -0,0 +1,195 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2010 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.
+
+"""Wrapper for tests that are run on builders."""
+
+import fileinput
+import optparse
+import os
+import sys
+import urllib
+
+sys.path.append(os.path.join(os.path.dirname(__file__), '../lib'))
+from cros_build_lib import Info, RunCommand, ReinterpretPathForChroot
+
+_IMAGE_TO_EXTRACT = 'chromiumos_test_image.bin'
+
+
+def ModifyBootDesc(download_folder, redirect_file=None):
+  """Modifies the boot description of a downloaded image to work with path.
+
+  The default boot.desc from another system is specific to the directory
+  it was created in.  This modifies the boot description to be compatiable
+  with the download folder.
+
+  Args:
+    download_folder: Absoulte path to the download folder.
+    redirect_file:  For testing.  Where to copy new boot desc.
+  """
+  boot_desc_path = os.path.join(download_folder, 'boot.desc')
+  in_chroot_folder = ReinterpretPathForChroot(download_folder)
+
+  for line in fileinput.input(boot_desc_path, inplace=1):
+    # Has to be done here to get changes to sys.stdout from fileinput.input.
+    if not redirect_file:
+      redirect_file = sys.stdout
+    split_line = line.split('=')
+    if len(split_line) > 1:
+      var_part = split_line[0]
+      potential_path = split_line[1].replace('"', '').strip()
+
+      if potential_path.startswith('/home') and not 'output_dir' in var_part:
+        new_path = os.path.join(in_chroot_folder,
+                                os.path.basename(potential_path))
+        new_line = '%s="%s"' % (var_part, new_path)
+        Info('Replacing line %s with %s' % (line, new_line))
+        redirect_file.write('%s\n' % new_line)
+        continue
+      elif 'output_dir' in var_part:
+        # Special case for output_dir.
+        new_line = '%s="%s"' % (var_part, in_chroot_folder)
+        Info('Replacing line %s with %s' % (line, new_line))
+        redirect_file.write('%s\n' % new_line)
+        continue
+
+    # Line does not need to be modified.
+    redirect_file.write(line)
+
+  fileinput.close()
+
+
+def GetLatestZipUrl(board, channel, latest_url_base, zip_server_base):
+  """Returns the url of the latest image zip for the given arguments.
+
+  Args:
+    board: board for the image zip.
+    channel: channel for the image zip.
+    latest_url_base: base url for latest links.
+    zip_server_base:  base url for zipped images.
+  """
+  # Grab the latest image info.
+  latest_file_url = os.path.join(latest_url_base, channel,
+                                 'LATEST-%s' % board)
+  latest_image_file = urllib.urlopen(latest_file_url)
+  latest_image = latest_image_file.read()
+  latest_image_file.close()
+
+  # Convert bin.gz into zip.
+  latest_image = latest_image.replace('.bin.gz', '.zip')
+  version = latest_image.split('-')[1]
+  zip_base = os.path.join(zip_server_base, channel, board)
+  return os.path.join(zip_base, version, latest_image)
+
+
+def GrabZipAndExtractImage(zip_url, download_folder, image_name) :
+  """Downloads the zip and extracts the given image.
+
+  Doesn't re-download if matching version found already in download folder.
+  Args:
+    zip_url - url for the image.
+    download_folder - download folder to store zip file and extracted images.
+    image_name - name of the image to extract from the zip file.
+  """
+  zip_path = os.path.join(download_folder, 'image.zip')
+  versioned_url_path = os.path.join(download_folder, 'download_url')
+  found_cached = False
+
+  if os.path.exists(versioned_url_path):
+    fh = open(versioned_url_path)
+    version_url = fh.read()
+    fh.close()
+
+    if version_url == zip_url and os.path.exists(os.path.join(download_folder,
+                                                 image_name)):
+      Info('Using cached %s' % image_name)
+      found_cached = True
+
+  if not found_cached:
+    Info('Downloading %s' % zip_url)
+    RunCommand(['rm', '-rf', download_folder], print_cmd=False)
+    os.mkdir(download_folder)
+    urllib.urlretrieve(zip_url, zip_path)
+
+    # Using unzip because python implemented unzip in native python so
+    # extraction is really slow.
+    Info('Unzipping image %s' % image_name)
+    RunCommand(['unzip', '-d', download_folder, zip_path],
+               print_cmd=False, error_message='Failed to download %s' % zip_url)
+
+    ModifyBootDesc(download_folder)
+
+    # Put url in version file so we don't have to do this every time.
+    fh = open(versioned_url_path, 'w+')
+    fh.write(zip_url)
+    fh.close()
+
+
+def RunAUTestHarness(board, channel, latest_url_base, zip_server_base):
+  """Runs the auto update test harness.
+
+  The auto update test harness encapsulates testing the auto-update mechanism
+  for the latest image against the latest official image from the channel.  This
+  also tests images with suite_Smoke (built-in as part of its verification
+  process).
+
+  Args:
+    board: the board for the latest image.
+    channel: the channel to run the au test harness against.
+    latest_url_base: base url for getting latest links.
+    zip_server_base:  base url for zipped images.
+  """
+  crosutils_root = os.path.join(os.path.dirname(__file__), '..')
+  download_folder = os.path.abspath('latest_download')
+  zip_url = GetLatestZipUrl(board, channel, latest_url_base, zip_server_base)
+  GrabZipAndExtractImage(zip_url, download_folder, _IMAGE_TO_EXTRACT)
+
+  # Tests go here.
+  latest_image = RunCommand(['./get_latest_image.sh', '--board=%s' % board],
+                            cwd=crosutils_root, redirect_stdout=True,
+                            print_cmd=True)
+
+  RunCommand(['bin/cros_au_test_harness',
+              '--base_image=%s' % os.path.join(download_folder,
+                                                 _IMAGE_TO_EXTRACT),
+              '--target_image=%s' % latest_image,
+              '--board=%s' % board], cwd=crosutils_root)
+
+
+def main():
+  parser = optparse.OptionParser()
+  parser.add_option('-b', '--board',
+                    help='board for the image to compare against.')
+  parser.add_option('-c', '--channel',
+                    help='channel for the image to compare against.')
+  parser.add_option('-l', '--latestbase',
+                    help='Base url for latest links.')
+  parser.add_option('-z', '--zipbase',
+                    help='Base url for hosted images.')
+  # Set the usage to include flags.
+  parser.set_usage(parser.format_help())
+  (options, args) = parser.parse_args()
+
+  if args:
+    parser.error('Extra args found %s.' % args)
+
+  if not options.board:
+    parser.error('Need board for image to compare against.')
+
+  if not options.channel:
+    parser.error('Need channel for image to compare against.')
+
+  if not options.latestbase:
+    parser.error('Need latest url base to get images.')
+
+  if not options.zipbase:
+    parser.error('Need zip url base to get images.')
+
+  RunAUTestHarness(options.board, options.channel, options.latestbase,
+                   options.zipbase)
+
+
+if __name__ == '__main__':
+  main()

bin/ctest_unittest.py

@@ -0,0 +1,165 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2010 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.
+
+"""Unit tests for ctest."""
+
+import ctest
+import mox
+import os
+import unittest
+import urllib
+
+_TEST_BOOT_DESC = """
+  --arch="x86"
+  --output_dir="/home/chrome-bot/0.8.70.5-a1"
+  --espfs_mountpoint="/home/chrome-bot/0.8.70.5-a1/esp"
+  --enable_rootfs_verification
+"""
+
+class CrosTestTest(mox.MoxTestBase):
+  """Test class for CTest."""
+
+  def setUp(self):
+    mox.MoxTestBase.setUp(self)
+    self.board = 'test-board'
+    self.channel = 'test-channel'
+    self.version = '1.2.3.4.5'
+    self.revision = '7ghfa9999-12345'
+    self.image_name = 'TestOS-%s-%s' % (self.version, self.revision)
+    self.download_folder = 'test_folder'
+    self.latestbase = 'http://test-latest/TestOS'
+    self.zipbase = 'http://test-zips/archive/TestOS'
+    self.image_url = '%s/%s/%s/%s/%s.zip' % (self.zipbase, self.channel,
+                                             self.board, self.version,
+                                             self.image_name)
+
+  def testModifyBootDesc(self):
+    """Tests to make sure we correctly modify a boot desc."""
+    in_chroot_path = ctest.ReinterpretPathForChroot(os.path.abspath(
+        self.download_folder))
+    self.mox.StubOutWithMock(__builtins__, 'open')
+    self.mox.StubOutWithMock(ctest.fileinput, 'input')
+    m_file = self.mox.CreateMock(file)
+
+    mock_file = _TEST_BOOT_DESC.splitlines(True)
+    ctest.fileinput.input('%s/%s' % (os.path.abspath(self.download_folder),
+                                     'boot.desc'),
+                          inplace=1).AndReturn(mock_file)
+
+    m_file.write('\n')
+    m_file.write('  --arch="x86"\n')
+    m_file.write('  --output_dir="%s"\n' % in_chroot_path)
+    m_file.write('  --espfs_mountpoint="%s/%s"\n' % (in_chroot_path, 'esp'))
+    m_file.write('  --enable_rootfs_verification\n')
+
+    self.mox.ReplayAll()
+    ctest.ModifyBootDesc(os.path.abspath(self.download_folder), m_file)
+    self.mox.VerifyAll()
+
+
+  def testGetLatestZipUrl(self):
+    """Test case that tests GetLatestZipUrl with test urls."""
+    self.mox.StubOutWithMock(urllib, 'urlopen')
+    m_file = self.mox.CreateMock(file)
+
+    urllib.urlopen('%s/%s/LATEST-%s' % (self.latestbase, self.channel,
+                   self.board)).AndReturn(m_file)
+    m_file.read().AndReturn('%s.bin.gz' % self.image_name)
+    m_file.close()
+
+    self.mox.ReplayAll()
+    self.assertEquals(ctest.GetLatestZipUrl(self.board, self.channel,
+                                            self.latestbase, self.zipbase),
+                      self.image_url)
+    self.mox.VerifyAll()
+
+  def testGrabZipAndExtractImageUseCached(self):
+    """Test case where cache holds our image."""
+    self.mox.StubOutWithMock(os.path, 'exists')
+    self.mox.StubOutWithMock(__builtins__, 'open')
+    m_file = self.mox.CreateMock(file)
+
+    os.path.exists('%s/%s' % (
+        self.download_folder, 'download_url')).AndReturn(True)
+
+    open('%s/%s' % (self.download_folder, 'download_url')).AndReturn(m_file)
+    m_file.read().AndReturn(self.image_url)
+    m_file.close()
+
+    os.path.exists('%s/%s' % (
+        self.download_folder, ctest._IMAGE_TO_EXTRACT)).AndReturn(True)
+
+    self.mox.ReplayAll()
+    ctest.GrabZipAndExtractImage(self.image_url, self.download_folder,
+                                 ctest._IMAGE_TO_EXTRACT)
+    self.mox.VerifyAll()
+
+  def CommonDownloadAndExtractImage(self):
+    """Common code to mock downloading image, unzipping it and setting url."""
+    zip_path = os.path.join(self.download_folder, 'image.zip')
+    m_file = self.mox.CreateMock(file)
+
+    ctest.RunCommand(['rm', '-rf', self.download_folder], print_cmd=False)
+    os.mkdir(self.download_folder)
+    urllib.urlretrieve(self.image_url, zip_path)
+    ctest.RunCommand(['unzip', '-d', self.download_folder, zip_path],
+                     print_cmd=False, error_message=mox.IgnoreArg())
+
+    ctest.ModifyBootDesc(self.download_folder)
+
+    open('%s/%s' % (self.download_folder, 'download_url'),
+         'w+').AndReturn(m_file)
+    m_file.write(self.image_url)
+    m_file.close()
+
+    self.mox.ReplayAll()
+    ctest.GrabZipAndExtractImage(self.image_url, self.download_folder,
+                                 ctest._IMAGE_TO_EXTRACT)
+    self.mox.VerifyAll()
+
+  def testGrabZipAndExtractImageNoCache(self):
+    """Test case where download_url doesn't exist."""
+    self.mox.StubOutWithMock(os.path, 'exists')
+    self.mox.StubOutWithMock(os, 'mkdir')
+    self.mox.StubOutWithMock(__builtins__, 'open')
+    self.mox.StubOutWithMock(ctest, 'RunCommand')
+    self.mox.StubOutWithMock(urllib, 'urlretrieve')
+    self.mox.StubOutWithMock(ctest, 'ModifyBootDesc')
+
+    m_file = self.mox.CreateMock(file)
+
+    os.path.exists('%s/%s' % (
+        self.download_folder, 'download_url')).AndReturn(False)
+
+    self.CommonDownloadAndExtractImage()
+
+
+  def testGrabZipAndExtractImageWrongCache(self):
+    """Test case where download_url exists but doesn't match our url."""
+    self.mox.StubOutWithMock(os.path, 'exists')
+    self.mox.StubOutWithMock(os, 'mkdir')
+    self.mox.StubOutWithMock(__builtins__, 'open')
+    self.mox.StubOutWithMock(ctest, 'RunCommand')
+    self.mox.StubOutWithMock(urllib, 'urlretrieve')
+    self.mox.StubOutWithMock(ctest, 'ModifyBootDesc')
+
+    m_file = self.mox.CreateMock(file)
+
+    os.path.exists('%s/%s' % (
+        self.download_folder, 'download_url')).AndReturn(True)
+
+    open('%s/%s' % (self.download_folder, 'download_url')).AndReturn(m_file)
+    m_file.read().AndReturn(self.image_url)
+    m_file.close()
+
+    os.path.exists('%s/%s' % (
+        self.download_folder, ctest._IMAGE_TO_EXTRACT)).AndReturn(False)
+
+    self.CommonDownloadAndExtractImage()
+
+
+if __name__ == '__main__':
+  unittest.main()