Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(211)

Unified Diff: bin/au_test_harness/au_test.py

Issue 6717011: Remove au_test_harness code and change symlinks to point to new location (Closed) Base URL: http://git.chromium.org/git/crosutils.git@master
Patch Set: Created 9 years, 9 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « bin/au_test_harness/__init__.py ('k') | bin/au_test_harness/au_worker.py » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: bin/au_test_harness/au_test.py
diff --git a/bin/au_test_harness/au_test.py b/bin/au_test_harness/au_test.py
deleted file mode 100644
index 47aad12500a257163be27d3456288262ac36f4eb..0000000000000000000000000000000000000000
--- a/bin/au_test_harness/au_test.py
+++ /dev/null
@@ -1,288 +0,0 @@
-# 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
- public_key_managers = []
-
- @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
- cls.target_image_path = options.target_image
- cls.clean = options.clean
-
- assert options.type in ['real', 'vm'], 'Failed to specify either real|vm.'
- if use_dummy_worker:
- cls.worker_class = dummy_au_worker.DummyAUWorker
- elif options.type == 'vm':
- cls.worker_class = vm_au_worker.VMAUWorker
- else:
- cls.worker_class = real_au_worker.RealAUWorker
-
- # Sanity checks.
- 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. Test root path must be in the chroot.
- if not cls.test_results_root:
- if options.test_results_root:
- assert 'chroot/tmp' in options.test_results_root, \
- 'Must specify a test results root inside tmp in a chroot.'
- cls.test_results_root = options.test_results_root
- else:
- cls.test_results_root = tempfile.mkdtemp(
- prefix='au_test_harness',
- dir=cros_lib.PrependChrootPath('/tmp'))
-
- 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):
- return
- else:
- cros_lib.Warning("Didn't find '%s' in:" % expected_msg)
- cros_lib.Warning(err.stdout)
-
- self.fail('We managed to update when failure was expected')
-
- def AttemptUpdateWithFilter(self, filter, proxy_port=8081):
- """Update through a proxy, with a specified filter, and expect success."""
- self.worker.PrepareBase(self.target_image_path)
-
- # The devserver runs at port 8080 by default. We assume that here, and
- # start our proxy at a different one. We then tell our update tools to
- # have the client connect to our proxy_port instead of 8080.
- proxy = cros_test_proxy.CrosTestProxy(port_in=proxy_port,
- address_out='127.0.0.1',
- port_out=8080,
- filter=filter)
- proxy.serve_forever_in_thread()
- try:
- self.worker.PerformUpdate(self.target_image_path, self.target_image_path,
- proxy_port=proxy_port)
- finally:
- 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, 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)
-
- # Update from - same percentage should pass that originally passed.
- self.worker.PerformUpdate(self.base_image_path, self.target_image_path,
- 'clean')
- self.worker.VerifyImage(self, percent_passed)
-
- def testInterruptedUpdate(self):
- """Tests what happens if we interrupt payload delivery 3 times."""
-
- class InterruptionFilter(cros_test_proxy.Filter):
- """This filter causes the proxy to interrupt the download 3 times
-
- It does this by closing the first three connections to transfer
- 2M total in the outbound connection after they transfer the
- 2M.
- """
- def __init__(self):
- """Defines variable shared across all connections"""
- self.close_count = 0
-
- def setup(self):
- """Called once at the start of each connection."""
- self.data_size = 0
-
- def OutBound(self, data):
- """Called once per packet for outgoing data.
-
- The first three connections transferring more than 2M
- 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.
- """
- def setup(self):
- """Called once at the start of each connection."""
- self.data_size = 0
- self.delay_count = 0
-
- def OutBound(self, data):
- """Called once per packet for outgoing data.
-
- The first three packets after we reach 2M transferred
- 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.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)
« no previous file with comments | « bin/au_test_harness/__init__.py ('k') | bin/au_test_harness/au_worker.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698