Add ability to pass a base test root and create results dirs relative.

bin/au_test_harness/au_test.py

 # Copyright (c) 2011 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.
 
 """Module containing a test suite that is run to test auto updates."""
 
 import os
+import tempfile
 import time
 import unittest
 
 import cros_build_lib as cros_lib
 
 import cros_test_proxy
 import dummy_au_worker
 import real_au_worker
 import vm_au_worker
 
 
 class AUTest(unittest.TestCase):
   """Test harness that uses an au_worker to perform and validate updates.
 
   Defines a test suite that is run using an au_worker.  An au_worker can
   be created to perform and validates updates on both virtual and real devices.
   See documentation for au_worker for more information.
   """
+  test_results_root = None
+
   @classmethod
   def ProcessOptions(cls, options, use_dummy_worker):
     """Processes options for the test suite and sets up the worker class.
 
     Args:
       options: options class to be parsed from main class.
       use_dummy_worker: If True, use a dummy_worker_class rather than deriving
         one from options.type.
     """
     cls.base_image_path = options.base_image
@@ skipping 12 matching lines @@
     if not cls.base_image_path:
       cros_lib.Die('Need path to base image for vm.')
     elif not os.path.exists(cls.base_image_path):
       cros_lib.Die('%s does not exist' % cls.base_image_path)
 
     if not cls.target_image_path:
       cros_lib.Die('Need path to target image to update with.')
     elif not os.path.exists(cls.target_image_path):
       cros_lib.Die('%s does not exist' % cls.target_image_path)
 
+    # Initialize test root.
+    if not cls.test_results_root:
+      if options.test_results_root:
+        cls.test_results_root = options.test_results_root
+      else:
+        cls.test_results_root = tempfile.mkdtemp(prefix='au_test_harness')
+
+      cros_lib.Info('Using %s as the test results root' % cls.test_results_root)
+
     # Cache away options to instantiate workers later.
     cls.options = options
 
   def AttemptUpdateWithPayloadExpectedFailure(self, payload, expected_msg):
     """Attempt a payload update, expect it to fail with expected log"""
     try:
       self.worker.UpdateUsingPayload(payload)
     except UpdateException as err:
       # Will raise ValueError if expected is not found.
       if re.search(re.escape(expected_msg), err.stdout, re.MULTILINE):
@@ skipping 23 matching lines @@
       proxy.shutdown()
 
   # --- UNITTEST SPECIFIC METHODS ---
 
   def setUp(self):
     """Overrides unittest.TestCase.setUp and called before every test.
 
     Sets instance specific variables and initializes worker.
     """
     unittest.TestCase.setUp(self)
-    self.worker = self.worker_class(self.options)
+    self.worker = self.worker_class(self.options, AUTest.test_results_root)
     self.crosutils = os.path.join(os.path.dirname(__file__), '..', '..')
     self.download_folder = os.path.join(self.crosutils, 'latest_download')
     if not os.path.exists(self.download_folder):
       os.makedirs(self.download_folder)
 
   def tearDown(self):
     """Overrides unittest.TestCase.tearDown and called after every test."""
     self.worker.CleanUp()
 
   def testUpdateKeepStateful(self):
     """Tests if we can update normally.
 
     This test checks that we can update by updating the stateful partition
     rather than wiping it.
     """
+    self.worker.InitializeResultsDirectory()
     # Just make sure some tests pass on original image.  Some old images
     # don't pass many tests.
     self.worker.PrepareBase(self.base_image_path)
     # TODO(sosa): move to 100% once we start testing using the autotest paired
     # with the dev channel.
     percent_passed = self.worker.VerifyImage(self, 10)
 
     # Update to - all tests should pass on new image.
     self.worker.PerformUpdate(self.target_image_path, self.base_image_path)
     percent_passed = self.worker.VerifyImage(self)
 
     # Update from - same percentage should pass that originally passed.
     self.worker.PerformUpdate(self.base_image_path, self.target_image_path)
     self.worker.VerifyImage(self, percent_passed)
 
   def testUpdateWipeStateful(self):
     """Tests if we can update after cleaning the stateful partition.
 
     This test checks that we can update successfully after wiping the
     stateful partition.
     """
