Fix an error in drone sets in monitor_db. Also added more unit tests.
Signed-off-by: James Ren <[email protected]>
--- autotest/scheduler/monitor_db.py 2010-04-26 11:29:56.000000000 -0700
+++ autotest/scheduler/monitor_db.py 2010-04-27 10:34:02.000000000 -0700
@@ -1700,7 +1700,7 @@
job = models.Job.objects.get(id=job_ids[0])
drone_set = job.drone_set
if not drone_set:
- return self_user_or_global_default_drone_set(job, job.user())
+ return self._user_or_global_default_drone_set(job, job.user())
return drone_set.get_drone_hostnames()
--- autotest/scheduler/monitor_db_unittest.py 2010-04-27 10:34:02.000000000
-0700
+++ autotest/scheduler/monitor_db_unittest.py 2010-04-27 10:34:02.000000000
-0700
@@ -1393,34 +1393,127 @@
job_3 = self._create_job_simple([self.hosts[0].id],
drone_set=drone_set_3)
- hqe_1 = job_1.hostqueueentry_set.all()[0].id
- hqe_2 = job_2.hostqueueentry_set.all()[0].id
- hqe_3 = job_3.hostqueueentry_set.all()[0].id
+ job_4 = self._create_job_simple([self.hosts[0].id])
+ job_4.drone_set = None
+ job_4.save()
- return (hqe_1, hqe_2, hqe_3), monitor_db.AgentTask()
+ hqe_1 = job_1.hostqueueentry_set.all()[0]
+ hqe_2 = job_2.hostqueueentry_set.all()[0]
+ hqe_3 = job_3.hostqueueentry_set.all()[0]
+ hqe_4 = job_4.hostqueueentry_set.all()[0]
+ return (hqe_1, hqe_2, hqe_3, hqe_4), monitor_db.AgentTask()
- def test_get_drone_hostnames_allowed_no_associated_sets(self):
+
+ def test_get_drone_hostnames_allowed_no_drones_in_set(self):
hqes, task = self._setup_drones()
- task.queue_entry_ids = (hqes[2],)
+ task.queue_entry_ids = (hqes[2].id,)
self.assertEqual(set(), task.get_drone_hostnames_allowed())
self.god.check_playback()
+ def test_get_drone_hostnames_allowed_no_drone_set(self):
+ hqes, task = self._setup_drones()
+ hqe = hqes[3]
+ task.queue_entry_ids = (hqe.id,)
+
+ result = object()
+
+ self.god.stub_function(task, '_user_or_global_default_drone_set')
+ task._user_or_global_default_drone_set.expect_call(
+ hqe.job, hqe.job.user()).and_return(result)
+
+ self.assertEqual(result, task.get_drone_hostnames_allowed())
+ self.god.check_playback()
+
+
def test_get_drone_hostnames_allowed_success(self):
hqes, task = self._setup_drones()
- task.queue_entry_ids = (hqes[0],)
+ task.queue_entry_ids = (hqes[0].id,)
self.assertEqual(set(('0','1')), task.get_drone_hostnames_allowed())
self.god.check_playback()
def test_get_drone_hostnames_allowed_multiple_jobs(self):
hqes, task = self._setup_drones()
- task.queue_entry_ids = (hqes[0], hqes[1])
+ task.queue_entry_ids = (hqes[0].id, hqes[1].id)
self.assertRaises(AssertionError,
task.get_drone_hostnames_allowed)
self.god.check_playback()
+ def test_get_drone_hostnames_allowed_no_hqe(self):
+ class MockSpecialTask(object):
+ requested_by = object()
+
+ class MockSpecialAgentTask(monitor_db.SpecialAgentTask):
+ task = MockSpecialTask()
+ queue_entry_ids = []
+ def __init__(self, *args, **kwargs):
+ pass
+
+ task = MockSpecialAgentTask()
+ self.god.stub_function(models.DroneSet, 'drone_sets_enabled')
+ self.god.stub_function(task, '_user_or_global_default_drone_set')
+
+ result = object()
+ models.DroneSet.drone_sets_enabled.expect_call().and_return(True)
+ task._user_or_global_default_drone_set.expect_call(
+ task.task, MockSpecialTask.requested_by).and_return(result)
+
+ self.assertEqual(result, task.get_drone_hostnames_allowed())
+ self.god.check_playback()
+
+
+ def _setup_test_user_or_global_default_drone_set(self):
+ result = object()
+ class MockDroneSet(object):
+ def get_drone_hostnames(self):
+ return result
+
+ self.god.stub_function(models.DroneSet, 'get_default')
+ models.DroneSet.get_default.expect_call().and_return(MockDroneSet())
+ return result
+
+
+ def test_user_or_global_default_drone_set(self):
+ expected = object()
+ class MockDroneSet(object):
+ def get_drone_hostnames(self):
+ return expected
+ class MockUser(object):
+ drone_set = MockDroneSet()
+
+ self._setup_test_user_or_global_default_drone_set()
+
+ actual = monitor_db.AgentTask()._user_or_global_default_drone_set(
+ None, MockUser())
+
+ self.assertEqual(expected, actual)
+ self.god.check_playback()
+
+
+ def test_user_or_global_default_drone_set_no_user(self):
+ expected = self._setup_test_user_or_global_default_drone_set()
+ actual = monitor_db.AgentTask()._user_or_global_default_drone_set(
+ None, None)
+
+ self.assertEqual(expected, actual)
+ self.god.check_playback()
+
+
+ def test_user_or_global_default_drone_set_no_user_drone_set(self):
+ class MockUser(object):
+ drone_set = None
+ login = None
+
+ expected = self._setup_test_user_or_global_default_drone_set()
+ actual = monitor_db.AgentTask()._user_or_global_default_drone_set(
+ None, MockUser())
+
+ self.assertEqual(expected, actual)
+ self.god.check_playback()
+
+
if __name__ == '__main__':
unittest.main()
_______________________________________________
Autotest mailing list
[email protected]
http://test.kernel.org/cgi-bin/mailman/listinfo/autotest
