Add testPartialUpdate, testCorruptedUpdate.

bin/cros_au_test_harness.py

 #!/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.
 
 import optparse
 import os
+import shutil
 import sys
 import unittest
+import urllib
 
 sys.path.append(os.path.join(os.path.dirname(__file__), '../lib'))
 from cros_build_lib import Die
 from cros_build_lib import Info
 from cros_build_lib import ReinterpretPathForChroot
 from cros_build_lib import RunCommand
+from cros_build_lib import RunCommandCaptureOutput
 from cros_build_lib import Warning
 
 # VM Constants.
 _FULL_VDISK_SIZE = 6072
 _FULL_STATEFULFS_SIZE = 3074
 _KVM_PID_FILE = '/tmp/harness_pid'
 _VERIFY_SUITE = 'suite_Smoke'
 
 # Globals to communicate options to unit tests.
 global base_image_path
 global board
 global remote
 global target_image_path
 global vm_graphics_flag
 
+class UpdateException(Exception):
+  """Exception thrown when UpdateImage or UpdatePayload fail"""
+  def __init__(self, code, stdout):
+    self.code = code
+    self.stdout = stdout
 
 class AUTest(object):
   """Abstract interface that defines an Auto Update test."""
   source_image = ''
   use_delta_updates = False
 
   def setUp(self):
     unittest.TestCase.setUp(self)
     # Set these up as they are used often.
     self.crosutils = os.path.join(os.path.dirname(__file__), '..')
     self.crosutilsbin = os.path.join(os.path.dirname(__file__))
+    self.download = os.path.join(self.crosutilsbin, 'latest_download')

[sosa, 2010/12/01 18:37:39]

self.download_folder?

[dgarrett, 2010/12/02 22:20:20]

Done.

 
   def GetStatefulChangeFlag(self, stateful_change):
     """Returns the flag to pass to image_to_vm for the stateful change."""
     stateful_change_flag = ''
     if stateful_change:
       stateful_change_flag = '--stateful_update_flag=%s' % stateful_change
 
     return stateful_change_flag
 
   def ParseGenerateTestReportOutput(self, output):
@@ skipping 14 matching lines @@
   def TryDeltaAndFallbackToFull(self, src_image, image, stateful_change='old'):
     """Tries the delta update first if set and falls back to full update."""
     if self.use_delta_updates:
       try:
         self.source_image = src_image
         self.UpdateImage(image)
       except:
         Warning('Delta update failed, disabling delta updates and retrying.')
         self.use_delta_updates = False
         self.source_image = ''
-        self.UpdateImage(image)
+        self.UpdateImageReportError(image)
     else:
-      self.UpdateImage(image)
+      self.UpdateImageReportError(image)
+
+  def UpdateImageReportError(self, image_path, stateful_change='old'):

[sosa, 2010/12/01 18:37:39]

needs docstring .. may also want to make private

[dgarrett, 2010/12/02 22:20:20]

Done.

+    try:
+      self.UpdateImage(image_path, stateful_change)
+    except UpdateException as err:
+      # If the update fails, print it out
+      print err.stdout

[sosa, 2010/12/01 18:37:39]

Use warning?

[dgarrett, 2010/12/02 22:20:20]

Done.

+      raise
 
   def PrepareBase(self):
     """Prepares target with base_image_path."""
     pass
 
   def UpdateImage(self, image_path, stateful_change='old'):
     """Updates target with the image given by the image_path.
 
     Args:
       image_path:  Path to the image to update with.  This image must be a test
         image.
       stateful_change: How to modify the stateful partition.  Values are:
           'old':  Don't modify stateful partition.  Just update normally.
           'clean':  Uses clobber-state to wipe the stateful partition with the
             exception of code needed for ssh.
     """
     pass
 
+  def UpdatePayload(self, update_path, stateful_change='old'):

[sosa, 2010/12/01 18:37:39]

Maybe UpdateUsingPayload

[dgarrett, 2010/12/02 22:20:20]

Done.

+    """Updates target with the pre-generated update stored in update_path
+
+    Args:
+      update_path:  Path to the image to update with. This directory should
+      contain both update.gz, and stateful.image.gz
+    """
+    pass
+
   def VerifyImage(self, percent_required_to_pass):
     """Verifies the image with tests.
 
     Verifies that the test images passes the percent required.
 
     Args:
       percent_required_to_pass:  percentage required to pass.  This should be
         fall between 0-100.
 
     Returns:
@@ skipping 63 matching lines @@
     # Update to - all tests should pass on new image.
     Info('Updating from base image on vm to target image and wiping stateful.')
     self.TryDeltaAndFallbackToFull(base_image_path, target_image_path, 'clean')
     self.VerifyImage(100)
 
     # Update from - same percentage should pass that originally passed.
     Info('Updating from updated image back to base image and wiping stateful.')
     self.TryDeltaAndFallbackToFull(target_image_path, base_image_path, 'clean')
     self.VerifyImage(percent_passed)
 
+  def testPartialUpdate(self):
+    """Tests what happens if we attempt to update with a truncated payload.

