1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# 4# Copyright (c) 2014 The Chromium OS Authors. All rights reserved. 5# Use of this source code is governed by a BSD-style license that can be 6# found in the LICENSE file. 7 8"""Unittest for suite_runner.""" 9 10from __future__ import print_function 11 12import json 13 14import unittest 15import unittest.mock as mock 16 17import suite_runner 18import label 19 20from benchmark import Benchmark 21 22from cros_utils import command_executer 23from cros_utils import logger 24from machine_manager import MockCrosMachine 25 26 27class SuiteRunnerTest(unittest.TestCase): 28 """Class of SuiteRunner test.""" 29 mock_json = mock.Mock(spec=json) 30 mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter) 31 mock_cmd_term = mock.Mock(spec=command_executer.CommandTerminator) 32 mock_logger = mock.Mock(spec=logger.Logger) 33 mock_label = label.MockLabel('lumpy', 'build', 'lumpy_chromeos_image', '', '', 34 '/tmp/chromeos', 'lumpy', 35 ['lumpy1.cros', 'lumpy.cros2'], '', '', False, 36 'average', 'gcc', False, '') 37 telemetry_crosperf_bench = Benchmark( 38 'b1_test', # name 39 'octane', # test_name 40 '', # test_args 41 3, # iterations 42 False, # rm_chroot_tmp 43 'record -e cycles', # perf_args 44 'telemetry_Crosperf', # suite 45 True) # show_all_results 46 47 crosperf_wrapper_bench = Benchmark( 48 'b2_test', # name 49 'webgl', # test_name 50 '', # test_args 51 3, # iterations 52 False, # rm_chroot_tmp 53 '', # perf_args 54 'crosperf_Wrapper') # suite 55 56 tast_bench = Benchmark( 57 'b3_test', # name 58 'platform.ReportDiskUsage', # test_name 59 '', # test_args 60 1, # iterations 61 False, # rm_chroot_tmp 62 '', # perf_args 63 'tast') # suite 64 65 def __init__(self, *args, **kwargs): 66 super(SuiteRunnerTest, self).__init__(*args, **kwargs) 67 self.crosfleet_run_args = [] 68 self.test_that_args = [] 69 self.tast_args = [] 70 self.call_crosfleet_run = False 71 self.call_test_that_run = False 72 self.call_tast_run = False 73 74 def setUp(self): 75 self.runner = suite_runner.SuiteRunner({}, self.mock_logger, 'verbose', 76 self.mock_cmd_exec, 77 self.mock_cmd_term) 78 79 def test_get_profiler_args(self): 80 input_str = ("--profiler=custom_perf --profiler_args='perf_options" 81 '="record -a -e cycles,instructions"\'') 82 output_str = ("profiler=custom_perf profiler_args='record -a -e " 83 "cycles,instructions'") 84 res = suite_runner.GetProfilerArgs(input_str) 85 self.assertEqual(res, output_str) 86 87 def test_get_dut_config_args(self): 88 dut_config = {'enable_aslr': False, 'top_interval': 1.0} 89 output_str = ('dut_config=' 90 "'" 91 '{"enable_aslr": ' 92 'false, "top_interval": 1.0}' 93 "'" 94 '') 95 res = suite_runner.GetDutConfigArgs(dut_config) 96 self.assertEqual(res, output_str) 97 98 def test_run(self): 99 100 def reset(): 101 self.test_that_args = [] 102 self.crosfleet_run_args = [] 103 self.tast_args = [] 104 self.call_test_that_run = False 105 self.call_crosfleet_run = False 106 self.call_tast_run = False 107 108 def FakeCrosfleetRun(test_label, benchmark, test_args, profiler_args): 109 self.crosfleet_run_args = [ 110 test_label, benchmark, test_args, profiler_args 111 ] 112 self.call_crosfleet_run = True 113 return 'Ran FakeCrosfleetRun' 114 115 def FakeTestThatRun(machine, test_label, benchmark, test_args, 116 profiler_args): 117 self.test_that_args = [ 118 machine, test_label, benchmark, test_args, profiler_args 119 ] 120 self.call_test_that_run = True 121 return 'Ran FakeTestThatRun' 122 123 def FakeTastRun(machine, test_label, benchmark): 124 self.tast_args = [machine, test_label, benchmark] 125 self.call_tast_run = True 126 return 'Ran FakeTastRun' 127 128 self.runner.Crosfleet_Run = FakeCrosfleetRun 129 self.runner.Test_That_Run = FakeTestThatRun 130 self.runner.Tast_Run = FakeTastRun 131 132 self.runner.dut_config['enable_aslr'] = False 133 self.runner.dut_config['cooldown_time'] = 0 134 self.runner.dut_config['governor'] = 'fake_governor' 135 self.runner.dut_config['cpu_freq_pct'] = 65 136 self.runner.dut_config['intel_pstate'] = 'no_hwp' 137 machine = 'fake_machine' 138 cros_machine = MockCrosMachine(machine, self.mock_label.chromeos_root, 139 self.mock_logger) 140 test_args = '' 141 profiler_args = '' 142 143 # Test crosfleet run for telemetry_Crosperf and crosperf_Wrapper benchmarks. 144 self.mock_label.crosfleet = True 145 reset() 146 self.runner.Run(cros_machine, self.mock_label, self.crosperf_wrapper_bench, 147 test_args, profiler_args) 148 self.assertTrue(self.call_crosfleet_run) 149 self.assertFalse(self.call_test_that_run) 150 self.assertEqual(self.crosfleet_run_args, 151 [self.mock_label, self.crosperf_wrapper_bench, '', '']) 152 153 reset() 154 self.runner.Run(cros_machine, self.mock_label, 155 self.telemetry_crosperf_bench, test_args, profiler_args) 156 self.assertTrue(self.call_crosfleet_run) 157 self.assertFalse(self.call_test_that_run) 158 self.assertEqual(self.crosfleet_run_args, 159 [self.mock_label, self.telemetry_crosperf_bench, '', '']) 160 161 # Test test_that run for telemetry_Crosperf and crosperf_Wrapper benchmarks. 162 self.mock_label.crosfleet = False 163 reset() 164 self.runner.Run(cros_machine, self.mock_label, self.crosperf_wrapper_bench, 165 test_args, profiler_args) 166 self.assertTrue(self.call_test_that_run) 167 self.assertFalse(self.call_crosfleet_run) 168 self.assertEqual( 169 self.test_that_args, 170 ['fake_machine', self.mock_label, self.crosperf_wrapper_bench, '', '']) 171 172 reset() 173 self.runner.Run(cros_machine, self.mock_label, 174 self.telemetry_crosperf_bench, test_args, profiler_args) 175 self.assertTrue(self.call_test_that_run) 176 self.assertFalse(self.call_crosfleet_run) 177 self.assertEqual(self.test_that_args, [ 178 'fake_machine', self.mock_label, self.telemetry_crosperf_bench, '', '' 179 ]) 180 181 # Test tast run for tast benchmarks. 182 reset() 183 self.runner.Run(cros_machine, self.mock_label, self.tast_bench, '', '') 184 self.assertTrue(self.call_tast_run) 185 self.assertFalse(self.call_test_that_run) 186 self.assertFalse(self.call_crosfleet_run) 187 self.assertEqual(self.tast_args, 188 ['fake_machine', self.mock_label, self.tast_bench]) 189 190 def test_gen_test_args(self): 191 test_args = '--iterations=2' 192 perf_args = 'record -a -e cycles' 193 194 # Test crosperf_Wrapper benchmarks arg list generation 195 args_list = ["test_args='--iterations=2'", "dut_config='{}'", 'test=webgl'] 196 res = self.runner.GenTestArgs(self.crosperf_wrapper_bench, test_args, '') 197 self.assertCountEqual(res, args_list) 198 199 # Test telemetry_Crosperf benchmarks arg list generation 200 args_list = [ 201 "test_args='--iterations=2'", "dut_config='{}'", 'test=octane', 202 'run_local=False' 203 ] 204 args_list.append(suite_runner.GetProfilerArgs(perf_args)) 205 res = self.runner.GenTestArgs(self.telemetry_crosperf_bench, test_args, 206 perf_args) 207 self.assertCountEqual(res, args_list) 208 209 @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') 210 @mock.patch.object(command_executer.CommandExecuter, 211 'ChrootRunCommandWOutput') 212 def test_tast_run(self, mock_chroot_runcmd, mock_cros_runcmd): 213 mock_chroot_runcmd.return_value = 0 214 self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd 215 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 216 res = self.runner.Tast_Run('lumpy1.cros', self.mock_label, self.tast_bench) 217 self.assertEqual(mock_cros_runcmd.call_count, 1) 218 self.assertEqual(mock_chroot_runcmd.call_count, 1) 219 self.assertEqual(res, 0) 220 self.assertEqual(mock_cros_runcmd.call_args_list[0][0], 221 ('rm -rf /usr/local/autotest/results/*',)) 222 args_list = mock_chroot_runcmd.call_args_list[0][0] 223 args_dict = mock_chroot_runcmd.call_args_list[0][1] 224 self.assertEqual(len(args_list), 2) 225 self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term) 226 227 @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') 228 @mock.patch.object(command_executer.CommandExecuter, 229 'ChrootRunCommandWOutput') 230 @mock.patch.object(logger.Logger, 'LogFatal') 231 def test_test_that_run(self, mock_log_fatal, mock_chroot_runcmd, 232 mock_cros_runcmd): 233 mock_log_fatal.side_effect = SystemExit() 234 self.runner.logger.LogFatal = mock_log_fatal 235 # Test crosperf_Wrapper benchmarks cannot take perf_args 236 raised_exception = False 237 try: 238 self.runner.Test_That_Run('lumpy1.cros', self.mock_label, 239 self.crosperf_wrapper_bench, '', 240 'record -a -e cycles') 241 except SystemExit: 242 raised_exception = True 243 self.assertTrue(raised_exception) 244 245 mock_chroot_runcmd.return_value = 0 246 self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd 247 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 248 res = self.runner.Test_That_Run('lumpy1.cros', self.mock_label, 249 self.crosperf_wrapper_bench, 250 '--iterations=2', '') 251 self.assertEqual(mock_cros_runcmd.call_count, 1) 252 self.assertEqual(mock_chroot_runcmd.call_count, 1) 253 self.assertEqual(res, 0) 254 self.assertEqual(mock_cros_runcmd.call_args_list[0][0], 255 ('rm -rf /usr/local/autotest/results/*',)) 256 args_list = mock_chroot_runcmd.call_args_list[0][0] 257 args_dict = mock_chroot_runcmd.call_args_list[0][1] 258 self.assertEqual(len(args_list), 2) 259 self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term) 260 261 @mock.patch.object(command_executer.CommandExecuter, 'RunCommandWOutput') 262 @mock.patch.object(json, 'loads') 263 def test_crosfleet_run_client(self, mock_json_loads, mock_runcmd): 264 265 def FakeDownloadResult(l, task_id): 266 if l and task_id: 267 self.assertEqual(task_id, '12345') 268 return 0 269 270 mock_runcmd.return_value = ( 271 0, 272 'Created Swarming task https://swarming/task/b12345', 273 '', 274 ) 275 self.mock_cmd_exec.RunCommandWOutput = mock_runcmd 276 277 mock_json_loads.return_value = { 278 'child-results': [{ 279 'success': True, 280 'task-run-url': 'https://swarming/task?id=12345' 281 }] 282 } 283 self.mock_json.loads = mock_json_loads 284 285 self.mock_label.crosfleet = True 286 self.runner.DownloadResult = FakeDownloadResult 287 res = self.runner.Crosfleet_Run(self.mock_label, 288 self.crosperf_wrapper_bench, '', '') 289 ret_tup = (0, '\nResults placed in tmp/swarming-12345\n', '') 290 self.assertEqual(res, ret_tup) 291 self.assertEqual(mock_runcmd.call_count, 2) 292 293 args_list = mock_runcmd.call_args_list[0][0] 294 args_dict = mock_runcmd.call_args_list[0][1] 295 self.assertEqual(len(args_list), 1) 296 self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term) 297 298 args_list = mock_runcmd.call_args_list[1][0] 299 self.assertEqual(args_list[0], ('crosfleet wait-task 12345')) 300 self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term) 301 302 303if __name__ == '__main__': 304 unittest.main() 305