• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2018 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
5import logging
6import re
7
8from autotest_lib.client.common_lib import autotemp
9from autotest_lib.client.common_lib import error
10from autotest_lib.client.common_lib import utils
11from autotest_lib.client.cros.enterprise import enterprise_policy_base
12from autotest_lib.client.cros.update_engine import nebraska_wrapper
13from autotest_lib.client.cros.update_engine import update_engine_test
14
15
16class policy_DeviceTargetVersionPrefix(
17        enterprise_policy_base.EnterprisePolicyTest,
18        update_engine_test.UpdateEngineTest):
19    """
20    Test for the DeviceTargetVersionPrefix policy.
21
22    Cases:
23    short_prefix - prefix value is a shortened string of the serving build.
24    full_prefix - prefix value is a full string of the serving build.
25    not_set - policy not set.
26
27    This test only checks whether the update request has the correct values,
28    since the client does not check if the version listed in the update
29    returned by the AU devserver matches or not.
30
31    """
32    version = 1
33    _POLICY_NAME = 'DeviceTargetVersionPrefix'
34    _TEST_CASES = {'short_prefix': '4444.',
35                   'full_prefix': '4444.4.4',
36                   'notset': None}
37
38    # The policy value -> what the device will send in the update request.
39    _POLICY_TO_REQUEST = {'4444.': '4444', '4444.4.4': '4444.4.'}
40
41
42    def _test_version_prefix(self, prefix_value, port):
43        """
44        Actual test.  Fail if update request doesn't match expected.
45
46        @param prefix_value: the value of this policy.
47        @param port: The port we should connect to Nebraska server.
48
49        @raises error.TestFail if test does not pass.
50
51        """
52        # E.g. <updatecheck targetversionprefix="10718.25.0.">
53        MATCH_STR = r'targetversionprefix="(.*?).?"'
54
55        self._check_for_update(port=port)
56
57        utils.poll_for_condition(
58                self._is_update_started,
59                timeout=60,
60                exception=error.TestFail('Update did not start!'))
61
62        latest_request = self._get_latest_initial_request()
63        if not latest_request:
64            raise error.TestFail('Could not find most recent update request!')
65
66        search = re.search(MATCH_STR, latest_request)
67        # If policy is not set, there should be no matches.
68        if not prefix_value:
69            if search:
70                raise error.TestFail('Version prefix was set when the policy '
71                                     'was not set!')
72        # If policy is set, the value should be in the update request.
73        else:
74            if not search:
75                raise error.TestFail('Update request did not contain target '
76                                     'version prefix!')
77            logging.info('Match value: %s', search.group(1))
78            if search.group(1) != self._POLICY_TO_REQUEST[prefix_value]:
79                raise error.TestFail('Version prefix in this update did not '
80                                     'match the value expected from policy!')
81
82
83    def run_once(self, case, image_url):
84        """
85        Entry point of this test.
86
87        @param case: Name of the testcase to run.
88        @param image_url: Url of update image (this build).
89
90        """
91        # Because we are doing polimorphism and the EnterprisePolicyTest is
92        # earlier in the python MRO, this class's initialize() will get called,
93        # but not the UpdateEngineTest's initialize(). So we need to call it
94        # manually.
95        update_engine_test.UpdateEngineTest.initialize(self)
96
97        case_value = self._TEST_CASES[case]
98        self.setup_case(device_policies={self._POLICY_NAME: case_value},
99                        enroll=True)
100
101        metadata_dir = autotemp.tempdir()
102        self._get_payload_properties_file(image_url, metadata_dir.name,
103                                          target_version='999999.9.9')
104        base_url = ''.join(image_url.rpartition('/')[0:2])
105        with nebraska_wrapper.NebraskaWrapper(
106                log_dir=self.resultsdir,
107                update_metadata_dir=metadata_dir.name,
108                update_payloads_address=base_url) as nebraska:
109
110            update_url = nebraska.get_update_url()
111            self._create_custom_lsb_release(update_url, build='1.1.1')
112
113            self._test_version_prefix(case_value, nebraska.get_port())
114