1# Copyright (c) 2012 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 re
6import logging
7
8from autotest_lib.client.common_lib import error
9from autotest_lib.server.cros import vboot_constants as vboot
10
11
12class FAFTCheckers(object):
13    """Class that contains FAFT checkers."""
14    version = 1
15
16    def __init__(self, faft_framework):
17        self.faft_framework = faft_framework
18        self.faft_client = faft_framework.faft_client
19        self.faft_config = faft_framework.faft_config
20        self.fw_vboot2 = self.faft_client.system.get_fw_vboot2()
21
22    def _parse_crossystem_output(self, lines):
23        """Parse the crossystem output into a dict.
24
25        @param lines: The list of crossystem output strings.
26        @return: A dict which contains the crossystem keys/values.
27        @raise TestError: If wrong format in crossystem output.
28
29        >>> seq = FAFTSequence()
30        >>> seq._parse_crossystem_output([ \
31                "arch          = x86    # Platform architecture", \
32                "cros_debug    = 1      # OS should allow debug", \
33            ])
34        {'cros_debug': '1', 'arch': 'x86'}
35        >>> seq._parse_crossystem_output([ \
36                "arch=x86", \
37            ])
38        Traceback (most recent call last):
39            ...
40        TestError: Failed to parse crossystem output: arch=x86
41        >>> seq._parse_crossystem_output([ \
42                "arch          = x86    # Platform architecture", \
43                "arch          = arm    # Platform architecture", \
44            ])
45        Traceback (most recent call last):
46            ...
47        TestError: Duplicated crossystem key: arch
48        """
49        pattern = "^([^ =]*) *= *(.*[^ ]) *# [^#]*$"
50        parsed_list = {}
51        for line in lines:
52            matched = re.match(pattern, line.strip())
53            if not matched:
54                raise error.TestError("Failed to parse crossystem output: %s"
55                                      % line)
56            (name, value) = (matched.group(1), matched.group(2))
57            if name in parsed_list:
58                raise error.TestError("Duplicated crossystem key: %s" % name)
59            parsed_list[name] = value
60        return parsed_list
61
62    def crossystem_checker(self, expected_dict, suppress_logging=False):
63        """Check the crossystem values matched.
64
65        Given an expect_dict which describes the expected crossystem values,
66        this function check the current crossystem values are matched or not.
67
68        @param expected_dict: A dict which contains the expected values.
69        @param suppress_logging: True to suppress any logging messages.
70        @return: True if the crossystem value matched; otherwise, False.
71        """
72        succeed = True
73        lines = self.faft_client.system.run_shell_command_get_output(
74                'crossystem')
75        got_dict = self._parse_crossystem_output(lines)
76        for key in expected_dict:
77            if key not in got_dict:
78                logging.warn('Expected key %r not in crossystem result', key)
79                succeed = False
80                continue
81            if isinstance(expected_dict[key], str):
82                if got_dict[key] != expected_dict[key]:
83                    message = ('Expected %r value %r but got %r' % (
84                               key, expected_dict[key], got_dict[key]))
85                    succeed = False
86                else:
87                    message = ('Expected %r value %r == real value %r' % (
88                               key, expected_dict[key], got_dict[key]))
89
90            elif isinstance(expected_dict[key], tuple):
91                # Expected value is a tuple of possible actual values.
92                if got_dict[key] not in expected_dict[key]:
93                    message = ('Expected %r values %r but got %r' % (
94                               key, expected_dict[key], got_dict[key]))
95                    succeed = False
96                else:
97                    message = ('Expected %r values %r == real value %r' % (
98                               key, expected_dict[key], got_dict[key]))
99            else:
100                logging.warn('The expected value of %r is neither a str nor a '
101                             'dict: %r', key, expected_dict[key])
102                succeed = False
103                continue
104            if not suppress_logging:
105                logging.info(message)
106        return succeed
107
108    def mode_checker(self, mode):
109        """Check the current system in the given mode.
110
111        @param mode: A string of mode, one of 'normal', 'dev', or 'rec'.
112        @return: True if the system in the given mode; otherwise, False.
113        """
114        is_devsw = (self.faft_config.mode_switcher_type ==
115                    'physical_button_switcher')
116        if mode == 'normal':
117            if is_devsw:
118                return self.crossystem_checker(
119                        {'devsw_cur': '0'},
120                        suppress_logging=True)
121            else:
122                return self.crossystem_checker(
123                        {'devsw_boot': '0',
124                         'mainfw_type': 'normal'},
125                        suppress_logging=True)
126        elif mode == 'dev':
127            if is_devsw:
128                return self.crossystem_checker(
129                        {'devsw_cur': '1'},
130                        suppress_logging=True)
131            else:
132                return self.crossystem_checker(
133                        {'devsw_boot': '1',
134                         'mainfw_type': 'developer'},
135                        suppress_logging=True)
136        elif mode == 'rec':
137            return self.crossystem_checker(
138                    {'mainfw_type': 'recovery'},
139                    suppress_logging=True)
140        else:
141            raise NotImplementedError('The given mode %s not supported' % mode)
142
143    def fw_tries_checker(self,
144                         expected_mainfw_act,
145                         expected_fw_tried=True,
146                         expected_try_count=0):
147        """Check the current FW booted and try_count
148
149        Mainly for dealing with the vboot1-specific flags fwb_tries and
150        tried_fwb fields in crossystem.  In vboot2, fwb_tries is meaningless and
151        is ignored while tried_fwb is translated into fw_try_count.
152
153        @param expected_mainfw_act: A string of expected firmware, 'A', 'B', or
154                       None if don't care.
155        @param expected_fw_tried: True if tried expected FW at last boot.
156                       This means that mainfw_act=A,tried_fwb=0 or
157                       mainfw_act=B,tried_fwb=1. Set to False if want to
158                       check the opposite case for the mainfw_act.  This
159                       check is only performed in vboot1 as tried_fwb is
160                       never set in vboot2.
161        @param expected_try_count: Number of times to try a FW slot.
162
163        @return: True if the correct boot firmware fields matched.  Otherwise,
164                       False.
165        """
166        crossystem_dict = {'mainfw_act': expected_mainfw_act.upper()}
167
168        if not self.fw_vboot2:
169            if expected_mainfw_act == 'B':
170                tried_fwb_val = True
171            else:
172                tried_fwb_val = False
173            if not expected_fw_tried:
174                tried_fwb_val = not tried_fwb_val
175            crossystem_dict['tried_fwb'] = '1' if tried_fwb_val else '0'
176
177            crossystem_dict['fwb_tries'] = str(expected_try_count)
178        else:
179            crossystem_dict['fw_try_count'] = str(expected_try_count)
180        return self.crossystem_checker(crossystem_dict)
181
182    def vdat_flags_checker(self, mask, value):
183        """Check the flags from VbSharedData matched.
184
185        This function checks the masked flags from VbSharedData using crossystem
186        are matched the given value.
187
188        @param mask: A bitmask of flags to be matched.
189        @param value: An expected value.
190        @return: True if the flags matched; otherwise, False.
191        """
192        lines = self.faft_client.system.run_shell_command_get_output(
193                    'crossystem vdat_flags')
194        vdat_flags = int(lines[0], 16)
195        if vdat_flags & mask != value:
196            logging.info("Expected vdat_flags 0x%x mask 0x%x but got 0x%x",
197                         value, mask, vdat_flags)
198            return False
199        return True
200
201    def ro_normal_checker(self, expected_fw=None, twostop=False):
202        """Check the current boot uses RO boot.
203
204        @param expected_fw: A string of expected firmware, 'A', 'B', or
205                            None if don't care.
206        @param twostop: True to expect a TwoStop boot; False to expect a RO
207                        boot.
208        @return: True if the currect boot firmware matched and used RO boot;
209                 otherwise, False.
210        """
211        crossystem_dict = {'tried_fwb': '0'}
212        if expected_fw:
213            crossystem_dict['mainfw_act'] = expected_fw.upper()
214        succeed = True
215        if not self.vdat_flags_checker(vboot.VDAT_FLAG_LF_USE_RO_NORMAL,
216                0 if twostop else vboot.VDAT_FLAG_LF_USE_RO_NORMAL):
217            succeed = False
218        if not self.crossystem_checker(crossystem_dict):
219            succeed = False
220        if self.faft_framework.check_ec_capability(suppress_warning=True):
221            expected_ec = ('RW' if twostop else 'RO')
222            if not self.ec_act_copy_checker(expected_ec):
223                succeed = False
224        return succeed
225
226    def dev_boot_usb_checker(self, dev_boot_usb=True):
227        """Check the current boot is from a developer USB (Ctrl-U trigger).
228
229        @param dev_boot_usb: True to expect an USB boot;
230                             False to expect an internal device boot.
231        @return: True if the currect boot device matched; otherwise, False.
232        """
233        return (self.crossystem_checker({'mainfw_type': 'developer'}) and
234            self.faft_client.system.is_removable_device_boot() == dev_boot_usb)
235
236    def root_part_checker(self, expected_part):
237        """Check the partition number of the root device matched.
238
239        @param expected_part: A string containing the number of the expected
240                              root partition.
241        @return: True if the currect root  partition number matched;
242                 otherwise, False.
243        """
244        part = self.faft_client.system.get_root_part()[-1]
245        if self.faft_framework.ROOTFS_MAP[expected_part] != part:
246            logging.info("Expected root part %s but got %s",
247                         self.faft_framework.ROOTFS_MAP[expected_part], part)
248            return False
249        return True
250
251    def ec_act_copy_checker(self, expected_copy):
252        """Check the EC running firmware copy matches.
253
254        @param expected_copy: A string containing 'RO', 'A', or 'B' indicating
255                              the expected copy of EC running firmware.
256        @return: True if the current EC running copy matches; otherwise, False.
257        """
258        if self.faft_client.system.has_host():
259            cmd = 'fwtool ec version'
260        else:
261            cmd = 'ectool version'
262        lines = self.faft_client.system.run_shell_command_get_output(cmd)
263        pattern = re.compile("Firmware copy: (.*)")
264        for line in lines:
265            matched = pattern.match(line)
266            if matched:
267                if matched.group(1) == expected_copy:
268                    return True
269                else:
270                    logging.info("Expected EC in %s but now in %s",
271                                 expected_copy, matched.group(1))
272                    return False
273        logging.info("Wrong output format of '%s':\n%s", cmd, '\n'.join(lines))
274        return False
275