| """Fetch OTA step.""" |
| |
| from __future__ import absolute_import |
| import importlib |
| import logging |
| import os |
| import sys |
| |
| # As per b/128461961 requests is an indirect requirement to import fetcher |
| import requests |
| from slave import base_step |
| |
| |
| class FetchOtaException(Exception): # pylint: disable=g-bad-exception-name |
| """Raised when a fetch_ota_step exception happens.""" |
| |
| |
| class FetchOtaStep(base_step.BaseStep): |
| """Step to fetch the required OTA for a device.""" |
| |
| def __init__( |
| self, branch, product, board, |
| flavor, work_dir, **kwargs): |
| """Creates a FetchOtaStep instance. |
| |
| Args: |
| branch: Which branch should be used. |
| product: Name of DUT. |
| board: Type of board of product. |
| flavor: Type of build. |
| work_dir: Location of files to work on. |
| **kwargs: Any additionar args to pass to BaseStep |
| """ |
| base_step.BaseStep.__init__(self, name='fetch_ota', **kwargs) |
| self._branch = branch |
| self._board_type = board |
| self._flavor = flavor |
| self._product = product |
| self._work_dir = work_dir |
| |
| def _get_fetcher_module(self): |
| """Returns the fetcher class from eurtest. |
| |
| Returns: |
| Fetcher module from eurtest. |
| """ |
| test_path = os.path.abspath(os.path.join( |
| os.pardir, self.get_project_path('test'))) |
| if test_path not in sys.path: |
| sys.path.insert( |
| 0, test_path) |
| return importlib.import_module('eurtest.tools.build.fetcher') |
| |
| def GetFetcher(self): |
| """Configures the fetcher for the OTA required. |
| |
| Returns: |
| A Fetcher instance for the OTA required for this test run. |
| """ |
| fetcher = self._get_fetcher_module() |
| ota_fetcher = fetcher.GetFetcher( |
| branch=self._branch, |
| product=self._product, |
| flavor=self._flavor, |
| board_type=self._board_type |
| ) |
| return ota_fetcher |
| |
| def _fetch(self): |
| """Fetches the OTA using the recipe information passed down. |
| |
| Returns: |
| If successfully fetched returns True, else False. |
| """ |
| success = True |
| fetcher_lib = self._get_fetcher_module() |
| ota_fetcher = self.GetFetcher() |
| try: |
| ota_fetcher.Fetch(self.get_property('build_number'), self._work_dir) |
| except fetcher_lib.FetcherError: |
| success = False |
| logging.error( |
| 'Failed to found ota for build %s for %s', |
| self.get_property('build_number'), |
| self._product) |
| return success |
| |
| def run(self): |
| """Fetches and saves in a temporary directory the OTA. |
| |
| Returns: |
| True if successfully fetched the OTA, False otherwise. |
| """ |
| success = self._fetch() |
| if not success: |
| review_message = 'Failed to fetch OTA for build %s and device %s' % ( |
| self.get_property('build_number'), |
| self._product) |
| self.add_review({'message': review_message}) |
| return success |
