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