Implementation of Test Planner execution engine. Is currently able to
schedule single-host tests and place the results in the proper planner
tables for analysis.
TODO: global support object, execution_engine.py unit tests
Signed-off-by: James Ren <jamesren@google.com>
git-svn-id: http://test.kernel.org/svn/autotest/trunk@4301 592f7852-d20e-0410-864c-8624ca9c26a4
diff --git a/frontend/planner/rpc_interface_unittest.py b/frontend/planner/rpc_interface_unittest.py
index 0f1c71b..081e433 100644
--- a/frontend/planner/rpc_interface_unittest.py
+++ b/frontend/planner/rpc_interface_unittest.py
@@ -3,10 +3,11 @@
import unittest
import common
from autotest_lib.frontend import setup_django_environment
-from autotest_lib.frontend.planner import planner_test_utils
+from autotest_lib.frontend.planner import planner_test_utils, model_attributes
+from autotest_lib.frontend.planner import rpc_interface, models, rpc_utils
from autotest_lib.frontend.afe import model_logic
from autotest_lib.frontend.afe import models as afe_models
-from autotest_lib.frontend.planner import rpc_interface, models, rpc_utils
+from autotest_lib.frontend.tko import models as tko_models
class RpcInterfaceTest(unittest.TestCase,
@@ -72,5 +73,110 @@
self.assertEqual(set(('host1', 'host2')), set(hosts))
+ def test_get_next_test_configs(self):
+ DUMMY_CONFIGS = {'host1': object(),
+ 'host2': object()}
+ DUMMY_COMPLETE = object()
+ self.god.stub_function(rpc_utils, 'compute_next_test_config')
+
+ for host in models.Host.objects.filter(plan=self._plan):
+ rpc_utils.compute_next_test_config.expect_call(
+ self._plan, host).and_return(
+ DUMMY_CONFIGS[host.host.hostname])
+
+ def _dummy_check_for_completion(plan):
+ plan.complete = DUMMY_COMPLETE
+ rpc_utils.check_for_completion = _dummy_check_for_completion
+
+ result = rpc_interface.get_next_test_configs(self._plan.id)
+
+ self.god.check_playback()
+ self.assertEqual(result['complete'], DUMMY_COMPLETE)
+ for config in result['next_configs']:
+ self.assertTrue(config['host'] in DUMMY_CONFIGS)
+ self.assertEqual(config['next_test_config_id'],
+ DUMMY_CONFIGS[config['host']])
+
+
+ def test_update_test_runs(self):
+ GOOD_STATUS_WORD = 'GOOD'
+ RUNNING_STATUS_WORD = 'RUNNING'
+ hostname = self.hosts[0].hostname
+
+ self.god.stub_function(rpc_utils, 'compute_test_run_status')
+ self.god.stub_function(rpc_utils, 'add_test_run')
+
+ control, _ = models.ControlFile.objects.get_or_create(
+ contents='test_control')
+ test_config = models.TestConfig.objects.create(plan=self._plan,
+ alias='config',
+ control_file=control,
+ execution_order=1,
+ estimated_runtime=1)
+ afe_job = self._create_job(hosts=(1,))
+ planner_host = models.Host.objects.create(plan=self._plan,
+ host=self.hosts[0])
+ planner_job = models.Job.objects.create(plan=self._plan,
+ test_config=test_config,
+ afe_job=afe_job)
+ tko_machine = tko_models.Machine.objects.create(hostname=hostname)
+ tko_job = tko_models.Job.objects.create(tag='job',
+ machine=tko_machine,
+ afe_job_id=afe_job.id)
+ tko_kernel = tko_models.Kernel.objects.create()
+ running_status = tko_models.Status.objects.create(
+ word=RUNNING_STATUS_WORD)
+ good_status = tko_models.Status.objects.create(word=GOOD_STATUS_WORD)
+
+ # No TKO tests
+ self.assertEqual([], rpc_interface.update_test_runs(self._plan.id))
+ self.god.check_playback()
+
+ # active TKO test
+ tko_test = tko_models.Test.objects.create(job=tko_job,
+ machine=tko_machine,
+ kernel=tko_kernel,
+ status=running_status)
+
+ rpc_utils.compute_test_run_status.expect_call(
+ RUNNING_STATUS_WORD).and_return(
+ model_attributes.TestRunStatus.ACTIVE)
+ rpc_utils.add_test_run.expect_call(
+ self._plan, planner_job, tko_test, hostname,
+ model_attributes.TestRunStatus.ACTIVE)
+ self.assertEqual(rpc_interface.update_test_runs(self._plan.id),
+ [{'status': model_attributes.TestRunStatus.ACTIVE,
+ 'tko_test_idx': tko_test.test_idx,
+ 'hostname': hostname}])
+ self.god.check_playback()
+ test_run = models.TestRun.objects.create(
+ plan=self._plan, test_job=planner_job,
+ tko_test=tko_test, host=planner_host,
+ status=model_attributes.TestRunStatus.ACTIVE)
+
+ # no change to TKO test
+ rpc_utils.compute_test_run_status.expect_call(
+ RUNNING_STATUS_WORD).and_return(
+ model_attributes.TestRunStatus.ACTIVE)
+ self.assertEqual([], rpc_interface.update_test_runs(self._plan.id))
+ self.god.check_playback()
+
+ # TKO test is now complete, passed
+ tko_test.status = good_status
+ tko_test.save()
+
+ rpc_utils.compute_test_run_status.expect_call(
+ GOOD_STATUS_WORD).and_return(
+ model_attributes.TestRunStatus.PASSED)
+ rpc_utils.add_test_run.expect_call(
+ self._plan, planner_job, tko_test, hostname,
+ model_attributes.TestRunStatus.PASSED)
+ self.assertEqual(rpc_interface.update_test_runs(self._plan.id),
+ [{'status': model_attributes.TestRunStatus.PASSED,
+ 'tko_test_idx': tko_test.test_idx,
+ 'hostname': hostname}])
+ self.god.check_playback()
+
+
if __name__ == '__main__':
unittest.main()