Add Planner RPC utils unit tests
Signed-off-by: James Ren <jamesren@google.com>
git-svn-id: http://test.kernel.org/svn/autotest/trunk@4346 592f7852-d20e-0410-864c-8624ca9c26a4
diff --git a/frontend/planner/rpc_interface_unittest.py b/frontend/planner/rpc_interface_unittest.py
index 081e433..56c8e0a 100644
--- a/frontend/planner/rpc_interface_unittest.py
+++ b/frontend/planner/rpc_interface_unittest.py
@@ -99,82 +99,58 @@
def test_update_test_runs(self):
- GOOD_STATUS_WORD = 'GOOD'
- RUNNING_STATUS_WORD = 'RUNNING'
- hostname = self.hosts[0].hostname
+ self._setup_active_plan()
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)
+ tko_test = tko_models.Test.objects.create(job=self._tko_job,
+ machine=self._tko_machine,
+ kernel=self._tko_kernel,
+ status=self._running_status)
rpc_utils.compute_test_run_status.expect_call(
- RUNNING_STATUS_WORD).and_return(
+ self.RUNNING_STATUS_WORD).and_return(
model_attributes.TestRunStatus.ACTIVE)
rpc_utils.add_test_run.expect_call(
- self._plan, planner_job, tko_test, hostname,
+ self._plan, self._planner_job, tko_test, self._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}])
+ 'hostname': self._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,
+ plan=self._plan, test_job=self._planner_job,
+ tko_test=tko_test, host=self._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(
+ self.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.status = self._good_status
tko_test.save()
rpc_utils.compute_test_run_status.expect_call(
- GOOD_STATUS_WORD).and_return(
+ self.GOOD_STATUS_WORD).and_return(
model_attributes.TestRunStatus.PASSED)
rpc_utils.add_test_run.expect_call(
- self._plan, planner_job, tko_test, hostname,
+ self._plan, self._planner_job, tko_test, self._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}])
+ 'hostname': self._hostname}])
self.god.check_playback()