1# Lint as: python2, python3
2# Copyright (c) 2012 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
7import os
8
9from autotest_lib.client.common_lib import error
10from autotest_lib.client.common_lib.cros import kernel_utils
11from autotest_lib.client.cros import constants
12from autotest_lib.server.cros import provisioner
13from autotest_lib.server.cros.update_engine import update_engine_test
14
15
16class autoupdate_EndToEndTest(update_engine_test.UpdateEngineTest):
17    """Complete update test between two ChromeOS releases.
18
19    Performs an end-to-end test of updating a ChromeOS device from one version
20    to another. The test performs the following steps:
21
22      - Stages the source (full) and target update payloads on a devserver.
23      - Installs source image on the DUT (if provided) and reboots to it.
24      - Verifies that sign in works correctly on the source image.
25      - Installs target image on the DUT and reboots.
26      - Does a final update check.
27      - Verifies that sign in works correctly on the target image.
28      - Returns the hostlogs collected during each update check for
29        verification against expected update events.
30
31    This class interacts with several others:
32    UpdateEngineTest: base class for comparing expected update events against
33                      the events listed in the hostlog.
34    UpdateEngineEvent: class representing a single expected update engine event.
35
36    """
37    version = 1
38
39
40    def cleanup(self):
41        """Save the logs from stateful_partition's preserved/log dir."""
42        stateful_preserved_logs = os.path.join(self.resultsdir,
43                                               '~stateful_preserved_logs')
44        os.makedirs(stateful_preserved_logs)
45        self._host.get_file(constants.AUTOUPDATE_PRESERVE_LOG,
46                            stateful_preserved_logs, safe_symlinks=True,
47                            preserve_perm=False)
48        super(autoupdate_EndToEndTest, self).cleanup()
49
50
51    def _print_rerun_command(self, test_conf):
52        """Prints the command to rerun a test run from the lab at your desk."""
53        logging.debug('Rerun this test run at your desk using this command:')
54        rerun_cmd = ('test_that <DUT NAME>.cros autoupdate_EndToEndTest '
55                     '--args="update_type=%s source_release=%s '
56                     'source_payload_uri=%s target_release=%s '
57                     'target_payload_uri=%s"')
58        rerun_cmd = rerun_cmd % (
59                test_conf['update_type'], test_conf['source_release'],
60                test_conf['source_payload_uri'], test_conf['target_release'],
61                test_conf['target_payload_uri'])
62        logging.debug(rerun_cmd)
63
64    def run_update_test(self, test_conf, m2n):
65        """Runs the update test and checks it succeeded.
66
67        @param test_conf: A dictionary containing test configuration values.
68        @param m2n: True for an m2n test run.
69
70        """
71        # Record the active root partition.
72        active, inactive = kernel_utils.get_kernel_state(self._host)
73        logging.info('Source active slot: %s', active)
74
75        source_release = test_conf['source_release']
76        target_release = test_conf['target_release']
77
78        self.update_device(test_conf['target_payload_uri'],
79                           tag='target',
80                           m2n=m2n)
81
82        # Compare hostlog events from the update to the expected ones.
83        rootfs, reboot = self._create_hostlog_files()
84        self.verify_update_events(source_release, rootfs)
85        self.verify_update_events(source_release, reboot, target_release)
86        kernel_utils.verify_boot_expectations(inactive, host=self._host)
87        logging.info('Update successful, test completed')
88
89
90    def run_once(self, test_conf, m2n=False, build=None):
91        """Performs a complete auto update test.
92
93        @param test_conf: a dictionary containing test configuration values.
94        @param m2n: M -> N update. This means we install the current stable
95                    version of this board before updating to ToT.
96        @param build: target build for the update, i.e. R102-14650.0.0. Optional
97                      argument for running locally.
98
99        """
100        if m2n:
101            if self._host.get_board().endswith("-kernelnext"):
102                raise error.TestNAError("Skipping test on kernelnext board")
103            # No test_conf is provided, we need to assemble it ourselves for
104            # the target update information.
105            source_release = self._get_latest_serving_stable_build().rsplit(
106                    '/')[-1]
107            target_release = build.split(
108                    '-')[1] if build else self._host.get_release_version()
109            target_uri = self.get_payload_for_nebraska(build=build)
110            test_conf = {
111                    'target_release': target_release,
112                    'target_payload_uri': target_uri,
113                    'source_release': source_release,
114                    'source_payload_uri': None
115            }
116
117        logging.debug('The test configuration supplied: %s', test_conf)
118        if not m2n:
119            self._print_rerun_command(test_conf)
120        self._autotest_devserver = self._get_devserver_for_test(test_conf)
121
122        # Install source image with quick-provision.
123        build_name = None
124        source_payload_uri = test_conf['source_payload_uri']
125        if m2n:
126            build_name = self._get_latest_serving_stable_build()
127        elif source_payload_uri:
128            build_name, _ = self._get_update_parameters_from_uri(
129                source_payload_uri)
130
131        if build_name is not None:
132            update_url = self._autotest_devserver.get_update_url(
133                build_name)
134            logging.info('Installing source image with update url: %s',
135                         update_url)
136            provisioner.ChromiumOSProvisioner(
137                    update_url, host=self._host,
138                    is_release_bucket=True).run_provision()
139
140            self._run_client_test_and_check_result(
141                    self._LOGIN_TEST,
142                    tag='source',
143                    username=self._LOGIN_TEST_USERNAME,
144                    password=self._LOGIN_TEST_PASSWORD)
145        # Start the update to the target image.
146        self.run_update_test(test_conf, m2n)
147
148        # Check we can login after the update.
149        self._run_client_test_and_check_result(
150                self._LOGIN_TEST,
151                tag='target',
152                username=self._LOGIN_TEST_USERNAME,
153                password=self._LOGIN_TEST_PASSWORD)
154