1 # Copyright 2017 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 import os 7 import urlparse 8 9 from autotest_lib.server import autotest 10 11 class ChromiumOSTestPlatform(object): 12 """Represents a CrOS device during autoupdate. 13 14 This class is used with autoupdate_EndToEndTest. It has functions for all 15 the device specific things that we need during an update: reboot, 16 check active slot, login, get logs, start an update etc. 17 """ 18 19 _UPDATE_ENGINE_PERF_PATH = '/mnt/stateful_partition/unencrypted/preserve' 20 _UPDATE_ENGINE_PERF_SCRIPT = 'update_engine_performance_monitor.py' 21 _UPDATE_ENGINE_PERF_RESULTS_FILE = 'perf_data_results.json' 22 23 def __init__(self, host, autotest_devserver, results_dir): 24 """Initialize the class. 25 26 @param: host: The DUT host. 27 @param: autotest_devserver: The devserver to call cros_au on. 28 @param: results_dir: Where to save the autoupdate logs files. 29 """ 30 self._host = host 31 self._autotest_devserver = autotest_devserver 32 self._results_dir = results_dir 33 34 35 def _install_version(self, payload_uri, clobber_stateful=False): 36 """Install the specified payload. 37 38 @param payload_uri: GS URI of the payload to install. 39 @param clobber_stateful: force a reinstall of the stateful image. 40 """ 41 build_name, payload_file = self._get_update_parameters_from_uri( 42 payload_uri) 43 logging.info('Installing %s on the DUT', payload_uri) 44 45 try: 46 ds = self._autotest_devserver 47 _, pid = ds.auto_update(host_name=self._host.hostname, 48 build_name=build_name, 49 force_update=True, 50 full_update=True, 51 log_dir=self._results_dir, 52 payload_filename=payload_file, 53 clobber_stateful=clobber_stateful) 54 except: 55 logging.fatal('ERROR: Failed to install image on the DUT.') 56 raise 57 return pid 58 59 60 def _run_login_test(self, tag): 61 """Runs login_LoginSuccess test on the DUT.""" 62 client_at = autotest.Autotest(self._host) 63 client_at.run_test('login_LoginSuccess', tag=tag) 64 65 66 @staticmethod 67 def _get_update_parameters_from_uri(payload_uri): 68 """Extract the two vars needed for cros_au from the Google Storage URI. 69 70 dev_server.auto_update needs two values from this test: 71 (1) A build_name string e.g samus-release/R60-9583.0.0 72 (2) A filename of the exact payload file to use for the update. This 73 payload needs to have already been staged on the devserver. 74 75 This function extracts those two values from a Google Storage URI. 76 77 @param payload_uri: Google Storage URI to extract values from 78 """ 79 archive_url, _, payload_file = payload_uri.rpartition('/') 80 build_name = urlparse.urlsplit(archive_url).path.strip('/') 81 82 # This test supports payload uris from two Google Storage buckets. 83 # They store their payloads slightly differently. One stores them in 84 # a separate payloads directory. E.g 85 # gs://chromeos-image-archive/samus-release/R60-9583.0.0/blah.bin 86 # gs://chromeos-releases/dev-channel/samus/9334.0.0/payloads/blah.bin 87 if build_name.endswith('payloads'): 88 build_name = build_name.rpartition('/')[0] 89 payload_file = 'payloads/' + payload_file 90 91 logging.debug('Extracted build_name: %s, payload_file: %s from %s.', 92 build_name, payload_file, payload_uri) 93 return build_name, payload_file 94 95 96 def reboot_device(self): 97 """Reboot the device.""" 98 self._host.reboot() 99 100 101 def install_source_image(self, source_payload_uri): 102 """Install source payload on device.""" 103 if source_payload_uri: 104 self._install_version(source_payload_uri, clobber_stateful=True) 105 106 107 def check_login_after_source_update(self): 108 """Make sure we can login before the target update.""" 109 self._run_login_test('source_update') 110 111 112 def get_active_slot(self): 113 """Returns the current active slot.""" 114 return self._host.run('rootdev -s').stdout.strip() 115 116 117 def copy_perf_script_to_device(self, bindir): 118 """Copy performance monitoring script to DUT. 119 120 The updater will kick off the script during the update. 121 """ 122 logging.info('Copying %s to device.', self._UPDATE_ENGINE_PERF_SCRIPT) 123 path = os.path.join(bindir, self._UPDATE_ENGINE_PERF_SCRIPT) 124 self._host.send_file(path, self._UPDATE_ENGINE_PERF_PATH) 125 126 127 def get_perf_stats_for_update(self, resultdir): 128 """ Get the performance metrics created during update.""" 129 try: 130 path = os.path.join('/var/log', 131 self._UPDATE_ENGINE_PERF_RESULTS_FILE) 132 self._host.get_file(path, resultdir) 133 self._host.run('rm %s' % path) 134 script = os.path.join(self._UPDATE_ENGINE_PERF_PATH, 135 self._UPDATE_ENGINE_PERF_SCRIPT) 136 self._host.run('rm %s' % script) 137 return os.path.join(resultdir, 138 self._UPDATE_ENGINE_PERF_RESULTS_FILE) 139 except: 140 logging.warning('Failed to copy performance metrics from DUT.') 141 return None 142 143 144 def install_target_image(self, target_payload_uri): 145 """Install target payload on the device.""" 146 logging.info('Updating device to target image.') 147 return self._install_version(target_payload_uri) 148 149 150 def get_update_log(self, num_lines): 151 """Get the latest lines from the update engine log.""" 152 return self._host.run_output( 153 'tail -n %d /var/log/update_engine.log' % num_lines, 154 stdout_tee=None) 155 156 157 def check_login_after_target_update(self): 158 """Check we can login after updating.""" 159 self._run_login_test('target_update') 160 161 162 def oobe_triggers_update(self): 163 """Check if this device has an OOBE that completes itself.""" 164 return self._host.oobe_triggers_update() 165 166 167 def get_cros_version(self): 168 """Returns the ChromeOS version installed on this device.""" 169 return self._host.get_release_version() 170
