You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
302 lines
11 KiB
302 lines
11 KiB
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
#
|
|
# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
|
|
# Use of this source code is governed by a BSD-style license that can be
|
|
# found in the LICENSE file.
|
|
|
|
"""Unittest for suite_runner."""
|
|
|
|
from __future__ import print_function
|
|
|
|
import json
|
|
|
|
import unittest
|
|
import unittest.mock as mock
|
|
|
|
import suite_runner
|
|
import label
|
|
|
|
from benchmark import Benchmark
|
|
|
|
from cros_utils import command_executer
|
|
from cros_utils import logger
|
|
from machine_manager import MockCrosMachine
|
|
|
|
|
|
class SuiteRunnerTest(unittest.TestCase):
|
|
"""Class of SuiteRunner test."""
|
|
mock_json = mock.Mock(spec=json)
|
|
mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter)
|
|
mock_cmd_term = mock.Mock(spec=command_executer.CommandTerminator)
|
|
mock_logger = mock.Mock(spec=logger.Logger)
|
|
mock_label = label.MockLabel('lumpy', 'build', 'lumpy_chromeos_image', '', '',
|
|
'/tmp/chromeos', 'lumpy',
|
|
['lumpy1.cros', 'lumpy.cros2'], '', '', False,
|
|
'average', 'gcc', False, '')
|
|
telemetry_crosperf_bench = Benchmark(
|
|
'b1_test', # name
|
|
'octane', # test_name
|
|
'', # test_args
|
|
3, # iterations
|
|
False, # rm_chroot_tmp
|
|
'record -e cycles', # perf_args
|
|
'telemetry_Crosperf', # suite
|
|
True) # show_all_results
|
|
|
|
crosperf_wrapper_bench = Benchmark(
|
|
'b2_test', # name
|
|
'webgl', # test_name
|
|
'', # test_args
|
|
3, # iterations
|
|
False, # rm_chroot_tmp
|
|
'', # perf_args
|
|
'crosperf_Wrapper') # suite
|
|
|
|
tast_bench = Benchmark(
|
|
'b3_test', # name
|
|
'platform.ReportDiskUsage', # test_name
|
|
'', # test_args
|
|
1, # iterations
|
|
False, # rm_chroot_tmp
|
|
'', # perf_args
|
|
'tast') # suite
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(SuiteRunnerTest, self).__init__(*args, **kwargs)
|
|
self.skylab_run_args = []
|
|
self.test_that_args = []
|
|
self.tast_args = []
|
|
self.call_skylab_run = False
|
|
self.call_test_that_run = False
|
|
self.call_tast_run = False
|
|
|
|
def setUp(self):
|
|
self.runner = suite_runner.SuiteRunner(
|
|
{}, self.mock_logger, 'verbose', self.mock_cmd_exec, self.mock_cmd_term)
|
|
|
|
def test_get_profiler_args(self):
|
|
input_str = ("--profiler=custom_perf --profiler_args='perf_options"
|
|
'="record -a -e cycles,instructions"\'')
|
|
output_str = ("profiler=custom_perf profiler_args='record -a -e "
|
|
"cycles,instructions'")
|
|
res = suite_runner.GetProfilerArgs(input_str)
|
|
self.assertEqual(res, output_str)
|
|
|
|
def test_get_dut_config_args(self):
|
|
dut_config = {'enable_aslr': False, 'top_interval': 1.0}
|
|
output_str = ('dut_config='
|
|
"'"
|
|
'{"enable_aslr": '
|
|
'false, "top_interval": 1.0}'
|
|
"'"
|
|
'')
|
|
res = suite_runner.GetDutConfigArgs(dut_config)
|
|
self.assertEqual(res, output_str)
|
|
|
|
def test_run(self):
|
|
|
|
def reset():
|
|
self.test_that_args = []
|
|
self.skylab_run_args = []
|
|
self.tast_args = []
|
|
self.call_test_that_run = False
|
|
self.call_skylab_run = False
|
|
self.call_tast_run = False
|
|
|
|
def FakeSkylabRun(test_label, benchmark, test_args, profiler_args):
|
|
self.skylab_run_args = [test_label, benchmark, test_args, profiler_args]
|
|
self.call_skylab_run = True
|
|
return 'Ran FakeSkylabRun'
|
|
|
|
def FakeTestThatRun(machine, test_label, benchmark, test_args,
|
|
profiler_args):
|
|
self.test_that_args = [
|
|
machine, test_label, benchmark, test_args, profiler_args
|
|
]
|
|
self.call_test_that_run = True
|
|
return 'Ran FakeTestThatRun'
|
|
|
|
def FakeTastRun(machine, test_label, benchmark):
|
|
self.tast_args = [machine, test_label, benchmark]
|
|
self.call_tast_run = True
|
|
return 'Ran FakeTastRun'
|
|
|
|
self.runner.Skylab_Run = FakeSkylabRun
|
|
self.runner.Test_That_Run = FakeTestThatRun
|
|
self.runner.Tast_Run = FakeTastRun
|
|
|
|
self.runner.dut_config['enable_aslr'] = False
|
|
self.runner.dut_config['cooldown_time'] = 0
|
|
self.runner.dut_config['governor'] = 'fake_governor'
|
|
self.runner.dut_config['cpu_freq_pct'] = 65
|
|
self.runner.dut_config['intel_pstate'] = 'no_hwp'
|
|
machine = 'fake_machine'
|
|
cros_machine = MockCrosMachine(machine, self.mock_label.chromeos_root,
|
|
self.mock_logger)
|
|
test_args = ''
|
|
profiler_args = ''
|
|
|
|
# Test skylab run for telemetry_Crosperf and crosperf_Wrapper benchmarks.
|
|
self.mock_label.skylab = True
|
|
reset()
|
|
self.runner.Run(cros_machine, self.mock_label, self.crosperf_wrapper_bench,
|
|
test_args, profiler_args)
|
|
self.assertTrue(self.call_skylab_run)
|
|
self.assertFalse(self.call_test_that_run)
|
|
self.assertEqual(self.skylab_run_args,
|
|
[self.mock_label, self.crosperf_wrapper_bench, '', ''])
|
|
|
|
reset()
|
|
self.runner.Run(cros_machine, self.mock_label,
|
|
self.telemetry_crosperf_bench, test_args, profiler_args)
|
|
self.assertTrue(self.call_skylab_run)
|
|
self.assertFalse(self.call_test_that_run)
|
|
self.assertEqual(self.skylab_run_args,
|
|
[self.mock_label, self.telemetry_crosperf_bench, '', ''])
|
|
|
|
# Test test_that run for telemetry_Crosperf and crosperf_Wrapper benchmarks.
|
|
self.mock_label.skylab = False
|
|
reset()
|
|
self.runner.Run(cros_machine, self.mock_label, self.crosperf_wrapper_bench,
|
|
test_args, profiler_args)
|
|
self.assertTrue(self.call_test_that_run)
|
|
self.assertFalse(self.call_skylab_run)
|
|
self.assertEqual(
|
|
self.test_that_args,
|
|
['fake_machine', self.mock_label, self.crosperf_wrapper_bench, '', ''])
|
|
|
|
reset()
|
|
self.runner.Run(cros_machine, self.mock_label,
|
|
self.telemetry_crosperf_bench, test_args, profiler_args)
|
|
self.assertTrue(self.call_test_that_run)
|
|
self.assertFalse(self.call_skylab_run)
|
|
self.assertEqual(self.test_that_args, [
|
|
'fake_machine', self.mock_label, self.telemetry_crosperf_bench, '', ''
|
|
])
|
|
|
|
# Test tast run for tast benchmarks.
|
|
reset()
|
|
self.runner.Run(cros_machine, self.mock_label, self.tast_bench, '', '')
|
|
self.assertTrue(self.call_tast_run)
|
|
self.assertFalse(self.call_test_that_run)
|
|
self.assertFalse(self.call_skylab_run)
|
|
self.assertEqual(self.tast_args,
|
|
['fake_machine', self.mock_label, self.tast_bench])
|
|
|
|
def test_gen_test_args(self):
|
|
test_args = '--iterations=2'
|
|
perf_args = 'record -a -e cycles'
|
|
|
|
# Test crosperf_Wrapper benchmarks arg list generation
|
|
args_list = ["test_args='--iterations=2'", "dut_config='{}'", 'test=webgl']
|
|
res = self.runner.GenTestArgs(self.crosperf_wrapper_bench, test_args, '')
|
|
self.assertCountEqual(res, args_list)
|
|
|
|
# Test telemetry_Crosperf benchmarks arg list generation
|
|
args_list = [
|
|
"test_args='--iterations=2'", "dut_config='{}'", 'test=octane',
|
|
'run_local=False'
|
|
]
|
|
args_list.append(suite_runner.GetProfilerArgs(perf_args))
|
|
res = self.runner.GenTestArgs(self.telemetry_crosperf_bench, test_args,
|
|
perf_args)
|
|
self.assertCountEqual(res, args_list)
|
|
|
|
@mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand')
|
|
@mock.patch.object(command_executer.CommandExecuter,
|
|
'ChrootRunCommandWOutput')
|
|
def test_tast_run(self, mock_chroot_runcmd, mock_cros_runcmd):
|
|
mock_chroot_runcmd.return_value = 0
|
|
self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd
|
|
self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd
|
|
res = self.runner.Tast_Run('lumpy1.cros', self.mock_label, self.tast_bench)
|
|
self.assertEqual(mock_cros_runcmd.call_count, 1)
|
|
self.assertEqual(mock_chroot_runcmd.call_count, 1)
|
|
self.assertEqual(res, 0)
|
|
self.assertEqual(mock_cros_runcmd.call_args_list[0][0],
|
|
('rm -rf /usr/local/autotest/results/*',))
|
|
args_list = mock_chroot_runcmd.call_args_list[0][0]
|
|
args_dict = mock_chroot_runcmd.call_args_list[0][1]
|
|
self.assertEqual(len(args_list), 2)
|
|
self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term)
|
|
|
|
@mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand')
|
|
@mock.patch.object(command_executer.CommandExecuter,
|
|
'ChrootRunCommandWOutput')
|
|
@mock.patch.object(logger.Logger, 'LogFatal')
|
|
def test_test_that_run(self, mock_log_fatal, mock_chroot_runcmd,
|
|
mock_cros_runcmd):
|
|
mock_log_fatal.side_effect = SystemExit()
|
|
self.runner.logger.LogFatal = mock_log_fatal
|
|
# Test crosperf_Wrapper benchmarks cannot take perf_args
|
|
raised_exception = False
|
|
try:
|
|
self.runner.Test_That_Run('lumpy1.cros', self.mock_label,
|
|
self.crosperf_wrapper_bench, '',
|
|
'record -a -e cycles')
|
|
except SystemExit:
|
|
raised_exception = True
|
|
self.assertTrue(raised_exception)
|
|
|
|
mock_chroot_runcmd.return_value = 0
|
|
self.mock_cmd_exec.ChrootRunCommandWOutput = mock_chroot_runcmd
|
|
self.mock_cmd_exec.CrosRunCommand = mock_cros_runcmd
|
|
res = self.runner.Test_That_Run('lumpy1.cros', self.mock_label,
|
|
self.crosperf_wrapper_bench,
|
|
'--iterations=2', '')
|
|
self.assertEqual(mock_cros_runcmd.call_count, 1)
|
|
self.assertEqual(mock_chroot_runcmd.call_count, 1)
|
|
self.assertEqual(res, 0)
|
|
self.assertEqual(mock_cros_runcmd.call_args_list[0][0],
|
|
('rm -rf /usr/local/autotest/results/*',))
|
|
args_list = mock_chroot_runcmd.call_args_list[0][0]
|
|
args_dict = mock_chroot_runcmd.call_args_list[0][1]
|
|
self.assertEqual(len(args_list), 2)
|
|
self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term)
|
|
|
|
@mock.patch.object(command_executer.CommandExecuter, 'RunCommandWOutput')
|
|
@mock.patch.object(json, 'loads')
|
|
def test_skylab_run_client(self, mock_json_loads, mock_runcmd):
|
|
|
|
def FakeDownloadResult(l, task_id):
|
|
if l and task_id:
|
|
self.assertEqual(task_id, '12345')
|
|
return 0
|
|
|
|
mock_runcmd.return_value = (
|
|
0,
|
|
'Created Swarming task https://swarming/task/b12345',
|
|
'',
|
|
)
|
|
self.mock_cmd_exec.RunCommandWOutput = mock_runcmd
|
|
|
|
mock_json_loads.return_value = {
|
|
'child-results': [{
|
|
'success': True,
|
|
'task-run-url': 'https://swarming/task?id=12345'
|
|
}]
|
|
}
|
|
self.mock_json.loads = mock_json_loads
|
|
|
|
self.mock_label.skylab = True
|
|
self.runner.DownloadResult = FakeDownloadResult
|
|
res = self.runner.Skylab_Run(self.mock_label, self.crosperf_wrapper_bench,
|
|
'', '')
|
|
ret_tup = (0, '\nResults placed in tmp/swarming-12345\n', '')
|
|
self.assertEqual(res, ret_tup)
|
|
self.assertEqual(mock_runcmd.call_count, 2)
|
|
|
|
args_list = mock_runcmd.call_args_list[0][0]
|
|
args_dict = mock_runcmd.call_args_list[0][1]
|
|
self.assertEqual(len(args_list), 1)
|
|
self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term)
|
|
|
|
args_list = mock_runcmd.call_args_list[1][0]
|
|
self.assertEqual(args_list[0], ('skylab wait-task 12345'))
|
|
self.assertEqual(args_dict['command_terminator'], self.mock_cmd_term)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|