Merge remote branch 'origin/upstream' into tempbranch

frontend/planner/execution_engine_unittest.py

+#!/usr/bin/python
+
+import unittest
+import common
+from autotest_lib.frontend import setup_django_environment
+from autotest_lib.client.common_lib.test_utils import mock
+from autotest_lib.frontend.afe import frontend_test_utils, models as afe_models
+from autotest_lib.frontend.afe import model_attributes as afe_model_attributes
+from autotest_lib.frontend.shared import rest_client
+from autotest_lib.frontend.planner import models, execution_engine, support
+from autotest_lib.frontend.planner import model_attributes
+
+
+class MockObject(object):
+    """
+    Empty mock object class, so that setattr() works on all names
+    """
+    pass
+
+
+class MockAfeRest(object):
+    jobs = MockObject()
+    execution_info = MockObject()
+    queue_entries_request = MockObject()
+
+
+class MockRestJobs(object):
+    def __init__(self, total_results):
+        self.total_results = total_results
+
+
+class MockExecutionInfo(object):
+    execution_info = {}
+
+
+class MockQueueEntriesRequest(object):
+    queue_entries = object()
+
+
+class MockExecutionEngine(execution_engine.ExecutionEngine):
+    _planner_rpc = MockObject()
+    _tko_rpc = object()
+    _plan_id = object()
+    _server = object()
+    _afe_rest = MockAfeRest()
+    _label_name = object()
+
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+
+class MockTestPlanController(support.TestPlanController):
+    def __init__(self, *args, **kwargs):
+        super(MockTestPlanController, self).__init__(machine=None,
+                                                     test_alias=None)
+
+
+class ExecutionEngineTest(unittest.TestCase,
+                          frontend_test_utils.FrontendTestMixin):
+    def setUp(self):
+        self._frontend_common_setup()
+        self.engine = MockExecutionEngine()
+
+
+    def tearDown(self):
+        self._frontend_common_teardown()
+
+
+    def _setup_test_initialize_plan(self):
+        self.god.stub_function(self.engine._planner_rpc, 'run')
+        self.god.stub_function(self.engine._afe_rest.jobs, 'get')
+        self.god.stub_function(self.engine, '_wait_for_initialization')
+
+
+    def test_initialize_plan_new_plan(self):
+        self._setup_test_initialize_plan()
+        self.god.stub_function(self.engine, '_launch_set_atomic_group_job')
+
+        self.engine._planner_rpc.run.expect_call(
+                'get_plan', id=self.engine._plan_id).and_return(
+                {'name': 'plan'})
+        self.engine._afe_rest.jobs.get.expect_call(
+                name='plan_set_atomic_group').and_return(MockRestJobs(None))
+        self.engine._launch_set_atomic_group_job.expect_call(
+                'plan_set_atomic_group')
+        self.engine._wait_for_initialization.expect_call()
+
+        self.engine._initialize_plan()
+        self.god.check_playback
+
+
+    def test_initialize_plan_existing(self):
+        self._setup_test_initialize_plan()
+
+        self.engine._planner_rpc.run.expect_call(
+                'get_plan', id=self.engine._plan_id).and_return(
+                {'name': 'plan'})
+        self.engine._afe_rest.jobs.get.expect_call(
+                name='plan_set_atomic_group').and_return(MockRestJobs(object()))
+        self.engine._wait_for_initialization.expect_call()
+
+        self.engine._initialize_plan()
+        self.god.check_playback
+
+
+    def _setup_test_launch_atomic_group_job(self, name):
+        DUMMY_CONTROL = object()
+        DUMMY_EXECUTION_INFO = MockExecutionInfo()
+        DUMMY_QUEUE_ENTRIES_REQUEST = MockQueueEntriesRequest()
+
+        self.god.stub_function(self.engine._planner_rpc, 'run')
+        self.god.stub_function(self.engine._afe_rest.execution_info, 'get')
+        self.god.stub_function(
+                self.engine._afe_rest.queue_entries_request, 'get')
+
+        self.engine._planner_rpc.run.expect_call(
+                'get_hosts', plan_id=self.engine._plan_id).and_return(
+                self.hosts)
+        self.engine._planner_rpc.run.expect_call(
+                'get_atomic_group_control_file').and_return(DUMMY_CONTROL)
+        self.engine._afe_rest.execution_info.get.expect_call().and_return(
+                DUMMY_EXECUTION_INFO)
+        self.engine._afe_rest.queue_entries_request.get.expect_call(
+                hosts=self.hosts).and_return(DUMMY_QUEUE_ENTRIES_REQUEST)
+
+        DUMMY_EXECUTION_INFO.execution_info = {
+                'control_file': DUMMY_CONTROL,
+                'cleanup_before_job': afe_model_attributes.RebootBefore.NEVER,
+                'cleanup_after_job': afe_model_attributes.RebootAfter.NEVER,
+                'run_verify': False,
+                'machines_per_execution': len(self.hosts)}
+
+        keyvals = {'server': self.engine._server,
+                   'label_name': self.engine._label_name,
+                   'plan_id': self.engine._plan_id}
+
+        job_req = {'name': name,
+                   'execution_info': DUMMY_EXECUTION_INFO.execution_info,
+                   'queue_entries': DUMMY_QUEUE_ENTRIES_REQUEST.queue_entries,
+                   'keyvals': keyvals}
+
+        return job_req
+
+
+    def test_launch_atomic_group_job(self):
+        job_req = self._setup_test_launch_atomic_group_job('atomic_group_job')
+        self.god.stub_function(self.engine._afe_rest.jobs, 'post')
+
+        self.engine._afe_rest.jobs.post.expect_call(job_req)
+
+        self.engine._launch_set_atomic_group_job('atomic_group_job')
+        self.god.check_playback()
+
+
+    def _setup_mock_controller(self, controller_options):
+        mock_controller = MockTestPlanController()
+        for key, value in controller_options.iteritems():
+            setattr(mock_controller, key, value)
+        self.god.stub_with(support, 'TestPlanController',
+                           lambda *args, **kwargs : mock_controller)
+        return mock_controller
+
+
+    def _test_process_finished_runs_helper(self, status, should_block=False,
+                                           controller_options={}):
+        Status = model_attributes.TestRunStatus
+        TEST_RUN_ID = object()
+        TKO_TEST_ID = object()
+        HOST_ID = object()
+
+        mock_controller = self._setup_mock_controller(controller_options)
+
+        self.god.stub_function(self.engine._planner_rpc, 'run')
+        self.god.stub_function(self.engine, '_run_execute_after')
+
+        test_run = {'id': TEST_RUN_ID,
+                    'host': {'host': self.hosts[0].hostname,
+                             'id': HOST_ID},
+                    'test_job': {'test_config': {'alias': 'test_alias'}},
+                    'tko_test': TKO_TEST_ID,
+                    'status': status}
+
+        self.engine._planner_rpc.run.expect_call(
+                'get_test_runs',
+                plan__id=self.engine._plan_id,
+                status__in=(Status.PASSED, Status.FAILED),
+                finalized=False).and_return([test_run])
+        self.engine._run_execute_after.expect_call(
+                mock_controller, tko_test_id=TKO_TEST_ID,
+                success=(status == Status.PASSED))
+        if should_block:
+            self.engine._planner_rpc.run.expect_call('modify_host', id=HOST_ID,
+                                                     blocked=True)
+        self.engine._planner_rpc.run.expect_call('modify_test_run',
+                                                 id=TEST_RUN_ID, finalized=True)
+
+        self.engine._process_finished_runs()
+
+        self.god.check_playback()
+
+
+    def test_process_finished_runs_pass(self):
+        self._test_process_finished_runs_helper(
+                model_attributes.TestRunStatus.PASSED)
+
+
+    def test_process_finished_runs_fail(self):
+        self._test_process_finished_runs_helper(
+                model_attributes.TestRunStatus.FAILED, should_block=True)
+
+
+    def test_process_finished_runs_fail_unblock(self):
+        self._test_process_finished_runs_helper(
+                model_attributes.TestRunStatus.FAILED, should_block=False,
+                controller_options={'_unblock': True})
+
+
+    def _test_schedule_new_runs_helper(self, complete=False, should_skip=False,
+                                       controller_options={}):
+        TEST_CONFIG_ID = object()
+
+        self.god.stub_function(self.engine._planner_rpc, 'run')
+        self.god.stub_function(self.engine, '_run_execute_before')
+
+        result = {'complete': complete,
+                  'next_configs': [{'next_test_config_id': TEST_CONFIG_ID,
+                                    'host': self.hosts[0].hostname,
+                                    'next_test_config_alias': object()}]}
+
+        mock_controller = self._setup_mock_controller(controller_options)
+
+        self.engine._planner_rpc.run.expect_call(
+                'get_next_test_configs',
+                plan_id=self.engine._plan_id).and_return(result)
+
+        if not complete:
+            self.engine._run_execute_before.expect_call(mock_controller)
+
+            if should_skip:
+                self.engine._planner_rpc.run.expect_call(
+                        'skip_test', test_config_id=TEST_CONFIG_ID,
+                        hostname=self.hosts[0].hostname)
+            else:
+                self.god.stub_function(self.engine, '_run_job')
+                self.engine._run_job.expect_call(
+                        hostname=self.hosts[0].hostname,
+                        test_config_id=TEST_CONFIG_ID,
+                        cleanup_before_job=mock_controller._reboot_before,
+                        cleanup_after_job=mock_controller._reboot_after,
+                        run_verify=mock_controller._run_verify)
+
+        self.engine._schedule_new_runs()
+
+        self.god.check_playback()
+
+
+    def test_schedule_new_runs(self):
+        self._test_schedule_new_runs_helper()
+
+
+    def test_schedule_new_runs_complete(self):
+        self._test_schedule_new_runs_helper(complete=True)
+
+
+    def test_schedule_new_runs_skip(self):
+        self._test_schedule_new_runs_helper(should_skip=True,
+                                            controller_options={'_skip': True})
+
+
+    def test_run_global_support(self):
+        self._ran_global_support = False
+        support = """
+def test_global_support(controller):
+    controller._ran_global_support = True
+"""
+
+        DUMMY_PLAN = {'support': support}
+
+        self.god.stub_function(self.engine._planner_rpc, 'run')
+
+        self.engine._planner_rpc.run.expect_call(
+                'get_plan', id=self.engine._plan_id).and_return(DUMMY_PLAN)
+
+        self.engine._run_global_support(controller=self,
+                                        function_name='test_global_support')
+
+        self.assertTrue(self._ran_global_support)
+        self.god.check_playback()
+
+
+if __name__ == '__main__':
+    unittest.main()