| # Copyright 2018 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. |
| |
| import logging |
| import re |
| |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros.update_engine import nano_omaha_devserver |
| from autotest_lib.client.cros.update_engine import update_engine_util |
| |
| _MIN_BUILD = '1.1.1' |
| _MAX_BUILD = '999999.9.9' |
| |
| # eventtype value sent by client in an AU request. Indicates that |
| # device rebooted after an update since last check. |
| # TODO: remove this, crbug.com/879687 |
| _EVENT_TYPE_REBOOTED_AFTER_UPDATE = '54' |
| |
| class NanoOmahaEnterpriseAUContext(object): |
| """ |
| Contains methods required for Enterprise AU tests using Nano Omaha. |
| |
| """ |
| def __init__(self, image_url, image_size, sha256, to_build=_MAX_BUILD, |
| from_build=_MIN_BUILD, is_rollback=False, is_critical=False): |
| """ |
| Start a Nano Omaha instance and intialize variables. |
| |
| @param image_url: Url of update image. |
| @param image_size: Size of the update. |
| @param sha256: Sha256 hash of the update. |
| @param to_build: String of the build number Nano Omaha should serve. |
| @param from_build: String of the build number this device should say |
| it is on by setting lsb_release. |
| @param is_rollback: whether the build should serve with the rollback |
| flag. |
| @param is_critical: whether the build should serve marked as critical. |
| |
| """ |
| self._omaha = nano_omaha_devserver.NanoOmahaDevserver() |
| self._omaha.set_image_params(image_url, image_size, sha256, |
| build=to_build, is_rollback=is_rollback) |
| self._omaha.start() |
| |
| self._au_util = update_engine_util.UpdateEngineUtil() |
| |
| update_url = self._omaha.get_update_url() |
| self._au_util._create_custom_lsb_release(from_build, update_url) |
| |
| self._is_rollback = is_rollback |
| self._is_critical = is_critical |
| |
| |
| def update_and_poll_for_update_start(self, is_interactive=False): |
| """ |
| Check for an update and wait until it starts. |
| |
| @param is_interactive: whether the request is interactive. |
| |
| @raises: error.TestFail when update does not start after timeout. |
| |
| """ |
| self._au_util._check_for_update(port=self._omaha.get_port(), |
| interactive=is_interactive) |
| |
| def update_started(): |
| """Polling function: True or False if update has started.""" |
| status = self._au_util._get_update_engine_status() |
| logging.info('Status: %s', status) |
| return (status[self._au_util._CURRENT_OP] |
| == self._au_util._UPDATE_ENGINE_DOWNLOADING) |
| |
| utils.poll_for_condition( |
| update_started, |
| timeout=30, |
| exception=error.TestFail('Update did not start!')) |
| |
| |
| def get_update_requests(self): |
| """ |
| Get the contents of all the update requests from the most recent log. |
| |
| @returns: a sequential list of <request> xml blocks or None if none. |
| |
| """ |
| return self._au_util._get_update_requests() |
| |
| |
| def get_time_of_last_update_request(self): |
| """ |
| Get the time of the last update request from most recent logfile. |
| |
| @returns: seconds since epoch of when last update request happened |
| (second accuracy), or None if no such timestamp exists. |
| |
| """ |
| return self._au_util._get_time_of_last_update_request() |
| |
| |
| def get_latest_initial_request(self): |
| """ |
| Return the most recent initial update request. |
| |
| AU requests occur in a chain of messages back and forth, e.g. the |
| initial request for an update -> the reply with the update -> the |
| report that install has started -> report that install has finished, |
| etc. This function finds the first request in the latest such chain. |
| |
| This message has no eventtype listed, or is rebooted_after_update |
| type (as an artifact from a previous update since this one). |
| Subsequent messages in the chain have different eventtype values. |
| |
| @returns: string of the entire update request or None. |
| |
| """ |
| requests = self.get_update_requests() |
| if not requests: |
| return None |
| |
| MATCH_STR = r'eventtype="(.*?)"' |
| for i in xrange(len(requests) - 1, -1, -1): |
| search = re.search(MATCH_STR, requests[i]) |
| if (not search or |
| search.group(1) == _EVENT_TYPE_REBOOTED_AFTER_UPDATE): |
| return requests[i] |
| |
| return None |