[sosa, 2010/12/01 18:37:39]

Only needs to be a one liner (i.e end triple quotes on same line)

[dgarrett, 2010/12/02 22:20:20]

Done.

+    """
+    # Preload with the version we are trying to test.
+    self.UpdateImageReportError(target_image_path)

[sosa, 2010/12/01 18:37:39]

Move into PrepareBase and use PrepareBase (e.g. we don't actually have to do the
update for a VM image)

+    percent_passed = self.VerifyImage(100)
+
+    # Image can be updated at:
+    # ~chrome-eng/chromeos/localmirror/autest-images
+    url = 'http://gsdview.appspot.com/chromeos-localmirror/autest-images/truncated_image.gz'

[sosa, 2010/12/01 18:37:39]

80 chars

[dgarrett, 2010/12/02 22:20:20]

Done.

+    payload = os.path.join(self.download, 'truncated_image.gz')
+
+    # Read from the URL and write to the local file
+    shutil.copyfileobj(urllib.urlopen(url), open(payload, 'w'))

[sosa, 2010/12/01 18:37:39]

Maybe use urllib.urlretrieve?

[dgarrett, 2010/12/02 22:20:20]

Nice, I looked for something like that, but missed it.

+
+    # This update is expected to fail...
+    try:
+      self.UpdatePayload(payload)
+    except UpdateException as err:
+      # Will raise ValueError if expected is not found.
+      try:
+        expectedMsg='download_hash_data == update_check_response_hash failed'

[sosa, 2010/12/01 18:37:39]

Might be easier to read with an expected regex rather than a nested try/except

+        err.stdout.index(expectedMsg)
+      except ValueError:
+        print "Didn't find '%s' in:" % expectedMsg

[sosa, 2010/12/01 18:37:39]

Use Info or Warning

[dgarrett, 2010/12/02 22:20:20]

Done.

+        print err.stdout
+        raise
+      return
+
+    unittest.fail('We managed to update with a partial payload')
+
+  def testCorruptedUpdate(self):
+    """Tests what happens if we attempt to update with a corrupted payload.

[sosa, 2010/12/01 18:37:39]

Don't need a second line for docstring

[dgarrett, 2010/12/02 22:20:20]

Done.

