1# Lint as: python2, python3 2# Copyright 2020 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 6"""Wrapper test to run verification on a servo_host/dut pair.""" 7 8import ast 9import logging 10 11from autotest_lib.client.common_lib import error 12from autotest_lib.server import test 13from autotest_lib.server.cros.dynamic_suite import suite 14 15class servo_Verification(test.test): 16 """A wrapper test to run the suite |servo_lab| against a dut/servo pair.""" 17 version = 1 18 19 DEFAULT_SUITE = "servo_lab" 20 21 def get_test_args_from_control(self, control_data): 22 """Helper to extract the control file information. 23 24 We leverage control files and suite matching to not have to duplicate 25 the work of writing the test arguments out. However, we cannot just 26 execute the control-file itself, but rather need to extract the args, 27 and then runsubtest ourselves. 28 29 Please make sure that the control files being run in this suite are 30 compatible with the limitations indicated below, otherwise, modify 31 the test, or add a new control file. 32 33 A few things to note: 34 - tests will always be run with disable_sysinfo 35 - args that are not literals e.g. local=local and local is defined 36 somewhere else in the control file will be set to None 37 - 'args' and 'args_dict' will be passed along as '' and {} and available 38 as such e.g. if an arg says 'cmdline_args=args' 39 40 @param control_data: ControlData of a parsed control file 41 42 @returns: tuple(test, args): where test is the main test name 43 args is a kwargs dict to pass to runsubtest 44 """ 45 # Skipped args that we do not evaluate 46 skipped_args = ['args', 'args_dict', 'disable_sysinfo', 'host'] 47 args = '' 48 args_dict = {} 49 # The result that we will populate. 50 test_args = {'args': args, 51 'args_dict': args_dict, 52 'disable_sysinfo': True} 53 cname = control_data.name 54 control_file = control_data.text 55 anchor = 'job.run_test' 56 if anchor not in control_file: 57 raise error.TestNAError('Control file for test %s does not define ' 58 '%s.' % (cname, anchor)) 59 # Find the substring only 60 run_test_str = control_file[control_file.index(anchor) + len(anchor):] 61 # Find the balanced parentheses 62 paran = 1 63 # This assumes that the string is job.run_test(...) so the first ( is 64 # at index 0. 65 for index in range(1, len(run_test_str)): 66 if run_test_str[index] == '(': paran += 1 67 if run_test_str[index] == ')': paran -= 1 68 if paran == 0: break 69 else: 70 # Failed to find balanced parentheses. 71 raise error.TestNAError('Unable to parse %s for %s.' % 72 (anchor, cname)) 73 # Extract only the args 74 run_test_str = run_test_str[1:index] 75 raw_args = run_test_str.split(',') 76 try: 77 base_test_name = ast.literal_eval(raw_args[0]) 78 except (ValueError, SyntaxError) as e: 79 logging.debug('invalid run_test_str: %s. %s', run_test_str, str(e)) 80 raise error.TestNAError('Unable to parse test name from %s for %s.' 81 % (anchor, cname)) 82 # Parse an evaluate the remaining args 83 for arg in raw_args[1:]: 84 # Issues here are also caught by ValueError below. 85 aname, aval = arg.split('=') 86 aname = aname.strip() 87 aval = aval.strip() 88 if aname not in skipped_args: 89 # eval() is used here as some test might make references 90 # to 'args' and 'args_dict'. Hence the BaseException below 91 # as any error might occur here. 92 try: 93 test_args[aname] = eval(aval) 94 except BaseException as e: 95 logging.debug(str(e)) 96 logging.info('Unable to parse value %r for arg %r. Setting ' 97 'to None.', aval, aname) 98 test_args[aname] = None 99 100 logging.info('Will run the test %s as %s with args: %s', cname, 101 base_test_name, test_args) 102 return base_test_name, test_args 103 104 def initialize(self, host, local=False): 105 """Prepare all test-names and args to be run. 106 107 @param host: cros host to run the test against. Needs to have a servo 108 @param: on False, the latest repair image is downloaded onto the usb 109 stick. Set to true to skip (reuse image on stick) 110 """ 111 fs_getter = suite.create_fs_getter(self.autodir) 112 # Find the test suite in autotest file system. 113 predicate = suite.name_in_tag_predicate(self.DEFAULT_SUITE) 114 tests = suite.find_and_parse_tests(fs_getter, predicate) 115 if not tests: 116 raise error.TestNAError('%r suite has no tests under it.' % 117 self.DEFAULT_SUITE) 118 self._tests = [] 119 for data in tests: 120 try: 121 self._tests.append(self.get_test_args_from_control(data)) 122 except error.TestNAError as e: 123 logging.info('Unable to parse %s. Skipping. %s', data.name, 124 str(e)) 125 if not self._tests: 126 raise error.TestFail('No test parsed successfully.') 127 self._tests.sort(key=lambda t: t[0]) 128 if not local: 129 # Pre-download the usb image onto the stick so that tests that 130 # need it can use it. 131 _, image_url = host.stage_image_for_servo() 132 host.servo.image_to_servo_usb(image_url) 133 # `image_to_servo_usb` turned DUT off while download image to usb 134 # drive, so we need to turn DUT back on as some tests assume DUT 135 # is sshable at begin. 136 host.servo.get_power_state_controller().power_on() 137 if not host.wait_up(timeout=host.BOOT_TIMEOUT): 138 logging.warning( 139 '%s failed to boot in %s seconds, some tests' 140 ' may fail due to not able to ssh to the DUT', 141 host.hostname, host.BOOT_TIMEOUT) 142 143 def run_once(self, host): 144 """Run through the test sequence. 145 146 @param host: cros host to run the test against. Needs to have a servo 147 148 @raises: error.TestFail if any test in the sequence fails 149 """ 150 success = True 151 for idx, test in enumerate(self._tests): 152 tname, targs = test 153 # Some tests might run multiple times e.g. 154 # platform_ServoPowerStateController with usb and without usb. 155 # The subdir task ensures that there won't ever be a naming 156 # collision. 157 subdir_tag = '%02d' % idx 158 success &= self.runsubtest(tname, subdir_tag=subdir_tag, 159 host=host, **targs) 160 if not success: 161 raise error.TestFail('At least one verification test failed. ' 162 'Check the logs.') 163