1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# 4# Copyright 2014 The ChromiumOS Authors 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 10 11import contextlib 12import json 13import unittest 14import unittest.mock as mock 15 16from benchmark import Benchmark 17from cros_utils import command_executer 18from cros_utils import logger 19import label 20from machine_manager import MockCrosMachine 21import suite_runner 22 23 24class SuiteRunnerTest(unittest.TestCase): 25 """Class of SuiteRunner test.""" 26 27 mock_json = mock.Mock(spec=json) 28 mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter) 29 mock_cmd_term = mock.Mock(spec=command_executer.CommandTerminator) 30 mock_logger = mock.Mock(spec=logger.Logger) 31 mock_label = label.MockLabel( 32 "lumpy", 33 "build", 34 "lumpy_chromeos_image", 35 "", 36 "", 37 "/tmp/chromeos", 38 "lumpy", 39 ["lumpy1.cros", "lumpy.cros2"], 40 "", 41 "", 42 False, 43 "average", 44 "gcc", 45 False, 46 "", 47 ) 48 telemetry_crosperf_bench = Benchmark( 49 "b1_test", # name 50 "octane", # test_name 51 "", # test_args 52 3, # iterations 53 False, # rm_chroot_tmp 54 "record -e cycles", # perf_args 55 "telemetry_Crosperf", # suite 56 True, 57 ) # show_all_results 58 59 crosperf_wrapper_bench = Benchmark( 60 "b2_test", # name 61 "webgl", # test_name 62 "", # test_args 63 3, # iterations 64 False, # rm_chroot_tmp 65 "", # perf_args 66 "crosperf_Wrapper", 67 ) # suite 68 69 tast_bench = Benchmark( 70 "b3_test", # name 71 "platform.ReportDiskUsage", # test_name 72 "", # test_args 73 1, # iterations 74 False, # rm_chroot_tmp 75 "", # perf_args 76 "tast", 77 ) # suite 78 79 def __init__(self, *args, **kwargs): 80 super(SuiteRunnerTest, self).__init__(*args, **kwargs) 81 self.crosfleet_run_args = [] 82 self.test_that_args = [] 83 self.tast_args = [] 84 self.call_crosfleet_run = False 85 self.call_test_that_run = False 86 self.call_tast_run = False 87 88 def setUp(self): 89 self.runner = suite_runner.SuiteRunner( 90 {}, 91 self.mock_logger, 92 "verbose", 93 self.mock_cmd_exec, 94 self.mock_cmd_term, 95 ) 96 97 def test_get_profiler_args(self): 98 input_str = ( 99 "--profiler=custom_perf --profiler_args='perf_options" 100 '="record -a -e cycles,instructions"\'' 101 ) 102 output_str = ( 103 "profiler=custom_perf profiler_args='record -a -e " 104 "cycles,instructions'" 105 ) 106 res = suite_runner.GetProfilerArgs(input_str) 107 self.assertEqual(res, output_str) 108 109 def test_get_dut_config_args(self): 110 dut_config = {"enable_aslr": False, "top_interval": 1.0} 111 output_str = ( 112 "dut_config=" 113 "'" 114 '{"enable_aslr": ' 115 'false, "top_interval": 1.0}' 116 "'" 117 "" 118 ) 119 res = suite_runner.GetDutConfigArgs(dut_config) 120 self.assertEqual(res, output_str) 121 122 @mock.patch("suite_runner.ssh_tunnel") 123 def test_run(self, ssh_tunnel): 124 @contextlib.contextmanager 125 def mock_ssh_tunnel(_watcher, _host): 126 yield "fakelocalhost:1234" 127 128 ssh_tunnel.side_effect = mock_ssh_tunnel 129 130 def reset(): 131 self.test_that_args = [] 132 self.crosfleet_run_args = [] 133 self.tast_args = [] 134 self.call_test_that_run = False 135 self.call_crosfleet_run = False 136 self.call_tast_run = False 137 138 def FakeCrosfleetRun(test_label, benchmark, test_args, profiler_args): 139 self.crosfleet_run_args = [ 140 test_label, 141 benchmark, 142 test_args, 143 profiler_args, 144 ] 145 self.call_crosfleet_run = True 146 return "Ran FakeCrosfleetRun" 147 148 def FakeTestThatRun( 149 machine, test_label, benchmark, test_args, profiler_args 150 ): 151 self.test_that_args = [ 152 machine, 153 test_label, 154 benchmark, 155 test_args, 156 profiler_args, 157 ] 158 self.call_test_that_run = True 159 return "Ran FakeTestThatRun" 160 161 def FakeTastRun(machine, test_label, benchmark): 162 self.tast_args = [machine, test_label, benchmark] 163 self.call_tast_run = True 164 return "Ran FakeTastRun" 165 166 self.runner.Crosfleet_Run = FakeCrosfleetRun 167 self.runner.Test_That_Run = FakeTestThatRun 168 self.runner.Tast_Run = FakeTastRun 169 170 self.runner.dut_config["enable_aslr"] = False 171 self.runner.dut_config["cooldown_time"] = 0 172 self.runner.dut_config["governor"] = "fake_governor" 173 self.runner.dut_config["cpu_freq_pct"] = 65 174 self.runner.dut_config["intel_pstate"] = "no_hwp" 175 machine = "fake_machine" 176 cros_machine = MockCrosMachine( 177 machine, self.mock_label.chromeos_root, self.mock_logger 178 ) 179 test_args = "" 180 profiler_args = "" 181 182 # Test crosfleet run for telemetry_Crosperf and crosperf_Wrapper benchmarks. 183 self.mock_label.crosfleet = True 184 reset() 185 self.runner.Run( 186 cros_machine, 187 self.mock_label, 188 self.crosperf_wrapper_bench, 189 test_args, 190 profiler_args, 191 ) 192 self.assertTrue(self.call_crosfleet_run) 193 self.assertFalse(self.call_test_that_run) 194 self.assertEqual( 195 self.crosfleet_run_args, 196 [self.mock_label, self.crosperf_wrapper_bench, "", ""], 197 ) 198 199 reset() 200 self.runner.Run( 201 cros_machine, 202 self.mock_label, 203 self.telemetry_crosperf_bench, 204 test_args, 205 profiler_args, 206 ) 207 self.assertTrue(self.call_crosfleet_run) 208 self.assertFalse(self.call_test_that_run) 209 self.assertEqual( 210 self.crosfleet_run_args, 211 [self.mock_label, self.telemetry_crosperf_bench, "", ""], 212 ) 213 214 # Test test_that run for telemetry_Crosperf and crosperf_Wrapper benchmarks. 215 self.mock_label.crosfleet = False 216 reset() 217 self.runner.Run( 218 cros_machine, 219 self.mock_label, 220 self.crosperf_wrapper_bench, 221 test_args, 222 profiler_args, 223 ) 224 self.assertTrue(self.call_test_that_run) 225 self.assertFalse(self.call_crosfleet_run) 226 self.assertEqual( 227 self.test_that_args, 228 [ 229 "fake_machine", 230 self.mock_label, 231 self.crosperf_wrapper_bench, 232 "", 233 "", 234 ], 235 ) 236 237 reset() 238 self.runner.Run( 239 cros_machine, 240 self.mock_label, 241 self.telemetry_crosperf_bench, 242 test_args, 243 profiler_args, 244 ) 245 self.assertTrue(self.call_test_that_run) 246 self.assertFalse(self.call_crosfleet_run) 247 self.assertEqual( 248 self.test_that_args, 249 [ 250 "fake_machine", 251 self.mock_label, 252 self.telemetry_crosperf_bench, 253 "", 254 "", 255 ], 256 ) 257 258 # Test tast run for tast benchmarks. 259 reset() 260 self.runner.Run(cros_machine, self.mock_label, self.tast_bench, "", "") 261 self.assertTrue(self.call_tast_run) 262 self.assertFalse(self.call_test_that_run) 263 self.assertFalse(self.call_crosfleet_run) 264 self.assertEqual( 265 self.tast_args, 266 ["fakelocalhost:1234", self.mock_label, self.tast_bench], 267 ) 268 269 def test_gen_test_args(self): 270 test_args = "--iterations=2" 271 perf_args = "record -a -e cycles" 272 273 # Test crosperf_Wrapper benchmarks arg list generation 274 args_list = [ 275 "test_args='--iterations=2'", 276 "dut_config='{}'", 277 "test=webgl", 278 ] 279 res = self.runner.GenTestArgs( 280 self.crosperf_wrapper_bench, test_args, "" 281 ) 282 self.assertCountEqual(res, args_list) 283 284 # Test telemetry_Crosperf benchmarks arg list generation 285 args_list = [ 286 "test_args='--iterations=2'", 287 "dut_config='{}'", 288 "test=octane", 289 "run_local=False", 290 ] 291 args_list.append(suite_runner.GetProfilerArgs(perf_args)) 292 res = self.runner.GenTestArgs( 293 self.telemetry_crosperf_bench, test_args, perf_args 294 ) 295 self.assertCountEqual(res, args_list) 296 297 @mock.patch.object(command_executer.CommandExecuter, "CrosRunCommand") 298 @mock.patch.object( 299 command_executer.CommandExecuter, "ChrootRunCommandWOutput" 300 ) 301 def test_tast_run(self, mock_chroot_runcmd, mock_cros_runcmd): 302 mock_chroot_runcmd.return_value = 0 303 self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd 304 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 305 res = self.runner.Tast_Run( 306 "lumpy1.cros", self.mock_label, self.tast_bench 307 ) 308 self.assertEqual(mock_cros_runcmd.call_count, 1) 309 self.assertEqual(mock_chroot_runcmd.call_count, 1) 310 self.assertEqual(res, 0) 311 self.assertEqual( 312 mock_cros_runcmd.call_args_list[0][0], 313 ("rm -rf /usr/local/autotest/results/*",), 314 ) 315 args_list = mock_chroot_runcmd.call_args_list[0][0] 316 args_dict = mock_chroot_runcmd.call_args_list[0][1] 317 self.assertEqual(len(args_list), 2) 318 self.assertEqual(args_dict["command_terminator"], self.mock_cmd_term) 319 320 @mock.patch.object(command_executer.CommandExecuter, "CrosRunCommand") 321 @mock.patch.object( 322 command_executer.CommandExecuter, "ChrootRunCommandWOutput" 323 ) 324 @mock.patch.object(logger.Logger, "LogFatal") 325 def test_test_that_run( 326 self, mock_log_fatal, mock_chroot_runcmd, mock_cros_runcmd 327 ): 328 mock_log_fatal.side_effect = SystemExit() 329 self.runner.logger.LogFatal = mock_log_fatal 330 # Test crosperf_Wrapper benchmarks cannot take perf_args 331 raised_exception = False 332 try: 333 self.runner.Test_That_Run( 334 "lumpy1.cros", 335 self.mock_label, 336 self.crosperf_wrapper_bench, 337 "", 338 "record -a -e cycles", 339 ) 340 except SystemExit: 341 raised_exception = True 342 self.assertTrue(raised_exception) 343 344 mock_chroot_runcmd.return_value = 0 345 self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd 346 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 347 res = self.runner.Test_That_Run( 348 "lumpy1.cros", 349 self.mock_label, 350 self.crosperf_wrapper_bench, 351 "--iterations=2", 352 "", 353 ) 354 self.assertEqual(mock_cros_runcmd.call_count, 1) 355 self.assertEqual(mock_chroot_runcmd.call_count, 1) 356 self.assertEqual(res, 0) 357 self.assertEqual( 358 mock_cros_runcmd.call_args_list[0][0], 359 ("rm -rf /usr/local/autotest/results/*",), 360 ) 361 args_list = mock_chroot_runcmd.call_args_list[0][0] 362 args_dict = mock_chroot_runcmd.call_args_list[0][1] 363 self.assertEqual(len(args_list), 2) 364 self.assertEqual(args_dict["command_terminator"], self.mock_cmd_term) 365 366 @mock.patch.object(command_executer.CommandExecuter, "RunCommandWOutput") 367 @mock.patch.object(json, "loads") 368 def test_crosfleet_run_client(self, mock_json_loads, mock_runcmd): 369 def FakeDownloadResult(l, task_id): 370 if l and task_id: 371 self.assertEqual(task_id, "12345") 372 return 0 373 374 mock_runcmd.return_value = ( 375 0, 376 "Created Swarming task https://swarming/task/b12345", 377 "", 378 ) 379 self.mock_cmd_exec.RunCommandWOutput = mock_runcmd 380 381 mock_json_loads.return_value = { 382 "child-results": [ 383 { 384 "success": True, 385 "task-run-url": "https://swarming/task?id=12345", 386 } 387 ] 388 } 389 self.mock_json.loads = mock_json_loads 390 391 self.mock_label.crosfleet = True 392 self.runner.DownloadResult = FakeDownloadResult 393 res = self.runner.Crosfleet_Run( 394 self.mock_label, self.crosperf_wrapper_bench, "", "" 395 ) 396 ret_tup = (0, "\nResults placed in tmp/swarming-12345\n", "") 397 self.assertEqual(res, ret_tup) 398 self.assertEqual(mock_runcmd.call_count, 2) 399 400 args_list = mock_runcmd.call_args_list[0][0] 401 args_dict = mock_runcmd.call_args_list[0][1] 402 self.assertEqual(len(args_list), 1) 403 self.assertEqual(args_dict["command_terminator"], self.mock_cmd_term) 404 405 args_list = mock_runcmd.call_args_list[1][0] 406 self.assertEqual(args_list[0], ("crosfleet wait-task 12345")) 407 self.assertEqual(args_dict["command_terminator"], self.mock_cmd_term) 408 409 410if __name__ == "__main__": 411 unittest.main() 412