blob: 231accf5044a562e2c4d5f37e4023881c8368022 [file] [log] [blame]
# 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 re
import time
import unittest
import urllib
import cros_build_lib as cros_lib
from crostestutils.au_test_harness import cros_test_proxy
from crostestutils.au_test_harness import real_au_worker
from crostestutils.au_test_harness import update_exception
from crostestutils.au_test_harness 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.
"""
@classmethod
def ProcessOptions(cls, options):
"""Processes options for the test suite and sets up the worker class.
Args:
options: options class to be parsed from main class.
"""
cls.base_image_path = options.base_image
cls.private_key = options.private_key
cls.target_image_path = options.target_image
cls.test_results_root = options.test_results_root
if options.type == 'vm':
cls.worker_class = vm_au_worker.VMAUWorker
else:
cls.worker_class = real_au_worker.RealAUWorker
# 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 update_exception.UpdateException as err:
# Will raise ValueError if expected is not found.
if re.search(re.escape(expected_msg), err.output, re.MULTILINE):
return
else:
cros_lib.Warning("Didn't find '%s' in:" % expected_msg)
cros_lib.Warning(err.output)
self.fail('We managed to update when failure was expected')
def AttemptUpdateWithFilter(self, update_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=update_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.
"""
super(AUTest, self).setUp()
self.worker = self.worker_class(self.options, AUTest.test_results_root)
self.download_folder = os.path.join(os.path.realpath(os.path.curdir),
'latest_download')
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.Initialize(9222)
# Just make sure some tests pass on original image. Some old images
# don't pass many tests.
self.worker.PrepareBase(self.base_image_path)
# Update to
self.worker.PerformUpdate(self.target_image_path, self.base_image_path)
self.assertTrue(self.worker.VerifyImage())
# Update from
self.worker.PerformUpdate(self.target_image_path, self.target_image_path)
self.assertTrue(self.worker.VerifyImage())
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.Initialize(9223)
# Just make sure some tests pass on original image. Some old images
# don't pass many tests.
self.worker.PrepareBase(self.base_image_path)
# Update to
self.worker.PerformUpdate(self.target_image_path, self.base_image_path,
'clean')
self.assertTrue(self.worker.VerifyImage())
# Update from
self.worker.PerformUpdate(self.target_image_path, self.target_image_path,
'clean')
self.assertTrue(self.worker.VerifyImage())
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 after they transfer
2M total in the outbound direction.
"""
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
# Overriden method. The first three connections transferring more than 2M
# outbound will be closed.
def OutBound(self, data):
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.Initialize(9224)
self.AttemptUpdateWithFilter(InterruptionFilter(), proxy_port=8082)
def testSimpleSignedUpdate(self):
"""Test that updates to itself with a signed payload."""
self.worker.Initialize(9226)
self.worker.PrepareBase(self.target_image_path, signed_base=True)
if self.private_key:
self.worker.PerformUpdate(self.target_image_path,
self.target_image_path + '.signed',
private_key_path=self.private_key)
else:
cros_lib.Info('No key found to use for signed testing.')
def SimpleTestUpdateAndVerify(self):
"""Test that updates to itself.
We explicitly don't use test prefix so that isn't run by default. Can be
run using test_prefix option.
"""
self.worker.Initialize(9227)
self.worker.PrepareBase(self.target_image_path)
self.worker.PerformUpdate(self.target_image_path, self.target_image_path)
self.assertTrue(self.worker.VerifyImage())
def SimpleTestVerify(self):
"""Test that only verifies the target image.
We explicitly don't use test prefix so that isn't run by default. Can be
run using test_prefix option.
"""
self.worker.Initialize(9228)
self.worker.PrepareBase(self.target_image_path)
self.assertTrue(self.worker.VerifyImage())
# --- DISABLED TESTS ---
def NoTestDelayedUpdate(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
# The first three packets after we reach 2M transferred
# are delayed by 20 seconds.
def OutBound(self, data):
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.Initialize(9225)
self.AttemptUpdateWithFilter(DelayedFilter(), proxy_port=8083)
def NotestPlatformToolchainOptions(self):
"""Tests the hardened toolchain options."""
self.worker.Initialize(9229)
self.worker.PrepareBase(self.base_image_path)
self.assertTrue(self.worker.VerifyImage('platform_ToolchainOptions'))
# 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.Initialize(9230)
# 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.Initialize(9231)
# 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)