+    self.worker.InitializeResultsDirectory()
     # Just make sure some tests pass on original image.  Some old images
     # don't pass many tests.
     self.worker.PrepareBase(self.base_image_path)
     percent_passed = self.worker.VerifyImage(self, 10)
 
     # Update to - all tests should pass on new image.
     self.worker.PerformUpdate(self.target_image_path, self.base_image_path,
                               'clean')
     self.worker.VerifyImage(self)
 
@@ skipping 27 matching lines @@
            outbound will be closed.
         """
         if self.close_count < 3:
           if self.data_size > (2 * 1024 * 1024):
             self.close_count += 1
             return None
 
         self.data_size += len(data)
         return data
 
+    self.worker.InitializeResultsDirectory()
     self.AttemptUpdateWithFilter(InterruptionFilter(), proxy_port=8082)
 
   def testDelayedUpdate(self):
     """Tests what happens if some data is delayed during update delivery"""
 
     class DelayedFilter(cros_test_proxy.Filter):
       """Causes intermittent delays in data transmission.
 
          It does this by inserting 3 20 second delays when transmitting
          data after 2M has been sent.
@@ skipping 10 matching lines @@
            are delayed by 20 seconds.
         """
         if self.delay_count < 3:
           if self.data_size > (2 * 1024 * 1024):
             self.delay_count += 1
             time.sleep(20)
 
         self.data_size += len(data)
         return data
 
+    self.worker.InitializeResultsDirectory()
     self.AttemptUpdateWithFilter(DelayedFilter(), proxy_port=8083)
 
   def SimpleTest(self):
     """A simple update that updates once from a base image to a target.
 
     We explicitly don't use test prefix so that isn't run by default.  Can be
     run using test_prefix option.
     """
+    self.worker.InitializeResultsDirectory()
     self.worker.PrepareBase(self.base_image_path)
-    self.worker.PerformUpdate(self.target_image_path, self.base_image_path)
+    #self.worker.PerformUpdate(self.target_image_path, self.base_image_path)
     self.worker.VerifyImage(self)
 
   # --- DISABLED TESTS ---
 
   # TODO(sosa): Get test to work with verbose.
   def NotestPartialUpdate(self):
     """Tests what happens if we attempt to update with a truncated payload."""
+    self.worker.InitializeResultsDirectory()
     # Preload with the version we are trying to test.
     self.worker.PrepareBase(self.target_image_path)
 
     # Image can be updated at:
     # ~chrome-eng/chromeos/localmirror/autest-images
     url = 'http://gsdview.appspot.com/chromeos-localmirror/' \
           'autest-images/truncated_image.gz'
     payload = os.path.join(self.download_folder, 'truncated_image.gz')
 
     # Read from the URL and write to the local file
     urllib.urlretrieve(url, payload)
 
     expected_msg = 'download_hash_data == update_check_response_hash failed'
     self.AttemptUpdateWithPayloadExpectedFailure(payload, expected_msg)
 
   # TODO(sosa): Get test to work with verbose.
   def NotestCorruptedUpdate(self):
     """Tests what happens if we attempt to update with a corrupted payload."""
+    self.worker.InitializeResultsDirectory()
     # Preload with the version we are trying to test.
     self.worker.PrepareBase(self.target_image_path)
 
     # Image can be updated at:
     # ~chrome-eng/chromeos/localmirror/autest-images
     url = 'http://gsdview.appspot.com/chromeos-localmirror/' \
           'autest-images/corrupted_image.gz'
     payload = os.path.join(self.download_folder, 'corrupted.gz')
 
     # Read from the URL and write to the local file
     urllib.urlretrieve(url, payload)
 
     # This update is expected to fail...
     expected_msg = 'zlib inflate() error:-3'
     self.AttemptUpdateWithPayloadExpectedFailure(payload, expected_msg)