diff options
Diffstat (limited to 'crosperf/machine_manager_unittest.py')
| -rwxr-xr-x | crosperf/machine_manager_unittest.py | 1456 |
1 files changed, 782 insertions, 674 deletions
diff --git a/crosperf/machine_manager_unittest.py b/crosperf/machine_manager_unittest.py index f47cc881..6324a227 100755 --- a/crosperf/machine_manager_unittest.py +++ b/crosperf/machine_manager_unittest.py @@ -1,493 +1,574 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -# Copyright (c) 2012 The Chromium OS Authors. All rights reserved. +# Copyright 2012 The ChromiumOS Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Unittest for machine_manager.""" -from __future__ import print_function +import hashlib import os.path import time -import hashlib import unittest import unittest.mock as mock -import label -import machine_manager -import image_checksummer -import test_flag - from benchmark import Benchmark from benchmark_run import MockBenchmarkRun from cros_utils import command_executer from cros_utils import logger +import image_checksummer +import label +import machine_manager +import test_flag + # pylint: disable=protected-access class MyMachineManager(machine_manager.MachineManager): - """Machine manager for test.""" - - def __init__(self, chromeos_root): - super(MyMachineManager, self).__init__(chromeos_root, 0, 'average', '') - - def _TryToLockMachine(self, cros_machine): - self._machines.append(cros_machine) - cros_machine.checksum = '' - - def AddMachine(self, machine_name): - with self._lock: - for m in self._all_machines: - assert m.name != machine_name, 'Tried to double-add %s' % machine_name - cm = machine_manager.MockCrosMachine(machine_name, self.chromeos_root, - 'average') - assert cm.machine_checksum, ('Could not find checksum for machine %s' % - machine_name) - self._all_machines.append(cm) - - -CHROMEOS_ROOT = '/tmp/chromeos-root' -MACHINE_NAMES = ['lumpy1', 'lumpy2', 'lumpy3', 'daisy1', 'daisy2'] -LABEL_LUMPY = label.MockLabel('lumpy', 'build', 'lumpy_chromeos_image', - 'autotest_dir', 'debug_dir', CHROMEOS_ROOT, - 'lumpy', ['lumpy1', 'lumpy2', 'lumpy3', 'lumpy4'], - '', '', False, 'average', 'gcc', False, None) -LABEL_MIX = label.MockLabel('mix', 'build', 'chromeos_image', 'autotest_dir', - 'debug_dir', CHROMEOS_ROOT, 'mix', - ['daisy1', 'daisy2', 'lumpy3', 'lumpy4'], '', '', - False, 'average', 'gcc', False, None) + """Machine manager for test.""" + + def __init__(self, chromeos_root): + super(MyMachineManager, self).__init__(chromeos_root, 0, "average", "") + + def _TryToLockMachine(self, cros_machine): + self._machines.append(cros_machine) + cros_machine.checksum = "" + + def AddMachine(self, machine_name): + with self._lock: + for m in self._all_machines: + assert m.name != machine_name, ( + "Tried to double-add %s" % machine_name + ) + cm = machine_manager.MockCrosMachine( + machine_name, self.chromeos_root, "average" + ) + assert cm.machine_checksum, ( + "Could not find checksum for machine %s" % machine_name + ) + self._all_machines.append(cm) + + +CHROMEOS_ROOT = "/tmp/chromeos-root" +MACHINE_NAMES = ["lumpy1", "lumpy2", "lumpy3", "daisy1", "daisy2"] +LABEL_LUMPY = label.MockLabel( + "lumpy", + "build", + "lumpy_chromeos_image", + "autotest_dir", + "debug_dir", + CHROMEOS_ROOT, + "lumpy", + ["lumpy1", "lumpy2", "lumpy3", "lumpy4"], + "", + "", + False, + "average", + "gcc", + False, + None, +) +LABEL_MIX = label.MockLabel( + "mix", + "build", + "chromeos_image", + "autotest_dir", + "debug_dir", + CHROMEOS_ROOT, + "mix", + ["daisy1", "daisy2", "lumpy3", "lumpy4"], + "", + "", + False, + "average", + "gcc", + False, + None, +) class MachineManagerTest(unittest.TestCase): - """Test for machine manager class.""" - - msgs = [] - image_log = [] - log_fatal_msgs = [] - fake_logger_count = 0 - fake_logger_msgs = [] - - mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter) - - mock_logger = mock.Mock(spec=logger.Logger) - - mock_lumpy1 = mock.Mock(spec=machine_manager.CrosMachine) - mock_lumpy2 = mock.Mock(spec=machine_manager.CrosMachine) - mock_lumpy3 = mock.Mock(spec=machine_manager.CrosMachine) - mock_lumpy4 = mock.Mock(spec=machine_manager.CrosMachine) - mock_daisy1 = mock.Mock(spec=machine_manager.CrosMachine) - mock_daisy2 = mock.Mock(spec=machine_manager.CrosMachine) - - @mock.patch.object(os.path, 'isdir') - - # pylint: disable=arguments-differ - def setUp(self, mock_isdir): - - mock_isdir.return_value = True - self.mm = machine_manager.MachineManager('/usr/local/chromeos', 0, - 'average', None, - self.mock_cmd_exec, - self.mock_logger) - - self.mock_lumpy1.name = 'lumpy1' - self.mock_lumpy2.name = 'lumpy2' - self.mock_lumpy3.name = 'lumpy3' - self.mock_lumpy4.name = 'lumpy4' - self.mock_daisy1.name = 'daisy1' - self.mock_daisy2.name = 'daisy2' - self.mock_lumpy1.machine_checksum = 'lumpy123' - self.mock_lumpy2.machine_checksum = 'lumpy123' - self.mock_lumpy3.machine_checksum = 'lumpy123' - self.mock_lumpy4.machine_checksum = 'lumpy123' - self.mock_daisy1.machine_checksum = 'daisy12' - self.mock_daisy2.machine_checksum = 'daisy12' - self.mock_lumpy1.checksum_string = 'lumpy_checksum_str' - self.mock_lumpy2.checksum_string = 'lumpy_checksum_str' - self.mock_lumpy3.checksum_string = 'lumpy_checksum_str' - self.mock_lumpy4.checksum_string = 'lumpy_checksum_str' - self.mock_daisy1.checksum_string = 'daisy_checksum_str' - self.mock_daisy2.checksum_string = 'daisy_checksum_str' - self.mock_lumpy1.cpuinfo = 'lumpy_cpu_info' - self.mock_lumpy2.cpuinfo = 'lumpy_cpu_info' - self.mock_lumpy3.cpuinfo = 'lumpy_cpu_info' - self.mock_lumpy4.cpuinfo = 'lumpy_cpu_info' - self.mock_daisy1.cpuinfo = 'daisy_cpu_info' - self.mock_daisy2.cpuinfo = 'daisy_cpu_info' - self.mm._all_machines.append(self.mock_daisy1) - self.mm._all_machines.append(self.mock_daisy2) - self.mm._all_machines.append(self.mock_lumpy1) - self.mm._all_machines.append(self.mock_lumpy2) - self.mm._all_machines.append(self.mock_lumpy3) - - def testGetMachines(self): - manager = MyMachineManager(CHROMEOS_ROOT) - for m in MACHINE_NAMES: - manager.AddMachine(m) - names = [m.name for m in manager.GetMachines(LABEL_LUMPY)] - self.assertEqual(names, ['lumpy1', 'lumpy2', 'lumpy3']) - - def testGetAvailableMachines(self): - manager = MyMachineManager(CHROMEOS_ROOT) - for m in MACHINE_NAMES: - manager.AddMachine(m) - for m in manager._all_machines: - if int(m.name[-1]) % 2: - manager._TryToLockMachine(m) - names = [m.name for m in manager.GetAvailableMachines(LABEL_LUMPY)] - self.assertEqual(names, ['lumpy1', 'lumpy3']) - - @mock.patch.object(time, 'sleep') - @mock.patch.object(command_executer.CommandExecuter, 'RunCommand') - @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') - @mock.patch.object(image_checksummer.ImageChecksummer, 'Checksum') - def test_image_machine(self, mock_checksummer, mock_run_croscmd, mock_run_cmd, - mock_sleep): - - def FakeMD5Checksum(_input_str): - return 'machine_fake_md5_checksum' - - self.fake_logger_count = 0 - self.fake_logger_msgs = [] - - def FakeLogOutput(msg): - self.fake_logger_count += 1 - self.fake_logger_msgs.append(msg) - - def ResetValues(): - self.fake_logger_count = 0 - self.fake_logger_msgs = [] - mock_run_cmd.reset_mock() - mock_run_croscmd.reset_mock() - mock_checksummer.reset_mock() - mock_sleep.reset_mock() - machine.checksum = 'fake_md5_checksum' - self.mm.checksum = None - self.mm.num_reimages = 0 - - self.mock_cmd_exec.CrosRunCommand = mock_run_croscmd - self.mock_cmd_exec.RunCommand = mock_run_cmd - - self.mm.logger.LogOutput = FakeLogOutput - machine = self.mock_lumpy1 - machine._GetMD5Checksum = FakeMD5Checksum - machine.checksum = 'fake_md5_checksum' - mock_checksummer.return_value = 'fake_md5_checksum' - self.mock_cmd_exec.log_level = 'verbose' - - test_flag.SetTestMode(True) - # Test 1: label.image_type == "local" - LABEL_LUMPY.image_type = 'local' - self.mm.ImageMachine(machine, LABEL_LUMPY) - self.assertEqual(mock_run_cmd.call_count, 0) - self.assertEqual(mock_run_croscmd.call_count, 0) - - # Test 2: label.image_type == "trybot" - ResetValues() - LABEL_LUMPY.image_type = 'trybot' - mock_run_cmd.return_value = 0 - self.mm.ImageMachine(machine, LABEL_LUMPY) - self.assertEqual(mock_run_croscmd.call_count, 0) - self.assertEqual(mock_checksummer.call_count, 0) - - # Test 3: label.image_type is neither local nor trybot; retval from - # RunCommand is 1, i.e. image_chromeos fails... - ResetValues() - LABEL_LUMPY.image_type = 'other' - mock_run_cmd.return_value = 1 - try: - self.mm.ImageMachine(machine, LABEL_LUMPY) - except RuntimeError: - self.assertEqual(mock_checksummer.call_count, 0) - self.assertEqual(mock_run_cmd.call_count, 2) - self.assertEqual(mock_run_croscmd.call_count, 1) - self.assertEqual(mock_sleep.call_count, 1) - image_call_args_str = mock_run_cmd.call_args[0][0] - image_call_args = image_call_args_str.split(' ') - self.assertEqual(image_call_args[0], 'python') - self.assertEqual(image_call_args[1].split('/')[-1], 'image_chromeos.pyc') - image_call_args = image_call_args[2:] - self.assertEqual(image_call_args, [ - '--chromeos_root=/tmp/chromeos-root', '--image=lumpy_chromeos_image', - '--image_args=', '--remote=lumpy1', '--logging_level=average', - '--board=lumpy' - ]) - self.assertEqual(mock_run_croscmd.call_args[0][0], 'reboot && exit') - - # Test 4: Everything works properly. Trybot image type. - ResetValues() - LABEL_LUMPY.image_type = 'trybot' - mock_run_cmd.return_value = 0 - self.mm.ImageMachine(machine, LABEL_LUMPY) - self.assertEqual(mock_checksummer.call_count, 0) - self.assertEqual(mock_run_croscmd.call_count, 0) - self.assertEqual(mock_sleep.call_count, 0) - - def test_compute_common_checksum(self): - self.mm.machine_checksum = {} - self.mm.ComputeCommonCheckSum(LABEL_LUMPY) - self.assertEqual(self.mm.machine_checksum['lumpy'], 'lumpy123') - self.assertEqual(len(self.mm.machine_checksum), 1) - - self.mm.machine_checksum = {} - self.assertRaisesRegex(machine_manager.BadChecksum, r'daisy.*\n.*lumpy', - self.mm.ComputeCommonCheckSum, LABEL_MIX) - - def test_compute_common_checksum_string(self): - self.mm.machine_checksum_string = {} - self.mm.ComputeCommonCheckSumString(LABEL_LUMPY) - self.assertEqual(len(self.mm.machine_checksum_string), 1) - self.assertEqual(self.mm.machine_checksum_string['lumpy'], - 'lumpy_checksum_str') - - self.mm.machine_checksum_string = {} - self.mm.ComputeCommonCheckSumString(LABEL_MIX) - self.assertEqual(len(self.mm.machine_checksum_string), 1) - self.assertEqual(self.mm.machine_checksum_string['mix'], - 'daisy_checksum_str') - - @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommandWOutput') - def test_try_to_lock_machine(self, mock_cros_runcmd): - mock_cros_runcmd.return_value = [0, 'false_lock_checksum', ''] - self.mock_cmd_exec.CrosRunCommandWOutput = mock_cros_runcmd - self.mm._machines = [] - self.mm._TryToLockMachine(self.mock_lumpy1) - self.assertEqual(len(self.mm._machines), 1) - self.assertEqual(self.mm._machines[0], self.mock_lumpy1) - self.assertEqual(self.mock_lumpy1.checksum, 'false_lock_checksum') - self.assertEqual(mock_cros_runcmd.call_count, 1) - cmd_str = mock_cros_runcmd.call_args[0][0] - self.assertEqual(cmd_str, 'cat /usr/local/osimage_checksum_file') - args_dict = mock_cros_runcmd.call_args[1] - self.assertEqual(len(args_dict), 2) - self.assertEqual(args_dict['machine'], self.mock_lumpy1.name) - self.assertEqual(args_dict['chromeos_root'], '/usr/local/chromeos') - - @mock.patch.object(machine_manager, 'CrosMachine') - def test_add_machine(self, mock_machine): - - mock_machine.machine_checksum = 'daisy123' - self.assertEqual(len(self.mm._all_machines), 5) - self.mm.AddMachine('daisy3') - self.assertEqual(len(self.mm._all_machines), 6) - - self.assertRaises(Exception, self.mm.AddMachine, 'lumpy1') - - def test_remove_machine(self): - self.mm._machines = self.mm._all_machines - self.assertTrue(self.mock_lumpy2 in self.mm._machines) - self.mm.RemoveMachine(self.mock_lumpy2.name) - self.assertFalse(self.mock_lumpy2 in self.mm._machines) - - def test_force_same_image_to_all_machines(self): - self.image_log = [] - - def FakeImageMachine(machine, label_arg): - image = label_arg.chromeos_image - self.image_log.append('Pushed %s onto %s' % (image, machine.name)) - - def FakeSetUpChecksumInfo(): - pass - - self.mm.ImageMachine = FakeImageMachine - self.mock_lumpy1.SetUpChecksumInfo = FakeSetUpChecksumInfo - self.mock_lumpy2.SetUpChecksumInfo = FakeSetUpChecksumInfo - self.mock_lumpy3.SetUpChecksumInfo = FakeSetUpChecksumInfo - - self.mm.ForceSameImageToAllMachines(LABEL_LUMPY) - self.assertEqual(len(self.image_log), 3) - self.assertEqual(self.image_log[0], - 'Pushed lumpy_chromeos_image onto lumpy1') - self.assertEqual(self.image_log[1], - 'Pushed lumpy_chromeos_image onto lumpy2') - self.assertEqual(self.image_log[2], - 'Pushed lumpy_chromeos_image onto lumpy3') - - @mock.patch.object(image_checksummer.ImageChecksummer, 'Checksum') - @mock.patch.object(hashlib, 'md5') - def test_acquire_machine(self, mock_md5, mock_checksum): - - self.msgs = [] - self.log_fatal_msgs = [] - - def FakeLock(machine): - self.msgs.append('Tried to lock %s' % machine.name) - - def FakeLogFatal(msg): - self.log_fatal_msgs.append(msg) - - self.mm._TryToLockMachine = FakeLock - self.mm.logger.LogFatal = FakeLogFatal - - mock_md5.return_value = '123456' - mock_checksum.return_value = 'fake_md5_checksum' - - self.mm._machines = self.mm._all_machines - self.mock_lumpy1.locked = True - self.mock_lumpy2.locked = True - self.mock_lumpy3.locked = False - self.mock_lumpy3.checksum = 'fake_md5_checksum' - self.mock_daisy1.locked = True - self.mock_daisy2.locked = False - self.mock_daisy2.checksum = 'fake_md5_checksum' - - self.mock_lumpy1.released_time = time.time() - self.mock_lumpy2.released_time = time.time() - self.mock_lumpy3.released_time = time.time() - self.mock_daisy1.released_time = time.time() - self.mock_daisy2.released_time = time.time() - - # Test 1. Basic test. Acquire lumpy3. - self.mm.AcquireMachine(LABEL_LUMPY) - m = self.mock_lumpy1 - self.assertEqual(m, self.mock_lumpy1) - self.assertTrue(self.mock_lumpy1.locked) - self.assertEqual(mock_md5.call_count, 0) - self.assertEqual(self.msgs, [ - 'Tried to lock lumpy1', 'Tried to lock lumpy2', 'Tried to lock lumpy3' - ]) - - # Test the second return statment (machine is unlocked, has no checksum) - save_locked = self.mock_lumpy1.locked - self.mock_lumpy1.locked = False - self.mock_lumpy1.checksum = None - m = self.mm.AcquireMachine(LABEL_LUMPY) - self.assertEqual(m, self.mock_lumpy1) - self.assertTrue(self.mock_lumpy1.locked) - - # Test the third return statement: - # - machine is unlocked - # - checksums don't match - # - current time minus release time is > 20. - self.mock_lumpy1.locked = False - self.mock_lumpy1.checksum = '123' - self.mock_lumpy1.released_time = time.time() - 8 - m = self.mm.AcquireMachine(LABEL_LUMPY) - self.assertEqual(m, self.mock_lumpy1) - self.assertTrue(self.mock_lumpy1.locked) - - # Test all machines are already locked. - m = self.mm.AcquireMachine(LABEL_LUMPY) - self.assertIsNone(m) - - # Restore values of mock_lumpy1, so other tests succeed. - self.mock_lumpy1.locked = save_locked - self.mock_lumpy1.checksum = '123' - - def test_get_available_machines(self): - self.mm._machines = self.mm._all_machines - - machine_list = self.mm.GetAvailableMachines() - self.assertEqual(machine_list, self.mm._all_machines) - - machine_list = self.mm.GetAvailableMachines(LABEL_MIX) - self.assertEqual(machine_list, - [self.mock_daisy1, self.mock_daisy2, self.mock_lumpy3]) - - machine_list = self.mm.GetAvailableMachines(LABEL_LUMPY) - self.assertEqual(machine_list, - [self.mock_lumpy1, self.mock_lumpy2, self.mock_lumpy3]) - - def test_get_machines(self): - machine_list = self.mm.GetMachines() - self.assertEqual(machine_list, self.mm._all_machines) - - machine_list = self.mm.GetMachines(LABEL_MIX) - self.assertEqual(machine_list, - [self.mock_daisy1, self.mock_daisy2, self.mock_lumpy3]) - - machine_list = self.mm.GetMachines(LABEL_LUMPY) - self.assertEqual(machine_list, - [self.mock_lumpy1, self.mock_lumpy2, self.mock_lumpy3]) - - def test_release_machines(self): - - self.mm._machines = [self.mock_lumpy1, self.mock_daisy2] - - self.mock_lumpy1.locked = True - self.mock_daisy2.locked = True - - self.assertTrue(self.mock_lumpy1.locked) - self.mm.ReleaseMachine(self.mock_lumpy1) - self.assertFalse(self.mock_lumpy1.locked) - self.assertEqual(self.mock_lumpy1.status, 'Available') - - self.assertTrue(self.mock_daisy2.locked) - self.mm.ReleaseMachine(self.mock_daisy2) - self.assertFalse(self.mock_daisy2.locked) - self.assertEqual(self.mock_daisy2.status, 'Available') - - # Test double-relase... - self.assertRaises(AssertionError, self.mm.ReleaseMachine, self.mock_lumpy1) - - def test_cleanup(self): - self.mock_logger.reset_mock() - self.mm.Cleanup() - self.assertEqual(self.mock_logger.call_count, 0) - - OUTPUT_STR = ('Machine Status:\nMachine Thread ' - 'Lock Status Checksum' - ' \nlumpy1 test ' - 'run True PENDING 123' - ' \nlumpy2 ' - 'test run False PENDING 123' - ' \nlumpy3 ' - 'test run False PENDING 123' - ' \ndaisy1 ' - 'test run False PENDING 678' - ' \ndaisy2 ' - 'test run True PENDING 678' - ' ') - - def test_as_string(self): + """Test for machine manager class.""" + + msgs = [] + image_log = [] + log_fatal_msgs = [] + fake_logger_count = 0 + fake_logger_msgs = [] + + mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter) mock_logger = mock.Mock(spec=logger.Logger) - bench = Benchmark( - 'page_cycler_v2.netsim.top_10', # name - 'page_cycler_v2.netsim.top_10', # test_name - '', # test_args - 1, # iteratins - False, # rm_chroot_tmp - '', # perf_args - suite='telemetry_Crosperf') # suite - - test_run = MockBenchmarkRun('test run', bench, LABEL_LUMPY, 1, [], self.mm, - mock_logger, 'verbose', '', {}) - - self.mm._machines = [ - self.mock_lumpy1, self.mock_lumpy2, self.mock_lumpy3, self.mock_daisy1, - self.mock_daisy2 - ] - - self.mock_lumpy1.test_run = test_run - self.mock_lumpy2.test_run = test_run - self.mock_lumpy3.test_run = test_run - self.mock_daisy1.test_run = test_run - self.mock_daisy2.test_run = test_run - - self.mock_lumpy1.locked = True - self.mock_lumpy2.locked = False - self.mock_lumpy3.locked = False - self.mock_daisy1.locked = False - self.mock_daisy2.locked = True - - self.mock_lumpy1.checksum = '123' - self.mock_lumpy2.checksum = '123' - self.mock_lumpy3.checksum = '123' - self.mock_daisy1.checksum = '678' - self.mock_daisy2.checksum = '678' - - output = self.mm.AsString() - self.assertEqual(output, self.OUTPUT_STR) - - def test_get_all_cpu_info(self): - info = self.mm.GetAllCPUInfo([LABEL_LUMPY, LABEL_MIX]) - self.assertEqual( - info, 'lumpy\n-------------------\nlumpy_cpu_info\n\n\nmix\n-' - '------------------\ndaisy_cpu_info\n\n\n') + mock_lumpy1 = mock.Mock(spec=machine_manager.CrosMachine) + mock_lumpy2 = mock.Mock(spec=machine_manager.CrosMachine) + mock_lumpy3 = mock.Mock(spec=machine_manager.CrosMachine) + mock_lumpy4 = mock.Mock(spec=machine_manager.CrosMachine) + mock_daisy1 = mock.Mock(spec=machine_manager.CrosMachine) + mock_daisy2 = mock.Mock(spec=machine_manager.CrosMachine) + + @mock.patch.object(os.path, "isdir") + + # pylint: disable=arguments-differ + def setUp(self, mock_isdir): + + mock_isdir.return_value = True + self.mm = machine_manager.MachineManager( + "/usr/local/chromeos", + 0, + "average", + None, + self.mock_cmd_exec, + self.mock_logger, + ) + + self.mock_lumpy1.name = "lumpy1" + self.mock_lumpy2.name = "lumpy2" + self.mock_lumpy3.name = "lumpy3" + self.mock_lumpy4.name = "lumpy4" + self.mock_daisy1.name = "daisy1" + self.mock_daisy2.name = "daisy2" + self.mock_lumpy1.machine_checksum = "lumpy123" + self.mock_lumpy2.machine_checksum = "lumpy123" + self.mock_lumpy3.machine_checksum = "lumpy123" + self.mock_lumpy4.machine_checksum = "lumpy123" + self.mock_daisy1.machine_checksum = "daisy12" + self.mock_daisy2.machine_checksum = "daisy12" + self.mock_lumpy1.checksum_string = "lumpy_checksum_str" + self.mock_lumpy2.checksum_string = "lumpy_checksum_str" + self.mock_lumpy3.checksum_string = "lumpy_checksum_str" + self.mock_lumpy4.checksum_string = "lumpy_checksum_str" + self.mock_daisy1.checksum_string = "daisy_checksum_str" + self.mock_daisy2.checksum_string = "daisy_checksum_str" + self.mock_lumpy1.cpuinfo = "lumpy_cpu_info" + self.mock_lumpy2.cpuinfo = "lumpy_cpu_info" + self.mock_lumpy3.cpuinfo = "lumpy_cpu_info" + self.mock_lumpy4.cpuinfo = "lumpy_cpu_info" + self.mock_daisy1.cpuinfo = "daisy_cpu_info" + self.mock_daisy2.cpuinfo = "daisy_cpu_info" + self.mm._all_machines.append(self.mock_daisy1) + self.mm._all_machines.append(self.mock_daisy2) + self.mm._all_machines.append(self.mock_lumpy1) + self.mm._all_machines.append(self.mock_lumpy2) + self.mm._all_machines.append(self.mock_lumpy3) + + def testGetMachines(self): + manager = MyMachineManager(CHROMEOS_ROOT) + for m in MACHINE_NAMES: + manager.AddMachine(m) + names = [m.name for m in manager.GetMachines(LABEL_LUMPY)] + self.assertEqual(names, ["lumpy1", "lumpy2", "lumpy3"]) + + def testGetAvailableMachines(self): + manager = MyMachineManager(CHROMEOS_ROOT) + for m in MACHINE_NAMES: + manager.AddMachine(m) + for m in manager._all_machines: + if int(m.name[-1]) % 2: + manager._TryToLockMachine(m) + names = [m.name for m in manager.GetAvailableMachines(LABEL_LUMPY)] + self.assertEqual(names, ["lumpy1", "lumpy3"]) + + @mock.patch.object(time, "sleep") + @mock.patch.object(command_executer.CommandExecuter, "RunCommand") + @mock.patch.object(command_executer.CommandExecuter, "CrosRunCommand") + @mock.patch.object(image_checksummer.ImageChecksummer, "Checksum") + def test_image_machine( + self, mock_checksummer, mock_run_croscmd, mock_run_cmd, mock_sleep + ): + def FakeMD5Checksum(_input_str): + return "machine_fake_md5_checksum" + + self.fake_logger_count = 0 + self.fake_logger_msgs = [] + + def FakeLogOutput(msg): + self.fake_logger_count += 1 + self.fake_logger_msgs.append(msg) + + def ResetValues(): + self.fake_logger_count = 0 + self.fake_logger_msgs = [] + mock_run_cmd.reset_mock() + mock_run_croscmd.reset_mock() + mock_checksummer.reset_mock() + mock_sleep.reset_mock() + machine.checksum = "fake_md5_checksum" + self.mm.checksum = None + self.mm.num_reimages = 0 + + self.mock_cmd_exec.CrosRunCommand = mock_run_croscmd + self.mock_cmd_exec.RunCommand = mock_run_cmd + + self.mm.logger.LogOutput = FakeLogOutput + machine = self.mock_lumpy1 + machine._GetMD5Checksum = FakeMD5Checksum + machine.checksum = "fake_md5_checksum" + mock_checksummer.return_value = "fake_md5_checksum" + self.mock_cmd_exec.log_level = "verbose" + + test_flag.SetTestMode(True) + # Test 1: label.image_type == "local" + LABEL_LUMPY.image_type = "local" + self.mm.ImageMachine(machine, LABEL_LUMPY) + self.assertEqual(mock_run_cmd.call_count, 0) + self.assertEqual(mock_run_croscmd.call_count, 0) + + # Test 2: label.image_type == "trybot" + ResetValues() + LABEL_LUMPY.image_type = "trybot" + mock_run_cmd.return_value = 0 + self.mm.ImageMachine(machine, LABEL_LUMPY) + self.assertEqual(mock_run_croscmd.call_count, 0) + self.assertEqual(mock_checksummer.call_count, 0) + + # Test 3: label.image_type is neither local nor trybot; retval from + # RunCommand is 1, i.e. image_chromeos fails... + ResetValues() + LABEL_LUMPY.image_type = "other" + mock_run_cmd.return_value = 1 + try: + self.mm.ImageMachine(machine, LABEL_LUMPY) + except RuntimeError: + self.assertEqual(mock_checksummer.call_count, 0) + self.assertEqual(mock_run_cmd.call_count, 2) + self.assertEqual(mock_run_croscmd.call_count, 1) + self.assertEqual(mock_sleep.call_count, 1) + image_call_args_str = mock_run_cmd.call_args[0][0] + image_call_args = image_call_args_str.split(" ") + self.assertEqual(image_call_args[0], "python") + self.assertEqual( + image_call_args[1].split("/")[-1], "image_chromeos.pyc" + ) + image_call_args = image_call_args[2:] + self.assertEqual( + image_call_args, + [ + "--chromeos_root=/tmp/chromeos-root", + "--image=lumpy_chromeos_image", + "--image_args=", + "--remote=lumpy1", + "--logging_level=average", + "--board=lumpy", + ], + ) + self.assertEqual(mock_run_croscmd.call_args[0][0], "reboot && exit") + + # Test 4: Everything works properly. Trybot image type. + ResetValues() + LABEL_LUMPY.image_type = "trybot" + mock_run_cmd.return_value = 0 + self.mm.ImageMachine(machine, LABEL_LUMPY) + self.assertEqual(mock_checksummer.call_count, 0) + self.assertEqual(mock_run_croscmd.call_count, 0) + self.assertEqual(mock_sleep.call_count, 0) + + def test_compute_common_checksum(self): + self.mm.machine_checksum = {} + self.mm.ComputeCommonCheckSum(LABEL_LUMPY) + self.assertEqual(self.mm.machine_checksum["lumpy"], "lumpy123") + self.assertEqual(len(self.mm.machine_checksum), 1) + + self.mm.machine_checksum = {} + self.assertRaisesRegex( + machine_manager.BadChecksum, + r"daisy.*\n.*lumpy", + self.mm.ComputeCommonCheckSum, + LABEL_MIX, + ) + + def test_compute_common_checksum_string(self): + self.mm.machine_checksum_string = {} + self.mm.ComputeCommonCheckSumString(LABEL_LUMPY) + self.assertEqual(len(self.mm.machine_checksum_string), 1) + self.assertEqual( + self.mm.machine_checksum_string["lumpy"], "lumpy_checksum_str" + ) + + self.mm.machine_checksum_string = {} + self.mm.ComputeCommonCheckSumString(LABEL_MIX) + self.assertEqual(len(self.mm.machine_checksum_string), 1) + self.assertEqual( + self.mm.machine_checksum_string["mix"], "daisy_checksum_str" + ) + + @mock.patch.object( + command_executer.CommandExecuter, "CrosRunCommandWOutput" + ) + def test_try_to_lock_machine(self, mock_cros_runcmd): + mock_cros_runcmd.return_value = [0, "false_lock_checksum", ""] + self.mock_cmd_exec.CrosRunCommandWOutput = mock_cros_runcmd + self.mm._machines = [] + self.mm._TryToLockMachine(self.mock_lumpy1) + self.assertEqual(len(self.mm._machines), 1) + self.assertEqual(self.mm._machines[0], self.mock_lumpy1) + self.assertEqual(self.mock_lumpy1.checksum, "false_lock_checksum") + self.assertEqual(mock_cros_runcmd.call_count, 1) + cmd_str = mock_cros_runcmd.call_args[0][0] + self.assertEqual(cmd_str, "cat /usr/local/osimage_checksum_file") + args_dict = mock_cros_runcmd.call_args[1] + self.assertEqual(len(args_dict), 2) + self.assertEqual(args_dict["machine"], self.mock_lumpy1.name) + self.assertEqual(args_dict["chromeos_root"], "/usr/local/chromeos") + + @mock.patch.object(machine_manager, "CrosMachine") + def test_add_machine(self, mock_machine): + + mock_machine.machine_checksum = "daisy123" + self.assertEqual(len(self.mm._all_machines), 5) + self.mm.AddMachine("daisy3") + self.assertEqual(len(self.mm._all_machines), 6) + + self.assertRaises(Exception, self.mm.AddMachine, "lumpy1") + + def test_remove_machine(self): + self.mm._machines = self.mm._all_machines + self.assertTrue(self.mock_lumpy2 in self.mm._machines) + self.mm.RemoveMachine(self.mock_lumpy2.name) + self.assertFalse(self.mock_lumpy2 in self.mm._machines) + + def test_force_same_image_to_all_machines(self): + self.image_log = [] + + def FakeImageMachine(machine, label_arg): + image = label_arg.chromeos_image + self.image_log.append("Pushed %s onto %s" % (image, machine.name)) + + def FakeSetUpChecksumInfo(): + pass + + self.mm.ImageMachine = FakeImageMachine + self.mock_lumpy1.SetUpChecksumInfo = FakeSetUpChecksumInfo + self.mock_lumpy2.SetUpChecksumInfo = FakeSetUpChecksumInfo + self.mock_lumpy3.SetUpChecksumInfo = FakeSetUpChecksumInfo + + self.mm.ForceSameImageToAllMachines(LABEL_LUMPY) + self.assertEqual(len(self.image_log), 3) + self.assertEqual( + self.image_log[0], "Pushed lumpy_chromeos_image onto lumpy1" + ) + self.assertEqual( + self.image_log[1], "Pushed lumpy_chromeos_image onto lumpy2" + ) + self.assertEqual( + self.image_log[2], "Pushed lumpy_chromeos_image onto lumpy3" + ) + + @mock.patch.object(image_checksummer.ImageChecksummer, "Checksum") + @mock.patch.object(hashlib, "md5") + def test_acquire_machine(self, mock_md5, mock_checksum): + + self.msgs = [] + self.log_fatal_msgs = [] + + def FakeLock(machine): + self.msgs.append("Tried to lock %s" % machine.name) + + def FakeLogFatal(msg): + self.log_fatal_msgs.append(msg) + + self.mm._TryToLockMachine = FakeLock + self.mm.logger.LogFatal = FakeLogFatal + + mock_md5.return_value = "123456" + mock_checksum.return_value = "fake_md5_checksum" + + self.mm._machines = self.mm._all_machines + self.mock_lumpy1.locked = True + self.mock_lumpy2.locked = True + self.mock_lumpy3.locked = False + self.mock_lumpy3.checksum = "fake_md5_checksum" + self.mock_daisy1.locked = True + self.mock_daisy2.locked = False + self.mock_daisy2.checksum = "fake_md5_checksum" + + self.mock_lumpy1.released_time = time.time() + self.mock_lumpy2.released_time = time.time() + self.mock_lumpy3.released_time = time.time() + self.mock_daisy1.released_time = time.time() + self.mock_daisy2.released_time = time.time() + + # Test 1. Basic test. Acquire lumpy3. + self.mm.AcquireMachine(LABEL_LUMPY) + m = self.mock_lumpy1 + self.assertEqual(m, self.mock_lumpy1) + self.assertTrue(self.mock_lumpy1.locked) + self.assertEqual(mock_md5.call_count, 0) + self.assertEqual( + self.msgs, + [ + "Tried to lock lumpy1", + "Tried to lock lumpy2", + "Tried to lock lumpy3", + ], + ) + + # Test the second return statment (machine is unlocked, has no checksum) + save_locked = self.mock_lumpy1.locked + self.mock_lumpy1.locked = False + self.mock_lumpy1.checksum = None + m = self.mm.AcquireMachine(LABEL_LUMPY) + self.assertEqual(m, self.mock_lumpy1) + self.assertTrue(self.mock_lumpy1.locked) + + # Test the third return statement: + # - machine is unlocked + # - checksums don't match + # - current time minus release time is > 20. + self.mock_lumpy1.locked = False + self.mock_lumpy1.checksum = "123" + self.mock_lumpy1.released_time = time.time() - 8 + m = self.mm.AcquireMachine(LABEL_LUMPY) + self.assertEqual(m, self.mock_lumpy1) + self.assertTrue(self.mock_lumpy1.locked) + + # Test all machines are already locked. + m = self.mm.AcquireMachine(LABEL_LUMPY) + self.assertIsNone(m) + + # Restore values of mock_lumpy1, so other tests succeed. + self.mock_lumpy1.locked = save_locked + self.mock_lumpy1.checksum = "123" + + def test_get_available_machines(self): + self.mm._machines = self.mm._all_machines + + machine_list = self.mm.GetAvailableMachines() + self.assertEqual(machine_list, self.mm._all_machines) + + machine_list = self.mm.GetAvailableMachines(LABEL_MIX) + self.assertEqual( + machine_list, [self.mock_daisy1, self.mock_daisy2, self.mock_lumpy3] + ) + + machine_list = self.mm.GetAvailableMachines(LABEL_LUMPY) + self.assertEqual( + machine_list, [self.mock_lumpy1, self.mock_lumpy2, self.mock_lumpy3] + ) + + def test_get_machines(self): + machine_list = self.mm.GetMachines() + self.assertEqual(machine_list, self.mm._all_machines) + + machine_list = self.mm.GetMachines(LABEL_MIX) + self.assertEqual( + machine_list, [self.mock_daisy1, self.mock_daisy2, self.mock_lumpy3] + ) + + machine_list = self.mm.GetMachines(LABEL_LUMPY) + self.assertEqual( + machine_list, [self.mock_lumpy1, self.mock_lumpy2, self.mock_lumpy3] + ) + + def test_release_machines(self): + + self.mm._machines = [self.mock_lumpy1, self.mock_daisy2] + + self.mock_lumpy1.locked = True + self.mock_daisy2.locked = True + + self.assertTrue(self.mock_lumpy1.locked) + self.mm.ReleaseMachine(self.mock_lumpy1) + self.assertFalse(self.mock_lumpy1.locked) + self.assertEqual(self.mock_lumpy1.status, "Available") + + self.assertTrue(self.mock_daisy2.locked) + self.mm.ReleaseMachine(self.mock_daisy2) + self.assertFalse(self.mock_daisy2.locked) + self.assertEqual(self.mock_daisy2.status, "Available") + + # Test double-relase... + self.assertRaises( + AssertionError, self.mm.ReleaseMachine, self.mock_lumpy1 + ) + + def test_cleanup(self): + self.mock_logger.reset_mock() + self.mm.Cleanup() + self.assertEqual(self.mock_logger.call_count, 0) + + OUTPUT_STR = ( + "Machine Status:\nMachine Thread " + "Lock Status Checksum" + " \nlumpy1 test " + "run True PENDING 123" + " \nlumpy2 " + "test run False PENDING 123" + " \nlumpy3 " + "test run False PENDING 123" + " \ndaisy1 " + "test run False PENDING 678" + " \ndaisy2 " + "test run True PENDING 678" + " " + ) + + def test_as_string(self): + + mock_logger = mock.Mock(spec=logger.Logger) + + bench = Benchmark( + "page_cycler_v2.netsim.top_10", # name + "page_cycler_v2.netsim.top_10", # test_name + "", # test_args + 1, # iteratins + False, # rm_chroot_tmp + "", # perf_args + suite="telemetry_Crosperf", + ) # suite + + test_run = MockBenchmarkRun( + "test run", + bench, + LABEL_LUMPY, + 1, + [], + self.mm, + mock_logger, + "verbose", + "", + {}, + ) + + self.mm._machines = [ + self.mock_lumpy1, + self.mock_lumpy2, + self.mock_lumpy3, + self.mock_daisy1, + self.mock_daisy2, + ] + + self.mock_lumpy1.test_run = test_run + self.mock_lumpy2.test_run = test_run + self.mock_lumpy3.test_run = test_run + self.mock_daisy1.test_run = test_run + self.mock_daisy2.test_run = test_run + + self.mock_lumpy1.locked = True + self.mock_lumpy2.locked = False + self.mock_lumpy3.locked = False + self.mock_daisy1.locked = False + self.mock_daisy2.locked = True + + self.mock_lumpy1.checksum = "123" + self.mock_lumpy2.checksum = "123" + self.mock_lumpy3.checksum = "123" + self.mock_daisy1.checksum = "678" + self.mock_daisy2.checksum = "678" + + output = self.mm.AsString() + self.assertEqual(output, self.OUTPUT_STR) + + def test_get_all_cpu_info(self): + info = self.mm.GetAllCPUInfo([LABEL_LUMPY, LABEL_MIX]) + self.assertEqual( + info, + "lumpy\n-------------------\nlumpy_cpu_info\n\n\nmix\n-" + "------------------\ndaisy_cpu_info\n\n\n", + ) MEMINFO_STRING = """MemTotal: 3990332 kB @@ -580,35 +661,37 @@ address sizes: 36 bits physical, 48 bits virtual power management: """ -CHECKSUM_STRING = ('processor: 0vendor_id: GenuineIntelcpu family: 6model: ' - '42model name: Intel(R) Celeron(R) CPU 867 @ ' - '1.30GHzstepping: 7microcode: 0x25cache size: 2048 ' - 'KBphysical id: 0siblings: 2cpu cores: 2' - 'fpu: yesfpu_exception: yescpuid level: ' - '13wp: yesflags: fpu vme de pse tsc msr pae mce cx8 apic sep' - ' mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse ' - 'sse2 ss ht tm pbe syscall nx rdtscp lm constant_tsc ' - 'arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc ' - 'aperfmperf pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ' - 'ssse3 cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic popcnt ' - 'tsc_deadline_timer xsave lahf_lm arat epb xsaveopt pln pts ' - 'dts tpr_shadow vnmi flexpriority ept vpidclflush size: ' - '64cache_alignment: 64address sizes: 36 bits physical, 48 ' - 'bits virtualpower management:processor: 1vendor_id: ' - 'GenuineIntelcpu family: 6model: 42model name: Intel(R) ' - 'Celeron(R) CPU 867 @ 1.30GHzstepping: 7microcode: 0x25cache' - ' size: 2048 KBphysical id: 0siblings: 2cpu cores:' - ' 2fpu: yesfpu_exception: yescpuid' - ' level: 13wp: yesflags: fpu vme de pse tsc msr pae mce cx8 ' - 'apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx ' - 'fxsr sse sse2 ss ht tm pbe syscall nx rdtscp lm ' - 'constant_tsc arch_perfmon pebs bts rep_good nopl xtopology ' - 'nonstop_tsc aperfmperf pni pclmulqdq dtes64 monitor ds_cpl ' - 'vmx est tm2 ssse3 cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic ' - 'popcnt tsc_deadline_timer xsave lahf_lm arat epb xsaveopt ' - 'pln pts dts tpr_shadow vnmi flexpriority ept vpidclflush ' - 'size: 64cache_alignment: 64address sizes: 36 bits physical,' - ' 48 bits virtualpower management: 4194304') +CHECKSUM_STRING = ( + "processor: 0vendor_id: GenuineIntelcpu family: 6model: " + "42model name: Intel(R) Celeron(R) CPU 867 @ " + "1.30GHzstepping: 7microcode: 0x25cache size: 2048 " + "KBphysical id: 0siblings: 2cpu cores: 2" + "fpu: yesfpu_exception: yescpuid level: " + "13wp: yesflags: fpu vme de pse tsc msr pae mce cx8 apic sep" + " mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse " + "sse2 ss ht tm pbe syscall nx rdtscp lm constant_tsc " + "arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc " + "aperfmperf pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 " + "ssse3 cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic popcnt " + "tsc_deadline_timer xsave lahf_lm arat epb xsaveopt pln pts " + "dts tpr_shadow vnmi flexpriority ept vpidclflush size: " + "64cache_alignment: 64address sizes: 36 bits physical, 48 " + "bits virtualpower management:processor: 1vendor_id: " + "GenuineIntelcpu family: 6model: 42model name: Intel(R) " + "Celeron(R) CPU 867 @ 1.30GHzstepping: 7microcode: 0x25cache" + " size: 2048 KBphysical id: 0siblings: 2cpu cores:" + " 2fpu: yesfpu_exception: yescpuid" + " level: 13wp: yesflags: fpu vme de pse tsc msr pae mce cx8 " + "apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx " + "fxsr sse sse2 ss ht tm pbe syscall nx rdtscp lm " + "constant_tsc arch_perfmon pebs bts rep_good nopl xtopology " + "nonstop_tsc aperfmperf pni pclmulqdq dtes64 monitor ds_cpl " + "vmx est tm2 ssse3 cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic " + "popcnt tsc_deadline_timer xsave lahf_lm arat epb xsaveopt " + "pln pts dts tpr_shadow vnmi flexpriority ept vpidclflush " + "size: 64cache_alignment: 64address sizes: 36 bits physical," + " 48 bits virtualpower management: 4194304" +) DUMP_VPD_STRING = """ "PBA_SN"="Pba.txt" @@ -667,187 +750,212 @@ wlan0: flags=4099<UP,BROADCAST,MULTICAST> mtu 1500 class CrosMachineTest(unittest.TestCase): - """Test for CrosMachine class.""" - - mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter) - - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_init(self, mock_setup): - - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - self.assertEqual(mock_setup.call_count, 1) - self.assertEqual(cm.chromeos_root, '/usr/local/chromeos') - self.assertEqual(cm.log_level, 'average') - - @mock.patch.object(machine_manager.CrosMachine, 'IsReachable') - @mock.patch.object(machine_manager.CrosMachine, '_GetMemoryInfo') - @mock.patch.object(machine_manager.CrosMachine, '_GetCPUInfo') - @mock.patch.object(machine_manager.CrosMachine, - '_ComputeMachineChecksumString') - @mock.patch.object(machine_manager.CrosMachine, '_GetMachineID') - @mock.patch.object(machine_manager.CrosMachine, '_GetMD5Checksum') - def test_setup_checksum_info(self, mock_md5sum, mock_machineid, - mock_checkstring, mock_cpuinfo, mock_meminfo, - mock_isreachable): - - # Test 1. Machine is not reachable; SetUpChecksumInfo is called via - # __init__. - mock_isreachable.return_value = False - mock_md5sum.return_value = 'md5_checksum' - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - cm.checksum_string = 'This is a checksum string.' - cm.machine_id = 'machine_id1' - self.assertEqual(mock_isreachable.call_count, 1) - self.assertIsNone(cm.machine_checksum) - self.assertEqual(mock_meminfo.call_count, 0) - - # Test 2. Machine is reachable. Call explicitly. - mock_isreachable.return_value = True - cm.checksum_string = 'This is a checksum string.' - cm.machine_id = 'machine_id1' - cm.SetUpChecksumInfo() - self.assertEqual(mock_isreachable.call_count, 2) - self.assertEqual(mock_meminfo.call_count, 1) - self.assertEqual(mock_cpuinfo.call_count, 1) - self.assertEqual(mock_checkstring.call_count, 1) - self.assertEqual(mock_machineid.call_count, 1) - self.assertEqual(mock_md5sum.call_count, 2) - self.assertEqual(cm.machine_checksum, 'md5_checksum') - self.assertEqual(cm.machine_id_checksum, 'md5_checksum') - self.assertEqual(mock_md5sum.call_args_list[0][0][0], - 'This is a checksum string.') - self.assertEqual(mock_md5sum.call_args_list[1][0][0], 'machine_id1') - - @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommand') - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_is_reachable(self, mock_setup, mock_run_cmd): - - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - self.mock_cmd_exec.CrosRunCommand = mock_run_cmd - - # Test 1. CrosRunCommand returns 1 (fail) - mock_run_cmd.return_value = 1 - result = cm.IsReachable() - self.assertFalse(result) - self.assertEqual(mock_setup.call_count, 1) - self.assertEqual(mock_run_cmd.call_count, 1) - - # Test 2. CrosRunCommand returns 0 (success) - mock_run_cmd.return_value = 0 - result = cm.IsReachable() - self.assertTrue(result) - self.assertEqual(mock_run_cmd.call_count, 2) - first_args = mock_run_cmd.call_args_list[0] - second_args = mock_run_cmd.call_args_list[1] - self.assertEqual(first_args[0], second_args[0]) - self.assertEqual(first_args[1], second_args[1]) - self.assertEqual(len(first_args[0]), 1) - self.assertEqual(len(first_args[1]), 2) - self.assertEqual(first_args[0][0], 'ls') - args_dict = first_args[1] - self.assertEqual(args_dict['machine'], 'daisy.cros') - self.assertEqual(args_dict['chromeos_root'], '/usr/local/chromeos') - - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_parse_memory_info(self, _mock_setup): - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - cm.meminfo = MEMINFO_STRING - cm._ParseMemoryInfo() - self.assertEqual(cm.phys_kbytes, 4194304) - - @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommandWOutput') - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_get_memory_info(self, _mock_setup, mock_run_cmd): - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - self.mock_cmd_exec.CrosRunCommandWOutput = mock_run_cmd - mock_run_cmd.return_value = [0, MEMINFO_STRING, ''] - cm._GetMemoryInfo() - self.assertEqual(mock_run_cmd.call_count, 1) - call_args = mock_run_cmd.call_args_list[0] - self.assertEqual(call_args[0][0], 'cat /proc/meminfo') - args_dict = call_args[1] - self.assertEqual(args_dict['machine'], 'daisy.cros') - self.assertEqual(args_dict['chromeos_root'], '/usr/local/chromeos') - self.assertEqual(cm.meminfo, MEMINFO_STRING) - self.assertEqual(cm.phys_kbytes, 4194304) - - mock_run_cmd.return_value = [1, MEMINFO_STRING, ''] - self.assertRaises(Exception, cm._GetMemoryInfo) - - @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommandWOutput') - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_get_cpu_info(self, _mock_setup, mock_run_cmd): - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - self.mock_cmd_exec.CrosRunCommandWOutput = mock_run_cmd - mock_run_cmd.return_value = [0, CPUINFO_STRING, ''] - cm._GetCPUInfo() - self.assertEqual(mock_run_cmd.call_count, 1) - call_args = mock_run_cmd.call_args_list[0] - self.assertEqual(call_args[0][0], 'cat /proc/cpuinfo') - args_dict = call_args[1] - self.assertEqual(args_dict['machine'], 'daisy.cros') - self.assertEqual(args_dict['chromeos_root'], '/usr/local/chromeos') - self.assertEqual(cm.cpuinfo, CPUINFO_STRING) - - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_compute_machine_checksum_string(self, _mock_setup): - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - cm.cpuinfo = CPUINFO_STRING - cm.meminfo = MEMINFO_STRING - cm._ParseMemoryInfo() - cm._ComputeMachineChecksumString() - self.assertEqual(cm.checksum_string, CHECKSUM_STRING) - - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_get_md5_checksum(self, _mock_setup): - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - temp_str = 'abcde' - checksum_str = cm._GetMD5Checksum(temp_str) - self.assertEqual(checksum_str, 'ab56b4d92b40713acc5af89985d4b786') - - temp_str = '' - checksum_str = cm._GetMD5Checksum(temp_str) - self.assertEqual(checksum_str, '') - - @mock.patch.object(command_executer.CommandExecuter, 'CrosRunCommandWOutput') - @mock.patch.object(machine_manager.CrosMachine, 'SetUpChecksumInfo') - def test_get_machine_id(self, _mock_setup, mock_run_cmd): - cm = machine_manager.CrosMachine('daisy.cros', '/usr/local/chromeos', - 'average', self.mock_cmd_exec) - self.mock_cmd_exec.CrosRunCommandWOutput = mock_run_cmd - mock_run_cmd.return_value = [0, DUMP_VPD_STRING, ''] - - cm._GetMachineID() - self.assertEqual(cm.machine_id, '"Product_S/N"="HT4L91SC300208"') - - mock_run_cmd.return_value = [0, IFCONFIG_STRING, ''] - cm._GetMachineID() - self.assertEqual( - cm.machine_id, - ' ether 00:50:b6:63:db:65 txqueuelen 1000 (Ethernet)_ ' - 'ether e8:03:9a:9c:50:3d txqueuelen 1000 (Ethernet)_ ether ' - '44:6d:57:20:4a:c5 txqueuelen 1000 (Ethernet)') - - mock_run_cmd.return_value = [0, 'invalid hardware config', ''] - self.assertRaises(Exception, cm._GetMachineID) - - def test_add_cooldown_waittime(self): - cm = machine_manager.CrosMachine('1.2.3.4.cros', '/usr/local/chromeos', - 'average') - self.assertEqual(cm.GetCooldownWaitTime(), 0) - cm.AddCooldownWaitTime(250) - self.assertEqual(cm.GetCooldownWaitTime(), 250) - cm.AddCooldownWaitTime(1) - self.assertEqual(cm.GetCooldownWaitTime(), 251) - - -if __name__ == '__main__': - unittest.main() + """Test for CrosMachine class.""" + + mock_cmd_exec = mock.Mock(spec=command_executer.CommandExecuter) + + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_init(self, mock_setup): + + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + self.assertEqual(mock_setup.call_count, 1) + self.assertEqual(cm.chromeos_root, "/usr/local/chromeos") + self.assertEqual(cm.log_level, "average") + + @mock.patch.object(machine_manager.CrosMachine, "IsReachable") + @mock.patch.object(machine_manager.CrosMachine, "_GetMemoryInfo") + @mock.patch.object(machine_manager.CrosMachine, "_GetCPUInfo") + @mock.patch.object( + machine_manager.CrosMachine, "_ComputeMachineChecksumString" + ) + @mock.patch.object(machine_manager.CrosMachine, "_GetMachineID") + @mock.patch.object(machine_manager.CrosMachine, "_GetMD5Checksum") + def test_setup_checksum_info( + self, + mock_md5sum, + mock_machineid, + mock_checkstring, + mock_cpuinfo, + mock_meminfo, + mock_isreachable, + ): + + # Test 1. Machine is not reachable; SetUpChecksumInfo is called via + # __init__. + mock_isreachable.return_value = False + mock_md5sum.return_value = "md5_checksum" + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + cm.checksum_string = "This is a checksum string." + cm.machine_id = "machine_id1" + self.assertEqual(mock_isreachable.call_count, 1) + self.assertIsNone(cm.machine_checksum) + self.assertEqual(mock_meminfo.call_count, 0) + + # Test 2. Machine is reachable. Call explicitly. + mock_isreachable.return_value = True + cm.checksum_string = "This is a checksum string." + cm.machine_id = "machine_id1" + cm.SetUpChecksumInfo() + self.assertEqual(mock_isreachable.call_count, 2) + self.assertEqual(mock_meminfo.call_count, 1) + self.assertEqual(mock_cpuinfo.call_count, 1) + self.assertEqual(mock_checkstring.call_count, 1) + self.assertEqual(mock_machineid.call_count, 1) + self.assertEqual(mock_md5sum.call_count, 2) + self.assertEqual(cm.machine_checksum, "md5_checksum") + self.assertEqual(cm.machine_id_checksum, "md5_checksum") + self.assertEqual( + mock_md5sum.call_args_list[0][0][0], "This is a checksum string." + ) + self.assertEqual(mock_md5sum.call_args_list[1][0][0], "machine_id1") + + @mock.patch.object(command_executer.CommandExecuter, "CrosRunCommand") + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_is_reachable(self, mock_setup, mock_run_cmd): + + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + self.mock_cmd_exec.CrosRunCommand = mock_run_cmd + + # Test 1. CrosRunCommand returns 1 (fail) + mock_run_cmd.return_value = 1 + result = cm.IsReachable() + self.assertFalse(result) + self.assertEqual(mock_setup.call_count, 1) + self.assertEqual(mock_run_cmd.call_count, 1) + + # Test 2. CrosRunCommand returns 0 (success) + mock_run_cmd.return_value = 0 + result = cm.IsReachable() + self.assertTrue(result) + self.assertEqual(mock_run_cmd.call_count, 2) + first_args = mock_run_cmd.call_args_list[0] + second_args = mock_run_cmd.call_args_list[1] + self.assertEqual(first_args[0], second_args[0]) + self.assertEqual(first_args[1], second_args[1]) + self.assertEqual(len(first_args[0]), 1) + self.assertEqual(len(first_args[1]), 2) + self.assertEqual(first_args[0][0], "ls") + args_dict = first_args[1] + self.assertEqual(args_dict["machine"], "daisy.cros") + self.assertEqual(args_dict["chromeos_root"], "/usr/local/chromeos") + + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_parse_memory_info(self, _mock_setup): + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + cm.meminfo = MEMINFO_STRING + cm._ParseMemoryInfo() + self.assertEqual(cm.phys_kbytes, 4194304) + + @mock.patch.object( + command_executer.CommandExecuter, "CrosRunCommandWOutput" + ) + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_get_memory_info(self, _mock_setup, mock_run_cmd): + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + self.mock_cmd_exec.CrosRunCommandWOutput = mock_run_cmd + mock_run_cmd.return_value = [0, MEMINFO_STRING, ""] + cm._GetMemoryInfo() + self.assertEqual(mock_run_cmd.call_count, 1) + call_args = mock_run_cmd.call_args_list[0] + self.assertEqual(call_args[0][0], "cat /proc/meminfo") + args_dict = call_args[1] + self.assertEqual(args_dict["machine"], "daisy.cros") + self.assertEqual(args_dict["chromeos_root"], "/usr/local/chromeos") + self.assertEqual(cm.meminfo, MEMINFO_STRING) + self.assertEqual(cm.phys_kbytes, 4194304) + + mock_run_cmd.return_value = [1, MEMINFO_STRING, ""] + self.assertRaises(Exception, cm._GetMemoryInfo) + + @mock.patch.object( + command_executer.CommandExecuter, "CrosRunCommandWOutput" + ) + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_get_cpu_info(self, _mock_setup, mock_run_cmd): + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + self.mock_cmd_exec.CrosRunCommandWOutput = mock_run_cmd + mock_run_cmd.return_value = [0, CPUINFO_STRING, ""] + cm._GetCPUInfo() + self.assertEqual(mock_run_cmd.call_count, 1) + call_args = mock_run_cmd.call_args_list[0] + self.assertEqual(call_args[0][0], "cat /proc/cpuinfo") + args_dict = call_args[1] + self.assertEqual(args_dict["machine"], "daisy.cros") + self.assertEqual(args_dict["chromeos_root"], "/usr/local/chromeos") + self.assertEqual(cm.cpuinfo, CPUINFO_STRING) + + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_compute_machine_checksum_string(self, _mock_setup): + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + cm.cpuinfo = CPUINFO_STRING + cm.meminfo = MEMINFO_STRING + cm._ParseMemoryInfo() + cm._ComputeMachineChecksumString() + self.assertEqual(cm.checksum_string, CHECKSUM_STRING) + + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_get_md5_checksum(self, _mock_setup): + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + temp_str = "abcde" + checksum_str = cm._GetMD5Checksum(temp_str) + self.assertEqual(checksum_str, "ab56b4d92b40713acc5af89985d4b786") + + temp_str = "" + checksum_str = cm._GetMD5Checksum(temp_str) + self.assertEqual(checksum_str, "") + + @mock.patch.object( + command_executer.CommandExecuter, "CrosRunCommandWOutput" + ) + @mock.patch.object(machine_manager.CrosMachine, "SetUpChecksumInfo") + def test_get_machine_id(self, _mock_setup, mock_run_cmd): + cm = machine_manager.CrosMachine( + "daisy.cros", "/usr/local/chromeos", "average", self.mock_cmd_exec + ) + self.mock_cmd_exec.CrosRunCommandWOutput = mock_run_cmd + mock_run_cmd.return_value = [0, DUMP_VPD_STRING, ""] + + cm._GetMachineID() + self.assertEqual(cm.machine_id, '"Product_S/N"="HT4L91SC300208"') + + mock_run_cmd.return_value = [0, IFCONFIG_STRING, ""] + cm._GetMachineID() + self.assertEqual( + cm.machine_id, + " ether 00:50:b6:63:db:65 txqueuelen 1000 (Ethernet)_ " + "ether e8:03:9a:9c:50:3d txqueuelen 1000 (Ethernet)_ ether " + "44:6d:57:20:4a:c5 txqueuelen 1000 (Ethernet)", + ) + + mock_run_cmd.return_value = [0, "invalid hardware config", ""] + self.assertRaises(Exception, cm._GetMachineID) + + def test_add_cooldown_waittime(self): + cm = machine_manager.CrosMachine( + "1.2.3.4.cros", "/usr/local/chromeos", "average" + ) + self.assertEqual(cm.GetCooldownWaitTime(), 0) + cm.AddCooldownWaitTime(250) + self.assertEqual(cm.GetCooldownWaitTime(), 250) + cm.AddCooldownWaitTime(1) + self.assertEqual(cm.GetCooldownWaitTime(), 251) + + +if __name__ == "__main__": + unittest.main() |