+    """
+    # Preload with the version we are trying to test.
+    self.UpdateImageReportError(target_image_path)

[sosa, 2010/12/01 18:37:39]

PrepareBase

[dgarrett, 2010/12/02 22:20:20]

Done.

+    percent_passed = self.VerifyImage(100)
+
+    # Image can be updated at:
+    # ~chrome-eng/chromeos/localmirror/autest-images
+    url = 'http://gsdview.appspot.com/chromeos-localmirror/autest-images/corrupted_image.gz'

[sosa, 2010/12/01 18:37:39]

80 chars

[dgarrett, 2010/12/02 22:20:20]

Done.

+    payload = os.path.join(self.download, 'corrupted.gz')
+
+    # Read from the URL and write to the local file
+    shutil.copyfileobj(urllib.urlopen(url), open(payload, 'w'))
+
+    # This update is expected to fail...
+    try:
+      self.UpdatePayload(payload)
+    except UpdateException as err:
+      # Will raise ValueError if expected is not found.
+      try:
+        # There are a number of different errors that can occur here, but
+        # we're just trying to test the whole thing.
+        expectedMsg='zlib inflate() error:-3'

[sosa, 2010/12/01 18:37:39]

expected_msg

[dgarrett, 2010/12/02 22:20:20]

Done.

+        err.stdout.index(expectedMsg)
+      except ValueError:
+        print "Didn't find '%s' in:" % expectedMsg
+        print err.stdoutKeepStateful
+        raise
+      return
+
+    unittest.fail('We managed to update with a corrupted payload')

[sosa, 2010/12/01 18:37:39]

This code seems almost the same as the test before it.  May want to try to write
a common function that takes args i.e.

(payload, expected_msg, test_fail_message)

[dgarrett, 2010/12/02 22:20:20]

Done.

 
 class RealAUTest(unittest.TestCase, AUTest):
   """Test harness for updating real images."""
 
   def setUp(self):
     AUTest.setUp(self)
 
   def PrepareBase(self):
     """Auto-update to base image to prepare for test."""
-    self.UpdateImage(base_image_path)
+    self.UpdateImageReportError(base_image_path)
 
   def UpdateImage(self, image_path, stateful_change='old'):
     """Updates a remote image using image_to_live.sh."""
     stateful_change_flag = self.GetStatefulChangeFlag(stateful_change)
 
-    RunCommand([
+    (code, stdout, stderr) = RunCommandCaptureOutput([
         '%s/image_to_live.sh' % self.crosutils,
         '--image=%s' % image_path,
         '--remote=%s' % remote,
         stateful_change_flag,
         '--verify',
-        '--src_image=%s' % self.source_image,
-        ], enter_chroot=False)
+        '--src_image=%s' % self.source_image
+        ])
 
+    if code != 0:
+      raise UpdateException(code, stdout)
+
+  def UpdatePayload(self, update_path, stateful_change='old'):
+    """Updates a remote image using image_to_live.sh."""
+    stateful_change_flag = self.GetStatefulChangeFlag(stateful_change)
+
+    (code, stdout, stderr) = RunCommandCaptureOutput([
+        '%s/image_to_live.sh' % self.crosutils,
+        '--payload=%s' % update_path,
+        '--remote=%s' % remote,
+        stateful_change_flag,
+        '--verify',
+        ])
+
+    if code != 0:
+      raise UpdateException(code, stdout)
 
   def VerifyImage(self, percent_required_to_pass):
     """Verifies an image using run_remote_tests.sh with verification suite."""
     output = RunCommand([
         '%s/run_remote_tests.sh' % self.crosutils,
         '--remote=%s' % remote,
         _VERIFY_SUITE,
        ], error_ok=True, enter_chroot=False, redirect_stdout=True)
     return self.CommonVerifyImage(self, output, percent_required_to_pass)
 
@@ skipping 37 matching lines @@
       Info('Using existing VM image %s' % self.vm_image_path)
 
     self.assertTrue(os.path.exists(self.vm_image_path))
 
   def UpdateImage(self, image_path, stateful_change='old'):
     """Updates VM image with image_path."""
     stateful_change_flag = self.GetStatefulChangeFlag(stateful_change)
     if self.source_image == base_image_path:
       self.source_image = self.vm_image_path
 
-    RunCommand(['%s/cros_run_vm_update' % self.crosutilsbin,
-                '--update_image_path=%s' % image_path,
-                '--vm_image_path=%s' % self.vm_image_path,
-                '--snapshot',
-                vm_graphics_flag,
-                '--persist',
-                '--kvm_pid=%s' % _KVM_PID_FILE,
-                stateful_change_flag,
-                '--src_image=%s' % self.source_image,
-               ], enter_chroot=False)
+    (code, stdout, stderr) = RunCommandCaptureOutput([
+        '%s/cros_run_vm_update' % self.crosutilsbin,
+        '--update_image_path=%s' % image_path,
+        '--vm_image_path=%s' % self.vm_image_path,
+        '--snapshot',
+        vm_graphics_flag,
+        '--persist',
+        '--kvm_pid=%s' % _KVM_PID_FILE,
+        stateful_change_flag,
+        '--src_image=%s' % self.source_image,
+        ])
+
+    if code != 0:
+      raise UpdateException(code, stdout)
+
+  def UpdatePayload(self, update_path, stateful_change='old'):
+    """Updates a remote image using image_to_live.sh."""
+    stateful_change_flag = self.GetStatefulChangeFlag(stateful_change)
+
+    (code, stdout, stderr) = RunCommandCaptureOutput([

[sosa, 2010/12/01 18:37:39]

Does this work for VM's?  The port isn't correct ... you'd need to use
cros_run_vm_update and pass payload through that

[dgarrett, 2010/12/02 22:20:20]

It did not, and the fix was larger than I was expecting.

On 2010/12/01 18:37:39, sosa wrote:

+        '%s/image_to_live.sh' % self.crosutils,
+        '--payload=%s' % update_path,
+        stateful_change_flag,
+        '--for_vm',
+        '--verify',
+        ])
+
+    if code != 0:
+      raise UpdateException(code, stdout)
 
   def VerifyImage(self, percent_required_to_pass):
     """Runs vm smoke suite to verify image."""
     # image_to_live already verifies lsb-release matching.  This is just
     # for additional steps.
 
     commandWithArgs = ['%s/cros_run_vm_test' % self.crosutilsbin,
                        '--image_path=%s' % self.vm_image_path,
                        '--snapshot',
                        '--persist',
@@ skipping 66 matching lines @@
     else:
       remote = options.remote
 
     suite = unittest.TestLoader().loadTestsFromTestCase(RealAUTest)
     test_result = unittest.TextTestRunner(verbosity=2).run(suite)
   else:
     parser.error('Could not parse harness type %s.' % options.type)
 
   if not test_result.wasSuccessful():
     Die('Test harness was not successful')