1# Lint as: python2, python3
2# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import logging
7
8from autotest_lib.client.common_lib import error
9from autotest_lib.client.cros.cellular import test_environment
10from autotest_lib.client.cros.update_engine import nebraska_wrapper
11from autotest_lib.client.cros.update_engine import update_engine_test
12
13class autoupdate_CannedOmahaUpdate(update_engine_test.UpdateEngineTest):
14    """ Updates a DUT with a given image using a Nebraska instance."""
15
16    version = 1
17
18
19    def run_canned_update(self, allow_failure, update_url, interactive):
20        """
21        Performs the update.
22
23        @param allow_failure: True if we dont raise an error on failure.
24        @param update_url: The URL to get an update.
25        @param interactive: Whether the update is interactive or not.
26
27        """
28
29        try:
30            self._check_for_update(update_url,
31                                   critical_update=True,
32                                   wait_for_completion=True,
33                                   interactive=interactive)
34        except error.CmdError as e:
35            if not allow_failure:
36                raise error.TestFail('Update attempt failed: %s' %
37                                     self._get_last_error_string())
38            else:
39                logging.info('Ignoring failed update. Failure reason: %s', e)
40
41
42    def run_once(self,
43                 payload_url,
44                 allow_failure=False,
45                 public_key=None,
46                 use_cellular=False,
47                 interactive=True):
48        """
49        Runs an update with a canned response using Nebraska.
50
51        @param payload_url: Path to a payload on Google storage.
52        @param allow_failure: If true, failing the update is expected.
53        @param public_key: The public key to serve to the update client.
54        @param use_cellular: True if this test uses cellular.
55        @param interactive: Whether the update is interactive or not.
56
57        """
58
59        with nebraska_wrapper.NebraskaWrapper(
60            log_dir=self.resultsdir, payload_url=payload_url,
61            public_key=public_key) as nebraska:
62
63            if not use_cellular:
64                self.run_canned_update(allow_failure,
65                                       nebraska.get_update_url(), interactive)
66                return
67
68            # Setup DUT so that we have ssh over ethernet but DUT uses
69            # cellular as main connection.
70            try:
71                with test_environment.CellularOTATestEnvironment() as test_env:
72                    service = test_env.shill.wait_for_cellular_service_object()
73                    if not service:
74                        raise error.TestError('No cellular service found.')
75
76                    CONNECT_TIMEOUT = 120
77                    test_env.shill.connect_service_synchronous(
78                            service, CONNECT_TIMEOUT)
79                    self.run_canned_update(allow_failure,
80                                           nebraska.get_update_url(),
81                                           interactive)
82            except error.TestError as e:
83                # Raise as test failure instead of test error so it is
84                # propagated to the server test's failure message.
85                logging.error('Failed setting up cellular connection.')
86                raise error.TestFail(e)
87