1#!/usr/bin/env python2 2# 3# Copyright 2014 Google Inc. All Rights Reserved. 4"""Unittest for suite_runner.""" 5 6from __future__ import print_function 7 8import os.path 9import time 10 11import mock 12import unittest 13 14import suite_runner 15import label 16import test_flag 17 18from benchmark import Benchmark 19 20from cros_utils import command_executer 21from cros_utils import logger 22 23 24class SuiteRunnerTest(unittest.TestCase): 25 """Class of SuiteRunner test.""" 26 real_logger = logger.GetLogger() 27 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', 'lumpy_chromeos_image', '', '/tmp/chromeos', 'lumpy', 33 ['lumpy1.cros', 'lumpy.cros2'], '', '', False, 'average', 'gcc', '') 34 telemetry_crosperf_bench = Benchmark( 35 'b1_test', # name 36 'octane', # test_name 37 '', # test_args 38 3, # iterations 39 False, # rm_chroot_tmp 40 'record -e cycles', # perf_args 41 'telemetry_Crosperf', # suite 42 True) # show_all_results 43 44 test_that_bench = Benchmark( 45 'b2_test', # name 46 'octane', # test_name 47 '', # test_args 48 3, # iterations 49 False, # rm_chroot_tmp 50 'record -e cycles') # perf_args 51 52 telemetry_bench = Benchmark( 53 'b3_test', # name 54 'octane', # test_name 55 '', # test_args 56 3, # iterations 57 False, # rm_chroot_tmp 58 'record -e cycles', # perf_args 59 'telemetry', # suite 60 False) # show_all_results 61 62 def __init__(self, *args, **kwargs): 63 super(SuiteRunnerTest, self).__init__(*args, **kwargs) 64 self.call_test_that_run = False 65 self.pin_governor_args = [] 66 self.test_that_args = [] 67 self.telemetry_run_args = [] 68 self.telemetry_crosperf_args = [] 69 self.call_telemetry_crosperf_run = False 70 self.call_pin_governor = False 71 self.call_telemetry_run = False 72 73 def setUp(self): 74 self.runner = suite_runner.SuiteRunner( 75 self.mock_logger, 'verbose', self.mock_cmd_exec, self.mock_cmd_term) 76 77 def test_get_profiler_args(self): 78 input_str = ('--profiler=custom_perf --profiler_args=\'perf_options' 79 '="record -a -e cycles,instructions"\'') 80 output_str = ("profiler=custom_perf profiler_args='record -a -e " 81 "cycles,instructions'") 82 res = suite_runner.GetProfilerArgs(input_str) 83 self.assertEqual(res, output_str) 84 85 def test_run(self): 86 87 def reset(): 88 self.call_pin_governor = False 89 self.call_test_that_run = False 90 self.call_telemetry_run = False 91 self.call_telemetry_crosperf_run = False 92 self.pin_governor_args = [] 93 self.test_that_args = [] 94 self.telemetry_run_args = [] 95 self.telemetry_crosperf_args = [] 96 97 def FakePinGovernor(machine, chroot): 98 self.call_pin_governor = True 99 self.pin_governor_args = [machine, chroot] 100 101 def FakeTelemetryRun(machine, test_label, benchmark, profiler_args): 102 self.telemetry_run_args = [machine, test_label, benchmark, profiler_args] 103 self.call_telemetry_run = True 104 return 'Ran FakeTelemetryRun' 105 106 def FakeTelemetryCrosperfRun(machine, test_label, benchmark, test_args, 107 profiler_args): 108 self.telemetry_crosperf_args = [ 109 machine, test_label, benchmark, test_args, profiler_args 110 ] 111 self.call_telemetry_crosperf_run = True 112 return 'Ran FakeTelemetryCrosperfRun' 113 114 def FakeTestThatRun(machine, test_label, benchmark, test_args, 115 profiler_args): 116 self.test_that_args = [ 117 machine, test_label, benchmark, test_args, profiler_args 118 ] 119 self.call_test_that_run = True 120 return 'Ran FakeTestThatRun' 121 122 self.runner.PinGovernorExecutionFrequencies = FakePinGovernor 123 self.runner.Telemetry_Run = FakeTelemetryRun 124 self.runner.Telemetry_Crosperf_Run = FakeTelemetryCrosperfRun 125 self.runner.Test_That_Run = FakeTestThatRun 126 127 machine = 'fake_machine' 128 test_args = '' 129 profiler_args = '' 130 reset() 131 self.runner.Run(machine, self.mock_label, self.telemetry_bench, test_args, 132 profiler_args) 133 self.assertTrue(self.call_pin_governor) 134 self.assertTrue(self.call_telemetry_run) 135 self.assertFalse(self.call_test_that_run) 136 self.assertFalse(self.call_telemetry_crosperf_run) 137 self.assertEqual( 138 self.telemetry_run_args, 139 ['fake_machine', self.mock_label, self.telemetry_bench, '']) 140 141 reset() 142 self.runner.Run(machine, self.mock_label, self.test_that_bench, test_args, 143 profiler_args) 144 self.assertTrue(self.call_pin_governor) 145 self.assertFalse(self.call_telemetry_run) 146 self.assertTrue(self.call_test_that_run) 147 self.assertFalse(self.call_telemetry_crosperf_run) 148 self.assertEqual( 149 self.test_that_args, 150 ['fake_machine', self.mock_label, self.test_that_bench, '', '']) 151 152 reset() 153 self.runner.Run(machine, self.mock_label, self.telemetry_crosperf_bench, 154 test_args, profiler_args) 155 self.assertTrue(self.call_pin_governor) 156 self.assertFalse(self.call_telemetry_run) 157 self.assertFalse(self.call_test_that_run) 158 self.assertTrue(self.call_telemetry_crosperf_run) 159 self.assertEqual(self.telemetry_crosperf_args, [ 160 'fake_machine', self.mock_label, self.telemetry_crosperf_bench, '', '' 161 ]) 162 163 @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') 164 def test_pin_governor_execution_frequencies(self, mock_cros_runcmd): 165 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 166 self.runner.PinGovernorExecutionFrequencies('lumpy1.cros', '/tmp/chromeos') 167 self.assertEqual(mock_cros_runcmd.call_count, 1) 168 cmd = mock_cros_runcmd.call_args_list[0][0] 169 # pyformat: disable 170 set_cpu_cmd = ( 171 'set -e && ' 172 # Disable Turbo in Intel pstate driver 173 'if [[ -e /sys/devices/system/cpu/intel_pstate/no_turbo ]]; then ' 174 ' if grep -q 0 /sys/devices/system/cpu/intel_pstate/no_turbo; then ' 175 ' echo -n 1 > /sys/devices/system/cpu/intel_pstate/no_turbo; ' 176 ' fi; ' 177 'fi; ' 178 # Set governor to performance for each cpu 179 'for f in /sys/devices/system/cpu/cpu*/cpufreq; do ' 180 'cd $f; ' 181 'echo performance > scaling_governor; ' 182 'done' 183 ) 184 # pyformat: enable 185 self.assertEqual(cmd, (set_cpu_cmd,)) 186 187 @mock.patch.object(time, 'sleep') 188 @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') 189 def test_reboot_machine(self, mock_cros_runcmd, mock_sleep): 190 191 def FakePinGovernor(machine_name, chromeos_root): 192 if machine_name or chromeos_root: 193 pass 194 195 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 196 self.runner.PinGovernorExecutionFrequencies = FakePinGovernor 197 self.runner.RestartUI('lumpy1.cros', '/tmp/chromeos') 198 self.assertEqual(mock_cros_runcmd.call_count, 1) 199 self.assertEqual(mock_cros_runcmd.call_args_list[0][0], 200 ('stop ui; sleep 5; start ui',)) 201 202 @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') 203 @mock.patch.object(command_executer.CommandExecuter, 204 'ChrootRunCommandWOutput') 205 def test_test_that_run(self, mock_chroot_runcmd, mock_cros_runcmd): 206 207 def FakeRebootMachine(machine, chroot): 208 if machine or chroot: 209 pass 210 211 def FakeLogMsg(fd, termfd, msg, flush=True): 212 if fd or termfd or msg or flush: 213 pass 214 215 save_log_msg = self.real_logger.LogMsg 216 self.real_logger.LogMsg = FakeLogMsg 217 self.runner.logger = self.real_logger 218 self.runner.RebootMachine = FakeRebootMachine 219 220 raised_exception = False 221 try: 222 self.runner.Test_That_Run('lumpy1.cros', self.mock_label, 223 self.test_that_bench, '', 'record -a -e cycles') 224 except SystemExit: 225 raised_exception = True 226 self.assertTrue(raised_exception) 227 228 mock_chroot_runcmd.return_value = 0 229 self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd 230 self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd 231 res = self.runner.Test_That_Run('lumpy1.cros', self.mock_label, 232 self.test_that_bench, '--iterations=2', '') 233 self.assertEqual(mock_cros_runcmd.call_count, 2) 234 self.assertEqual(mock_chroot_runcmd.call_count, 1) 235 self.assertEqual(res, 0) 236 self.assertEqual(mock_cros_runcmd.call_args_list[0][0], 237 ('rm -rf /usr/local/autotest/results/*',)) 238 args_list = mock_chroot_runcmd.call_args_list[0][0] 239 args_dict = mock_chroot_runcmd.call_args_list[0][1] 240 self.assertEqual(len(args_list), 2) 241 self.assertEqual(args_list[0], '/tmp/chromeos') 242 self.assertEqual(args_list[1], ('/usr/bin/test_that ' 243 '--fast --board=lumpy ' 244 '--iterations=2 lumpy1.cros octane')) 245 self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term) 246 self.real_logger.LogMsg = save_log_msg 247 248 @mock.patch.object(os.path, 'isdir') 249 @mock.patch.object(command_executer.CommandExecuter, 250 'ChrootRunCommandWOutput') 251 def test_telemetry_crosperf_run(self, mock_chroot_runcmd, mock_isdir): 252 253 mock_isdir.return_value = True 254 mock_chroot_runcmd.return_value = 0 255 self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd 256 profiler_args = ('--profiler=custom_perf --profiler_args=\'perf_options' 257 '="record -a -e cycles,instructions"\'') 258 res = self.runner.Telemetry_Crosperf_Run('lumpy1.cros', self.mock_label, 259 self.telemetry_crosperf_bench, '', 260 profiler_args) 261 self.assertEqual(res, 0) 262 self.assertEqual(mock_chroot_runcmd.call_count, 1) 263 args_list = mock_chroot_runcmd.call_args_list[0][0] 264 args_dict = mock_chroot_runcmd.call_args_list[0][1] 265 self.assertEqual(args_list[0], '/tmp/chromeos') 266 self.assertEqual(args_list[1], 267 ('/usr/bin/test_that --autotest_dir ' 268 '~/trunk/src/third_party/autotest/files ' 269 ' --board=lumpy --args=" run_local=False test=octane ' 270 'profiler=custom_perf profiler_args=\'record -a -e ' 271 'cycles,instructions\'" lumpy1.cros telemetry_Crosperf')) 272 self.assertEqual(args_dict['cros_sdk_options'], 273 ('--no-ns-pid --chrome_root= ' 274 '--chrome_root_mount=/tmp/chrome_root ' 275 'FEATURES="-usersandbox" CHROME_ROOT=/tmp/chrome_root')) 276 self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term) 277 self.assertEqual(len(args_dict), 2) 278 279 @mock.patch.object(os.path, 'isdir') 280 @mock.patch.object(os.path, 'exists') 281 @mock.patch.object(command_executer.CommandExecuter, 'RunCommandWOutput') 282 def test_telemetry_run(self, mock_runcmd, mock_exists, mock_isdir): 283 284 def FakeLogMsg(fd, termfd, msg, flush=True): 285 if fd or termfd or msg or flush: 286 pass 287 288 save_log_msg = self.real_logger.LogMsg 289 self.real_logger.LogMsg = FakeLogMsg 290 mock_runcmd.return_value = 0 291 292 self.mock_cmd_exec.RunCommandWOutput = mock_runcmd 293 self.runner.logger = self.real_logger 294 295 profiler_args = ('--profiler=custom_perf --profiler_args=\'perf_options' 296 '="record -a -e cycles,instructions"\'') 297 298 raises_exception = False 299 mock_isdir.return_value = False 300 try: 301 self.runner.Telemetry_Run('lumpy1.cros', self.mock_label, 302 self.telemetry_bench, '') 303 except SystemExit: 304 raises_exception = True 305 self.assertTrue(raises_exception) 306 307 raises_exception = False 308 mock_isdir.return_value = True 309 mock_exists.return_value = False 310 try: 311 self.runner.Telemetry_Run('lumpy1.cros', self.mock_label, 312 self.telemetry_bench, '') 313 except SystemExit: 314 raises_exception = True 315 self.assertTrue(raises_exception) 316 317 raises_exception = False 318 mock_isdir.return_value = True 319 mock_exists.return_value = True 320 try: 321 self.runner.Telemetry_Run('lumpy1.cros', self.mock_label, 322 self.telemetry_bench, profiler_args) 323 except SystemExit: 324 raises_exception = True 325 self.assertTrue(raises_exception) 326 327 test_flag.SetTestMode(True) 328 res = self.runner.Telemetry_Run('lumpy1.cros', self.mock_label, 329 self.telemetry_bench, '') 330 self.assertEqual(res, 0) 331 self.assertEqual(mock_runcmd.call_count, 1) 332 self.assertEqual( 333 mock_runcmd.call_args_list[0][0], 334 (('cd src/tools/perf && ./run_measurement ' 335 '--browser=cros-chrome --output-format=csv ' 336 '--remote=lumpy1.cros --identity /tmp/chromeos/src/scripts' 337 '/mod_for_test_scripts/ssh_keys/testing_rsa octane '),)) 338 339 self.real_logger.LogMsg = save_log_msg 340 341 342if __name__ == '__main__': 343 unittest.main() 344