1 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import logging 6 7 from autotest_lib.client.common_lib import error 8 from autotest_lib.client.common_lib import utils 9 from autotest_lib.client.cros.cellular import test_environment 10 from autotest_lib.client.cros.update_engine import nano_omaha_devserver 11 from autotest_lib.client.cros.update_engine import update_engine_test 12 13 class autoupdate_CannedOmahaUpdate(update_engine_test.UpdateEngineTest): 14 """ 15 Client-side mechanism to update a DUT with a given image. 16 17 Restarts update_engine and attempts an update from the image 18 pointed to by |image_url| of size |image_size| with checksum 19 |image_sha256|. The rest of the parameters are optional. 20 21 If the |allow_failure| parameter is True, then the test will 22 succeed even if the update failed. 23 24 """ 25 version = 1 26 27 28 def run_canned_update(self, allow_failure): 29 """ 30 Performs the update. 31 32 @param allow_failure: True if we dont raise an error on failure. 33 34 """ 35 self._omaha.start() 36 try: 37 utils.run('update_engine_client -update -omaha_url=' + 38 'http://127.0.0.1:%d/update ' % self._omaha.get_port()) 39 except error.CmdError as e: 40 self._omaha.stop() 41 if not allow_failure: 42 raise error.TestFail('Update attempt failed: %s' % 43 self._get_last_error_string()) 44 else: 45 logging.info('Ignoring failed update. Failure reason: %s', e) 46 47 48 def run_once(self, image_url, image_size, image_sha256, 49 allow_failure=False, metadata_size=None, 50 metadata_signature=None, public_key=None, use_cellular=False, 51 is_delta=False): 52 53 self._omaha = nano_omaha_devserver.NanoOmahaDevserver() 54 self._omaha.set_image_params(image_url, image_size, image_sha256, 55 metadata_size, metadata_signature, 56 public_key, is_delta) 57 58 if use_cellular: 59 # Setup DUT so that we have ssh over ethernet but DUT uses 60 # cellular as main connection. 61 try: 62 test_env = test_environment.CellularOTATestEnvironment() 63 CONNECT_TIMEOUT = 120 64 with test_env: 65 service = test_env.shill.wait_for_cellular_service_object() 66 if not service: 67 raise error.TestError('No cellular service found.') 68 test_env.shill.connect_service_synchronous( 69 service, CONNECT_TIMEOUT) 70 self.run_canned_update(allow_failure) 71 except error.TestError as e: 72 # Raise as test failure so it is propagated to server test 73 # failure message. 74 logging.error('Failed setting up cellular connection.') 75 raise error.TestFail(e) 76 else: 77 self.run_canned_update(allow_failure) 78