| import mock |
| import unittest |
| |
| from autotest_lib.client.common_lib.cros.cfm.usb import usb_device |
| from autotest_lib.client.common_lib.cros.cfm.usb import usb_device_spec |
| from autotest_lib.server.cros.cfm.configurable_test import actions |
| from autotest_lib.server.cros.cfm.configurable_test import action_context |
| from autotest_lib.server.cros.cfm.configurable_test import scenario |
| |
| # Constants to use in case the actual values are irrelevant. |
| USB_DEVICE_SPEC = usb_device_spec.UsbDeviceSpec( |
| 'vid', 'pid', 'product', ['iface']) |
| |
| USB_DEVICE = usb_device.UsbDevice('v', 'p', 'prod', ['if'], 1, 2, 1) |
| |
| |
| # Test, disable missing-docstring |
| # pylint: disable=missing-docstring |
| class TestActions(unittest.TestCase): |
| """ |
| Tests for the available actions for configurable CFM tests to run. |
| """ |
| |
| def setUp(self): |
| self.host_mock = mock.MagicMock() |
| self.cfm_facade_mock = mock.MagicMock() |
| self.usb_device_collector_mock = mock.MagicMock() |
| self.usb_port_manager_mock = mock.MagicMock() |
| self.crash_detector_mock = mock.MagicMock() |
| self.metrics_collector_mock = mock.MagicMock() |
| self.context_with_mocks = action_context.ActionContext( |
| host=self.host_mock, |
| cfm_facade=self.cfm_facade_mock, |
| usb_device_collector=self.usb_device_collector_mock, |
| usb_port_manager=self.usb_port_manager_mock, |
| crash_detector=self.crash_detector_mock, |
| perf_metrics_collector=self.metrics_collector_mock) |
| |
| |
| def test_assert_file_does_not_contain_no_match(self): |
| action = actions.AssertFileDoesNotContain('/foo', ['EE', 'WW']) |
| context = action_context.ActionContext( |
| file_contents_collector=FakeCollector('abc\ndef')) |
| action.execute(context) |
| |
| def test_assert_file_does_not_contain_match(self): |
| action = actions.AssertFileDoesNotContain('/foo', ['EE', 'WW']) |
| context = action_context.ActionContext( |
| file_contents_collector=FakeCollector('abc\naWWd')) |
| self.assertRaises(AssertionError, lambda: action.execute(context)) |
| |
| def test_assert_file_does_not_contain_regex_match(self): |
| action = actions.AssertFileDoesNotContain('/foo', ['EE', 'W{3}Q+']) |
| context = action_context.ActionContext( |
| file_contents_collector=FakeCollector('abc\naWWWQQd')) |
| self.assertRaises(AssertionError, lambda: action.execute(context)) |
| |
| def test_reboot_dut_no_restart(self): |
| action = actions.RebootDut() |
| action.execute(self.context_with_mocks) |
| self.host_mock.reboot.assert_called_once_with() |
| self.assertFalse(self.cfm_facade_mock.method_calls) |
| |
| def test_reboot_dut_with_restart(self): |
| action = actions.RebootDut(restart_chrome_for_cfm=True) |
| action.execute(self.context_with_mocks) |
| self.host_mock.reboot.assert_called_once_with() |
| (self.cfm_facade_mock.restart_chrome_for_cfm |
| .assert_called_once_with()) |
| (self.cfm_facade_mock.wait_for_telemetry_commands |
| .assert_called_once_with()) |
| |
| def test_assert_usb_device_collector(self): |
| spec = usb_device_spec.UsbDeviceSpec( |
| 'vid', 'pid', 'product', ['iface']) |
| action = actions.AssertUsbDevices([spec], lambda x: True) |
| action.execute(self.context_with_mocks) |
| |
| def test_assert_usb_device_collector_matching_predicate(self): |
| spec = usb_device_spec.UsbDeviceSpec( |
| 'vid', 'pid', 'product', ['iface']) |
| device = usb_device.UsbDevice( |
| 'v', 'p', 'prod', ['if'], 1, 2, 1) |
| self.usb_device_collector_mock.get_devices_by_spec = mock.Mock( |
| return_value=[device]) |
| action = actions.AssertUsbDevices( |
| [spec], lambda x: x[0].product_id == 'p') |
| action.execute(self.context_with_mocks) |
| |
| def test_assert_usb_device_collector_non_matching_predicate(self): |
| spec = usb_device_spec.UsbDeviceSpec( |
| 'vid', 'pid', 'product', ['iface']) |
| device = usb_device.UsbDevice( |
| 'v', 'p', 'prod', ['if'], 1, 2, 1) |
| self.usb_device_collector_mock.get_devices_by_spec = mock.Mock( |
| return_value=[device]) |
| action = actions.AssertUsbDevices( |
| [spec], lambda x: x[0].product_id == 'r') |
| self.assertRaises(AssertionError, lambda: action.execute( |
| self.context_with_mocks)) |
| |
| def test_assert_usb_device_collector_default_predicate(self): |
| self.usb_device_collector_mock.get_devices_by_spec = mock.Mock( |
| return_value=[USB_DEVICE]) # Default checks list is of size 1 |
| action = actions.AssertUsbDevices([USB_DEVICE_SPEC]) |
| action.execute(self.context_with_mocks) |
| |
| def test_select_scenario_at_random(self): |
| dummy_action1 = DummyAction() |
| dummy_action2 = DummyAction() |
| scenarios = [scenario.Scenario(dummy_action1), |
| scenario.Scenario(dummy_action2)] |
| action = actions.SelectScenarioAtRandom(scenarios, 10) |
| action.execute(self.context_with_mocks) |
| # Assert that our actions were executed the expected number of times. |
| total_executes = (dummy_action1.executed_times |
| + dummy_action2.executed_times) |
| self.assertEqual(10, total_executes) |
| |
| def test_select_scenario_at_random_str_contains_seed(self): |
| action = actions.SelectScenarioAtRandom([], 10, 123) |
| self.assertTrue('seed=123' in str(action)) |
| |
| def test_select_scenario_at_random_same_seed_same_actions(self): |
| scenario1_action1 = DummyAction() |
| scenario1_action2 = DummyAction() |
| scenarios1 = [scenario.Scenario(scenario1_action1), |
| scenario.Scenario(scenario1_action2)] |
| scenario2_action1 = DummyAction() |
| scenario2_action2 = DummyAction() |
| scenarios2 = [scenario.Scenario(scenario2_action1), |
| scenario.Scenario(scenario2_action2)] |
| action1 = actions.SelectScenarioAtRandom(scenarios1, 100, 0) |
| action2 = actions.SelectScenarioAtRandom(scenarios2, 100, 0) |
| action1.execute(self.context_with_mocks) |
| action2.execute(self.context_with_mocks) |
| self.assertEqual(scenario1_action1.executed_times, |
| scenario2_action1.executed_times) |
| self.assertEqual(scenario1_action2.executed_times, |
| scenario2_action2.executed_times) |
| |
| def test_power_cycle_usb_port(self): |
| device = usb_device.UsbDevice( |
| 'v', 'p', 'prod', ['if'], 1, 2, 1) |
| self.usb_device_collector_mock.get_devices_by_spec = mock.Mock( |
| side_effect=[[device, device], [device], [device, device]]) |
| action = actions.PowerCycleUsbPort( |
| [USB_DEVICE_SPEC], 0, lambda x: [x[0]]) |
| action.execute(self.context_with_mocks) |
| self.usb_port_manager_mock.set_port_power.assert_has_calls( |
| [mock.call([(1, 2)], False), mock.call([(1, 2)], True)]) |
| |
| def test_power_cycle_usb_port_device_does_not_turn_off(self): |
| # Return the same device all the time - i.e., it does not turn off. |
| self.usb_device_collector_mock.get_devices_by_spec = mock.Mock( |
| return_value=[USB_DEVICE]) |
| action = actions.PowerCycleUsbPort([USB_DEVICE_SPEC], 0) |
| self.assertRaises( |
| actions.TimeoutError, |
| lambda: action.execute(self.context_with_mocks)) |
| |
| def test_power_cycle_usb_port_device_does_not_turn_on(self): |
| self.usb_device_collector_mock.get_devices_by_spec = mock.Mock( |
| side_effect=[[USB_DEVICE, USB_DEVICE], [], [USB_DEVICE]]) |
| action = actions.PowerCycleUsbPort([USB_DEVICE_SPEC], 0) |
| self.assertRaises( |
| actions.TimeoutError, |
| lambda: action.execute(self.context_with_mocks)) |
| |
| def test_retry_action_success_after_retry(self): |
| action = actions.RetryAssertAction(RaisesFirstTimeAction(), 3, 0) |
| action.execute(self.context_with_mocks) |
| |
| def test_retry_action_fail_when_no_more_retries(self): |
| action = actions.RetryAssertAction(RaisesFirstTimeAction(), 1) |
| self.assertRaises( |
| AssertionError, lambda: action.execute(self.context_with_mocks)) |
| |
| def test_assert_no_new_crashes(self): |
| action = actions.AssertNoNewCrashes() |
| self.crash_detector_mock.get_new_crash_files = mock.Mock( |
| return_value=[]) |
| action.do_execute(self.context_with_mocks) |
| |
| def test_assert_no_new_crashes_crash_detected(self): |
| action = actions.AssertNoNewCrashes() |
| self.crash_detector_mock.get_new_crash_files = mock.Mock( |
| return_value=['/a/new/crash/file']) |
| self.assertRaises( |
| AssertionError, |
| lambda: action.do_execute(self.context_with_mocks)) |
| |
| def test_start_metrics_colllection(self): |
| action = actions.StartPerfMetricsCollection() |
| action.execute(self.context_with_mocks) |
| self.metrics_collector_mock.start.assert_called_once_with() |
| |
| def test_stop_metrics_colllection(self): |
| action = actions.StopPerfMetricsCollection() |
| action.execute(self.context_with_mocks) |
| self.metrics_collector_mock.stop.assert_called_once_with() |
| |
| def test_upload_metrics(self): |
| action = actions.UploadPerfMetrics() |
| action.execute(self.context_with_mocks) |
| self.metrics_collector_mock.upload_metrics.assert_called_once_with() |
| |
| |
| |
| class FakeCollector(object): |
| def __init__(self, contents): |
| self.contents = contents |
| |
| def collect_file_contents(self, path): |
| return self.contents |
| |
| class DummyAction(actions.Action): |
| def __init__(self): |
| self.executed_times = 0 |
| |
| def do_execute(self, context): |
| self.executed_times += 1 |
| |
| class RaisesFirstTimeAction(actions.Action): |
| def __init__(self): |
| self.executed = False |
| |
| def do_execute(self, context): |
| if not self.executed: |
| self.executed = True |
| raise AssertionError() |
| |
