1# Copyright 2022 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 os 6import shutil 7 8from autotest_lib.server import autotest, test 9from autotest_lib.client.common_lib import error 10 11 12class test_sequence(test.test): 13 """ 14 test_sequence extends the base test implementation to allow for 15 encapsulating a series of (client or server) tests which must 16 be run in a given sequence. 17 """ 18 19 def initialize(self, sequence): 20 """ 21 initialize implements the initialize call in test.test, is called before 22 execution of the test 23 24 @param sequence: the sequence of tests constructed in the wrapper 25 @param sequence_verdicts: verdicts from each executed test in the 26 sequence. Passed by reference and used by the caller to 27 annotate results. 28 """ 29 self._sequenced_tests = sequence 30 self._sequence_verdicts = {} 31 self._results_path = self.job._server_offload_dir_path() 32 self._wrapper_results_dir = os.path.join(self._results_path, 33 self.tagged_testname) 34 35 def process_test_results_post_hook(self): 36 """ 37 process_test_results is used as a post_run_hook to record results to the 38 status.log file following the execution of the run. For tests that were 39 completed (i.e. no exceptions occurred to end the sequence), results are 40 moved to the top level from the child results directory 41 """ 42 for test, args, server_test in self._sequenced_tests: 43 if test not in self._sequence_verdicts: 44 continue 45 46 if server_test: 47 self.surface_server_test_resultsdir(test) 48 else: 49 self.surface_client_test_resultsdir(test) 50 annotated_testname = self.tagged_testname + "." + test 51 self.job.record('START', None, annotated_testname) 52 self.job.record('INFO', None, annotated_testname) 53 if self._sequence_verdicts[test]: 54 self.job.record('END GOOD', None, annotated_testname, "") 55 else: 56 self.job.record('END FAIL', None, annotated_testname, "") 57 58 def execute_sequenced_test(self, client, test, argv, server_test): 59 """ 60 execute_sequenced_test runs a single test from the sequence with the 61 given argument vector 62 63 @param test: test name (url) to run 64 @param argv: argument dictionary to run the test with 65 66 @raises error.TestFail: on failure of the wrapped tests 67 """ 68 try: 69 self._sequence_verdicts[test] = True 70 if server_test: 71 err = self.job.run_test(test, **argv) 72 if err == False: 73 raise error.TestFail() 74 else: 75 client.run_test(test, check_client_result=True, **argv) 76 except: 77 self._sequence_verdicts[test] = False 78 self.postprocess() 79 raise error.TestFail('Sequenced test %s failed, reason: ' % test) 80 81 def surface_client_test_resultsdir(self, test): 82 """ 83 surface_client_test_resultsdir retrieves the child test results from a 84 sequenced client job 85 86 @param test: the child test name to grab results from 87 """ 88 wrapped_test_results_path = os.path.join(self._wrapper_results_dir, 89 test) 90 tagged_destination = os.path.join(self._results_path, 91 self.tagged_testname + "." + test) 92 shutil.move(wrapped_test_results_path, tagged_destination) 93 94 def surface_server_test_resultsdir(self, test): 95 """ 96 surface_server_test_resultsdir renames the server test results from a sequenced child 97 98 @param test: the child test name to grab results from 99 """ 100 wrapped_test_results_path = os.path.join(self._results_path, test) 101 tagged_destination = os.path.join(self._results_path, 102 self.tagged_testname + "." + test) 103 shutil.move(wrapped_test_results_path, tagged_destination) 104 105 def run_once(self, host=None): 106 """ 107 run_once implements the run_once call in test.test, is called to begin 108 execution of the test 109 110 @param host: host from control file with which to run the test 111 """ 112 client_at = autotest.Autotest(host) 113 for test, argv, server_test in self._sequenced_tests: 114 self.execute_sequenced_test(client_at, test, argv, server_test) 115 116 def postprocess(self): 117 """ 118 postprocess is the post routine for test.test. We must add our post_hook 119 in this function (as opposed to the initialize call) because if added in 120 initialize, this will be called after each child server test as well as 121 at the end of the function 122 """ 123 self.job.add_post_run_hook(self.process_test_results_post_hook) 124