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