Add Planner RPC utils unit tests
Signed-off-by: James Ren <[email protected]>
--- autotest/frontend/planner/planner_test_utils.py 2010-03-24
12:39:54.000000000 -0700
+++ autotest/frontend/planner/planner_test_utils.py 2010-03-24
12:39:54.000000000 -0700
@@ -1,11 +1,14 @@
import common
from autotest_lib.frontend.afe import frontend_test_utils
from autotest_lib.frontend.afe import models as afe_models
+from autotest_lib.frontend.tko import models as tko_models
from autotest_lib.frontend.planner import models
from autotest_lib.client.common_lib import utils
class PlannerTestMixin(frontend_test_utils.FrontendTestMixin):
_PLAN_NAME = 'plan'
+ GOOD_STATUS_WORD = 'GOOD'
+ RUNNING_STATUS_WORD = 'RUNNING'
def _planner_common_setup(self):
self._frontend_common_setup()
@@ -24,3 +27,50 @@
def _planner_common_teardown(self):
self._plan.delete()
self._frontend_common_teardown()
+
+
+ def _setup_active_plan(self):
+ """
+ Create an active test plan
+
+ Sets up all the infrastructure for a active test plan. Stores the
+ following in self:
+
+ _hostname: hostname of the machine under test
+ _control: the models.ControlFile object
+ _test_config: the models.TestConfig object
+ _afe_job: the AFE job started by the plan
+ _planner_host: the models.Host object
+ _planner_job: the models.Job object
+ _tko_machine: the TKO machine (as a tko_models.Machine object) for the
+ results
+ _tko_job: the TKO job (as a tko_models.Job object) for the results
+ _tko_kernel: the TKO kernel (as a tko_models.Kernel object) associated
+ with the TKO machine
+ _running_status: the TKO status (as a tko_models.Status object) that
+ indicates a running TKO test
+ _good_status: the TKO status (as a tko_models.Status object) that
+ indicates a completed and passed TKO test
+ """
+ self._hostname = self.hosts[0].hostname
+ self._control, _ = models.ControlFile.objects.get_or_create(
+ contents='test_control')
+ self._test_config = models.TestConfig.objects.create(
+ plan=self._plan, alias='config', control_file=self._control,
+ execution_order=1, estimated_runtime=1)
+ self._afe_job = self._create_job(hosts=(1,))
+ self._planner_host = models.Host.objects.create(plan=self._plan,
+ host=self.hosts[0])
+ self._planner_job = models.Job.objects.create(
+ plan=self._plan, test_config=self._test_config,
+ afe_job=self._afe_job)
+ self._tko_machine = tko_models.Machine.objects.create(
+ hostname=self._hostname)
+ self._tko_job = tko_models.Job.objects.create(
+ tag='job', machine=self._tko_machine,
+ afe_job_id=self._afe_job.id)
+ self._tko_kernel = tko_models.Kernel.objects.create()
+ self._running_status = tko_models.Status.objects.create(
+ word=self.RUNNING_STATUS_WORD)
+ self._good_status = tko_models.Status.objects.create(
+ word=self.GOOD_STATUS_WORD)
--- autotest/frontend/planner/rpc_interface_unittest.py 2010-03-24
12:39:54.000000000 -0700
+++ autotest/frontend/planner/rpc_interface_unittest.py 2010-03-24
12:39:54.000000000 -0700
@@ -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()
--- autotest/frontend/planner/rpc_utils_unittest.py 2010-03-24
12:39:54.000000000 -0700
+++ autotest/frontend/planner/rpc_utils_unittest.py 2010-03-24
12:39:54.000000000 -0700
@@ -7,7 +7,7 @@
from autotest_lib.frontend.afe import model_logic, models as afe_models
from autotest_lib.frontend.afe import rpc_interface as afe_rpc_interface
from autotest_lib.frontend.planner import models, rpc_utils
-from autotest_lib.client.common_lib import utils
+from autotest_lib.client.common_lib import utils, host_queue_entry_states
class RpcUtilsTest(unittest.TestCase,
@@ -65,5 +65,73 @@
self.god.check_playback()
+ def test_update_hosts_table(self):
+ label = self.labels[3]
+ default_hosts = set(self._plan.hosts.all())
+
+ rpc_utils.update_hosts_table(self._plan)
+ self.assertEqual(default_hosts, set(self._plan.hosts.all()))
+ self.assertEqual(set(), self._get_added_by_label_hosts())
+
+ self._plan.host_labels.add(label)
+ rpc_utils.update_hosts_table(self._plan)
+ self.assertEqual(default_hosts.union(label.host_set.all()),
+ set(self._plan.hosts.all()))
+ self.assertEqual(set(label.host_set.all()),
+ self._get_added_by_label_hosts())
+
+ self._plan.host_labels.remove(label)
+ rpc_utils.update_hosts_table(self._plan)
+ self.assertEqual(default_hosts, set(self._plan.hosts.all()))
+ self.assertEqual(set(), self._get_added_by_label_hosts())
+
+
+ def _get_added_by_label_hosts(self):
+ return set(host.host for host in models.Host.objects.filter(
+ plan=self._plan, added_by_label=True))
+
+
+ def test_compute_next_test_config(self):
+ self._setup_active_plan()
+ test_config = models.TestConfig.objects.create(
+ plan=self._plan, alias='config2', control_file=self._control,
+ execution_order=2, estimated_runtime=1)
+
+ self.assertEqual(1, self._afe_job.hostqueueentry_set.count())
+ self.assertEqual(
+ None, rpc_utils.compute_next_test_config(self._plan,
+ self._planner_host))
+ self.assertFalse(self._planner_host.complete)
+
+ hqe = self._afe_job.hostqueueentry_set.all()[0]
+ hqe.status = host_queue_entry_states.Status.COMPLETED
+ hqe.save()
+
+ self.assertEqual(
+ test_config.id,
+ rpc_utils.compute_next_test_config(self._plan,
+ self._planner_host))
+ self.assertFalse(self._planner_host.complete)
+
+ afe_job = self._create_job(hosts=(1,))
+ planner_job = models.Job.objects.create(plan=self._plan,
+ test_config=test_config,
+ afe_job=afe_job)
+
+ self.assertEqual(
+ None, rpc_utils.compute_next_test_config(self._plan,
+ self._planner_host))
+ self.assertFalse(self._planner_host.complete)
+
+ hqe = afe_job.hostqueueentry_set.all()[0]
+ hqe.status = host_queue_entry_states.Status.COMPLETED
+ hqe.save()
+
+ self.assertEqual(
+ None, rpc_utils.compute_next_test_config(self._plan,
+ self._planner_host))
+ self.assertTrue(self._planner_host.complete)
+
+
if __name__ == '__main__':
unittest.main()
_______________________________________________
Autotest mailing list
[email protected]
http://test.kernel.org/cgi-bin/mailman/listinfo/autotest
