#!/usr/bin/python2 # Copyright (c) 2013 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. """Unit test for experiment_factory.py""" from __future__ import print_function import StringIO import socket import mock import unittest from cros_utils.file_utils import FileUtils from experiment_factory import ExperimentFactory from experiment_file import ExperimentFile import test_flag import benchmark import experiment_factory import settings_factory EXPERIMENT_FILE_1 = """ board: x86-alex remote: chromeos-alex3 benchmark: PageCycler { iterations: 3 } image1 { chromeos_image: /usr/local/google/cros_image1.bin } image2 { chromeos_image: /usr/local/google/cros_image2.bin } """ # pylint: disable=too-many-function-args class ExperimentFactoryTest(unittest.TestCase): """Class for running experiment factory unittests.""" def setUp(self): self.append_benchmark_call_args = [] def testLoadExperimentFile1(self): experiment_file = ExperimentFile(StringIO.StringIO(EXPERIMENT_FILE_1)) exp = ExperimentFactory().GetExperiment(experiment_file, working_directory='', log_dir='') self.assertEqual(exp.remote, ['chromeos-alex3']) self.assertEqual(len(exp.benchmarks), 1) self.assertEqual(exp.benchmarks[0].name, 'PageCycler') self.assertEqual(exp.benchmarks[0].test_name, 'PageCycler') self.assertEqual(exp.benchmarks[0].iterations, 3) self.assertEqual(len(exp.labels), 2) self.assertEqual(exp.labels[0].chromeos_image, '/usr/local/google/cros_image1.bin') self.assertEqual(exp.labels[0].board, 'x86-alex') def test_append_benchmark_set(self): ef = ExperimentFactory() bench_list = [] ef.AppendBenchmarkSet(bench_list, experiment_factory.telemetry_perfv2_tests, '', 1, False, '', 'telemetry_Crosperf', False, 0, False) self.assertEqual( len(bench_list), len(experiment_factory.telemetry_perfv2_tests)) self.assertTrue(type(bench_list[0]) is benchmark.Benchmark) bench_list = [] ef.AppendBenchmarkSet(bench_list, experiment_factory.telemetry_pagecycler_tests, '', 1, False, '', 'telemetry_Crosperf', False, 0, False) self.assertEqual( len(bench_list), len(experiment_factory.telemetry_pagecycler_tests)) self.assertTrue(type(bench_list[0]) is benchmark.Benchmark) bench_list = [] ef.AppendBenchmarkSet( bench_list, experiment_factory.telemetry_toolchain_perf_tests, '', 1, False, '', 'telemetry_Crosperf', False, 0, False) self.assertEqual( len(bench_list), len(experiment_factory.telemetry_toolchain_perf_tests)) self.assertTrue(type(bench_list[0]) is benchmark.Benchmark) @mock.patch.object(socket, 'gethostname') def test_get_experiment(self, mock_socket): test_flag.SetTestMode(False) self.append_benchmark_call_args = [] def FakeAppendBenchmarkSet(bench_list, set_list, args, iters, rm_ch, perf_args, suite, show_all): 'Helper function for test_get_experiment' arg_list = [bench_list, set_list, args, iters, rm_ch, perf_args, suite, show_all] self.append_benchmark_call_args.append(arg_list) def FakeGetDefaultRemotes(board): if not board: return [] return ['fake_chromeos_machine1.cros', 'fake_chromeos_machine2.cros'] def FakeGetXbuddyPath(build, board, chroot, log_level): if not build or not board or not chroot or not log_level: return '' return 'fake_image_path' ef = ExperimentFactory() ef.AppendBenchmarkSet = FakeAppendBenchmarkSet ef.GetDefaultRemotes = FakeGetDefaultRemotes label_settings = settings_factory.LabelSettings('image_label') benchmark_settings = settings_factory.BenchmarkSettings('bench_test') global_settings = settings_factory.GlobalSettings('test_name') label_settings.GetXbuddyPath = FakeGetXbuddyPath mock_experiment_file = ExperimentFile(StringIO.StringIO('')) mock_experiment_file.all_settings = [] test_flag.SetTestMode(True) # Basic test. global_settings.SetField('name', 'unittest_test') global_settings.SetField('board', 'lumpy') global_settings.SetField('remote', '123.45.67.89 123.45.76.80') benchmark_settings.SetField('test_name', 'kraken') benchmark_settings.SetField('suite', 'telemetry_Crosperf') benchmark_settings.SetField('iterations', 1) label_settings.SetField( 'chromeos_image', 'chromeos/src/build/images/lumpy/latest/chromiumos_test_image.bin') label_settings.SetField('chrome_src', '/usr/local/google/home/chrome-top') mock_experiment_file.global_settings = global_settings mock_experiment_file.all_settings.append(label_settings) mock_experiment_file.all_settings.append(benchmark_settings) mock_experiment_file.all_settings.append(global_settings) mock_socket.return_value = '' # First test. General test. exp = ef.GetExperiment(mock_experiment_file, '', '') self.assertEqual(exp.remote, ['123.45.67.89', '123.45.76.80']) self.assertEqual(exp.cache_conditions, [0, 2, 1]) self.assertEqual(exp.log_level, 'average') self.assertEqual(len(exp.benchmarks), 1) self.assertEqual(exp.benchmarks[0].name, 'kraken') self.assertEqual(exp.benchmarks[0].test_name, 'kraken') self.assertEqual(exp.benchmarks[0].iterations, 1) self.assertEqual(exp.benchmarks[0].suite, 'telemetry_Crosperf') self.assertFalse(exp.benchmarks[0].show_all_results) self.assertEqual(len(exp.labels), 1) self.assertEqual(exp.labels[0].chromeos_image, 'chromeos/src/build/images/lumpy/latest/' 'chromiumos_test_image.bin') self.assertEqual(exp.labels[0].board, 'lumpy') # Second test: Remotes listed in labels. test_flag.SetTestMode(True) label_settings.SetField('remote', 'chromeos1.cros chromeos2.cros') exp = ef.GetExperiment(mock_experiment_file, '', '') self.assertEqual(exp.remote, ['chromeos1.cros', 'chromeos2.cros', '123.45.67.89', '123.45.76.80']) # Third test: Automatic fixing of bad logging_level param: global_settings.SetField('logging_level', 'really loud!') exp = ef.GetExperiment(mock_experiment_file, '', '') self.assertEqual(exp.log_level, 'verbose') # Fourth test: Setting cache conditions; only 1 remote with "same_machine" global_settings.SetField('rerun_if_failed', 'true') global_settings.SetField('rerun', 'true') global_settings.SetField('same_machine', 'true') global_settings.SetField('same_specs', 'true') self.assertRaises(Exception, ef.GetExperiment, mock_experiment_file, '', '') label_settings.SetField('remote', '') global_settings.SetField('remote', '123.45.67.89') exp = ef.GetExperiment(mock_experiment_file, '', '') self.assertEqual(exp.cache_conditions, [0, 2, 3, 4, 6, 1]) # Fifth Test: Adding a second label; calling GetXbuddyPath; omitting all # remotes (Call GetDefaultRemotes). mock_socket.return_value = 'test.corp.google.com' global_settings.SetField('remote', '') global_settings.SetField('same_machine', 'false') label_settings_2 = settings_factory.LabelSettings('official_image_label') label_settings_2.SetField('chromeos_root', 'chromeos') label_settings_2.SetField('build', 'official-dev') label_settings_2.GetXbuddyPath = FakeGetXbuddyPath mock_experiment_file.all_settings.append(label_settings_2) exp = ef.GetExperiment(mock_experiment_file, '', '') self.assertEqual(len(exp.labels), 2) self.assertEqual(exp.labels[1].chromeos_image, 'fake_image_path') self.assertEqual(exp.remote, ['fake_chromeos_machine1.cros', 'fake_chromeos_machine2.cros']) def test_get_default_remotes(self): board_list = ['x86-alex', 'lumpy', 'stumpy', 'parrot', 'daisy', 'peach_pit', 'peppy', 'squawks'] ef = ExperimentFactory() self.assertRaises(Exception, ef.GetDefaultRemotes, 'bad-board') # Verify that we have entries for every board, and that we get three # machines back for each board. for b in board_list: remotes = ef.GetDefaultRemotes(b) self.assertEqual(len(remotes), 3) if __name__ == '__main__': FileUtils.Configure(True) test_flag.SetTestMode(True) unittest.main()