Add a RH-style init script for the autotest scheduler
[autotest-zwu.git] / frontend / planner / models_test.py
blob21219921f476d2bc55f656f69402a07926ded46a
1 #!/usr/bin/python
3 import unittest
4 import common
5 from autotest_lib.frontend import setup_django_environment
6 from autotest_lib.frontend.afe import frontend_test_utils, rpc_utils
7 from autotest_lib.frontend.tko import models as tko_models
8 from autotest_lib.frontend.planner import models, model_attributes
9 from autotest_lib.frontend.planner import planner_test_utils
12 class ModelWithHashTestBase(frontend_test_utils.FrontendTestMixin):
13 def setUp(self):
14 self._frontend_common_setup(fill_data=False)
17 def tearDown(self):
18 self._frontend_common_teardown()
21 def _model_class(self):
22 raise NotImplementedError('Subclasses must override _model_class()')
25 def _test_data(self):
26 raise NotImplementedError('Subclasses must override _test_data()')
29 def test_disallowed_operations(self):
30 def _call_create():
31 self._model_class().objects.create(**self._test_data())
32 self.assertRaises(Exception, _call_create)
34 model = self._model_class().objects.get_or_create(
35 **self._test_data())[0]
36 self.assertRaises(Exception, model.save)
39 def test_hash_field(self):
40 model = self._model_class().objects.get_or_create(
41 **self._test_data())[0]
42 self.assertNotEqual(model.id, None)
43 self.assertEqual(self._model_class()._compute_hash(**self._test_data()),
44 model.the_hash)
47 class ControlFileTest(ModelWithHashTestBase, unittest.TestCase):
48 def _model_class(self):
49 return models.ControlFile
52 def _test_data(self):
53 return {'contents' : 'test_control'}
56 class KeyValTest(ModelWithHashTestBase, unittest.TestCase):
57 def _model_class(self):
58 return models.KeyVal
61 def _test_data(self):
62 return {'key' : 'test_key',
63 'value' : 'test_value'}
66 class AdditionalParameterTest(frontend_test_utils.FrontendTestMixin,
67 unittest.TestCase):
68 def setUp(self):
69 self._frontend_common_setup()
70 self.plan = models.Plan.objects.create(name='plan')
71 self.param_type = model_attributes.AdditionalParameterType.VERIFY
73 def tearDown(self):
74 self._frontend_common_teardown()
77 def test_find_applicable_control_parameter_match(self):
78 parameter = models.AdditionalParameter.objects.create(
79 plan=self.plan, hostname_regex='host.*',
80 param_type=self.param_type, application_order=0)
81 found = models.AdditionalParameter.find_applicable_additional_parameter(
82 plan=self.plan, hostname='host1', param_type=self.param_type)
84 self.assertEqual(parameter, found)
87 def test_find_applicable_additional_parameter_ordered(self):
88 additional1 = models.AdditionalParameter.objects.create(
89 plan=self.plan, hostname_regex='host.*',
90 param_type=self.param_type, application_order=0)
91 additional2 = models.AdditionalParameter.objects.create(
92 plan=self.plan, hostname_regex='.*',
93 param_type=self.param_type, application_order=1)
95 found1 = (
96 models.AdditionalParameter.find_applicable_additional_parameter(
97 plan=self.plan, hostname='host1',
98 param_type=self.param_type))
99 found2 = (
100 models.AdditionalParameter.find_applicable_additional_parameter(
101 plan=self.plan, hostname='other',
102 param_type=self.param_type))
104 self.assertEqual(additional1, found1)
105 self.assertEqual(additional2, found2)
108 def test_find_applicable_additional_parameter_no_match(self):
109 models.AdditionalParameter.objects.create(
110 plan=self.plan, hostname_regex='host.*',
111 param_type=self.param_type, application_order=0)
112 found = models.AdditionalParameter.find_applicable_additional_parameter(
113 plan=self.plan, hostname='other', param_type=self.param_type)
115 self.assertEqual(None, found)
118 class JobTest(planner_test_utils.PlannerTestMixin,
119 unittest.TestCase):
120 def setUp(self):
121 self._planner_common_setup()
122 self._setup_active_plan()
125 def tearDown(self):
126 self._planner_common_teardown()
129 def test_active(self):
130 self.assertEqual(True, self._planner_job.active())
131 self._afe_job.hostqueueentry_set.update(complete=True)
132 self.assertEqual(False, self._planner_job.active())
135 def test_all_tests_passed_active(self):
136 self.assertEqual(True, self._planner_job.active())
137 self.assertEqual(False, self._planner_job.all_tests_passed())
140 def test_all_tests_passed_failed_queue_entry(self):
141 self._afe_job.hostqueueentry_set.update(complete=True, status='Failed')
142 self.assertEqual(False, self._planner_job.active())
144 self.assertEqual(False, self._planner_job.all_tests_passed())
147 def _setup_test_all_tests_passed(self, status):
148 self._afe_job.hostqueueentry_set.update(complete=True,
149 status='Completed')
150 tko_test = tko_models.Test.objects.create(job=self._tko_job,
151 status=status,
152 kernel=self._tko_kernel,
153 machine=self._tko_machine)
154 self.assertEqual(False, self._planner_job.active())
157 def test_all_tests_passed_success(self):
158 self._setup_test_all_tests_passed(self._good_status)
159 self.assertEqual(True, self._planner_job.all_tests_passed())
162 def test_all_tests_passed_failure(self):
163 self._setup_test_all_tests_passed(self._fail_status)
164 self.assertEqual(False, self._planner_job.all_tests_passed())
167 if __name__ == '__main__':
168 unittest.main()