| #!/usr/bin/python |
| # pylint: disable=missing-docstring |
| |
| import unittest |
| |
| import common |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.frontend import setup_django_environment |
| from autotest_lib.frontend.afe import frontend_test_utils |
| from autotest_lib.frontend.afe import models, model_logic |
| |
| |
| class AclGroupTest(unittest.TestCase, |
| frontend_test_utils.FrontendTestMixin): |
| def setUp(self): |
| self._frontend_common_setup() |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def _check_acls(self, host, acl_name_list): |
| actual_acl_names = [acl_group.name for acl_group |
| in host.aclgroup_set.all()] |
| self.assertEquals(set(actual_acl_names), set(acl_name_list)) |
| |
| |
| def test_on_host_membership_change(self): |
| host1, host2 = self.hosts[1:3] |
| everyone_acl = models.AclGroup.objects.get(name='Everyone') |
| |
| host1.aclgroup_set.clear() |
| self._check_acls(host1, []) |
| host2.aclgroup_set.add(everyone_acl) |
| self._check_acls(host2, ['Everyone', 'my_acl']) |
| |
| models.AclGroup.on_host_membership_change() |
| |
| self._check_acls(host1, ['Everyone']) |
| self._check_acls(host2, ['my_acl']) |
| |
| |
| class HostTest(unittest.TestCase, |
| frontend_test_utils.FrontendTestMixin): |
| def setUp(self): |
| self._frontend_common_setup() |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def _get_attributes(self, host): |
| models.Host.objects.populate_relationships( |
| [host], models.HostAttribute, 'attribute_list') |
| return dict((attribute.attribute, attribute.value) |
| for attribute in host.attribute_list) |
| |
| def test_delete_attribute(self): |
| previous_config = models.RESPECT_STATIC_ATTRIBUTES |
| models.RESPECT_STATIC_ATTRIBUTES = False |
| host1 = models.Host.objects.create(hostname='test_host1') |
| host1.set_attribute('test_attribute1', 'test_value1') |
| |
| attributes = self._get_attributes(host1) |
| self.assertEquals(attributes['test_attribute1'], 'test_value1') |
| |
| host1.set_or_delete_attribute('test_attribute1', None) |
| attributes = self._get_attributes(host1) |
| self.assertNotIn('test_attribute1', attributes.keys()) |
| |
| models.RESPECT_STATIC_ATTRIBUTES = previous_config |
| |
| |
| def test_delete_static_attribute(self): |
| previous_config = models.RESPECT_STATIC_ATTRIBUTES |
| models.RESPECT_STATIC_ATTRIBUTES = True |
| host1 = models.Host.objects.create(hostname='test_host1') |
| host1.set_attribute('test_attribute1', 'test_value1') |
| self._set_static_attribute(host1, 'test_attribute1', 'test_value1') |
| |
| self.assertRaises( |
| error.UnmodifiableAttributeException, |
| host1.set_or_delete_attribute, |
| 'test_attribute1', None) |
| |
| models.RESPECT_STATIC_ATTRIBUTES = previous_config |
| |
| |
| def test_set_attribute(self): |
| previous_config = models.RESPECT_STATIC_ATTRIBUTES |
| models.RESPECT_STATIC_ATTRIBUTES = False |
| host1 = models.Host.objects.create(hostname='test_host1') |
| host1.set_attribute('test_attribute1', 'test_value1') |
| |
| host1.set_or_delete_attribute('test_attribute1', 'test_new_value1') |
| |
| attributes = self._get_attributes(host1) |
| self.assertEquals(attributes['test_attribute1'], 'test_new_value1') |
| |
| models.RESPECT_STATIC_ATTRIBUTES = previous_config |
| |
| |
| def test_set_static_attribute(self): |
| previous_config = models.RESPECT_STATIC_ATTRIBUTES |
| models.RESPECT_STATIC_ATTRIBUTES = True |
| host1 = models.Host.objects.create(hostname='test_host1') |
| host1.set_attribute('test_attribute1', 'test_value1') |
| self._set_static_attribute(host1, 'test_attribute1', 'test_value1') |
| |
| self.assertRaises( |
| error.UnmodifiableAttributeException, |
| host1.set_or_delete_attribute, |
| 'test_attribute1', 'test_value2') |
| |
| models.RESPECT_STATIC_ATTRIBUTES = previous_config |
| |
| |
| def test_add_host_previous_one_time_host(self): |
| # ensure that when adding a host which was previously used as a one-time |
| # host, the status isn't reset, since this can interfere with the |
| # scheduler. |
| host = models.Host.create_one_time_host('othost') |
| self.assertEquals(host.invalid, True) |
| self.assertEquals(host.status, models.Host.Status.READY) |
| |
| host.status = models.Host.Status.RUNNING |
| host.save() |
| |
| host2 = models.Host.add_object(hostname='othost') |
| self.assertEquals(host2.id, host.id) |
| self.assertEquals(host2.status, models.Host.Status.RUNNING) |
| |
| |
| def test_check_board_labels_allowed(self): |
| host = models.Host.create_one_time_host('othost') |
| # First check with host with no board label. |
| self.assertEqual(host.check_board_labels_allowed([host]), None) |
| |
| # Second check with host with board label |
| label = models.Label.add_object(name='board:test') |
| label.host_set.add(host) |
| self.assertRaises(model_logic.ValidationError, |
| host.check_board_labels_allowed, [host], |
| ['board:new_board']) |
| |
| |
| class SpecialTaskUnittest(unittest.TestCase, |
| frontend_test_utils.FrontendTestMixin): |
| def setUp(self): |
| self._frontend_common_setup() |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def _create_task(self): |
| return models.SpecialTask.objects.create( |
| host=self.hosts[0], task=models.SpecialTask.Task.VERIFY, |
| requested_by=models.User.current_user()) |
| |
| |
| def test_execution_path(self): |
| task = self._create_task() |
| self.assertEquals(task.execution_path(), 'hosts/host1/1-verify') |
| |
| |
| def test_status(self): |
| task = self._create_task() |
| self.assertEquals(task.status, 'Queued') |
| |
| task.update_object(is_active=True) |
| self.assertEquals(task.status, 'Running') |
| |
| task.update_object(is_active=False, is_complete=True, success=True) |
| self.assertEquals(task.status, 'Completed') |
| |
| task.update_object(success=False) |
| self.assertEquals(task.status, 'Failed') |
| |
| |
| def test_activate(self): |
| task = self._create_task() |
| task.activate() |
| self.assertTrue(task.is_active) |
| self.assertFalse(task.is_complete) |
| |
| |
| def test_finish(self): |
| task = self._create_task() |
| task.activate() |
| task.finish(True) |
| self.assertFalse(task.is_active) |
| self.assertTrue(task.is_complete) |
| self.assertTrue(task.success) |
| |
| |
| def test_requested_by_from_queue_entry(self): |
| job = self._create_job(hosts=[0]) |
| task = models.SpecialTask.objects.create( |
| host=self.hosts[0], task=models.SpecialTask.Task.VERIFY, |
| queue_entry=job.hostqueueentry_set.all()[0]) |
| self.assertEquals(task.requested_by.login, 'autotest_system') |
| |
| |
| class HostQueueEntryUnittest(unittest.TestCase, |
| frontend_test_utils.FrontendTestMixin): |
| def setUp(self): |
| self._frontend_common_setup() |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def test_execution_path(self): |
| entry = self._create_job(hosts=[1]).hostqueueentry_set.all()[0] |
| entry.execution_subdir = 'subdir' |
| entry.save() |
| |
| self.assertEquals(entry.execution_path(), '1-autotest_system/subdir') |
| |
| |
| class ModelWithInvalidTest(unittest.TestCase, |
| frontend_test_utils.FrontendTestMixin): |
| def setUp(self): |
| self._frontend_common_setup() |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def test_model_with_invalid_delete(self): |
| self.assertFalse(self.hosts[0].invalid) |
| self.hosts[0].delete() |
| self.assertTrue(self.hosts[0].invalid) |
| self.assertTrue(models.Host.objects.get(id=self.hosts[0].id)) |
| |
| |
| def test_model_with_invalid_delete_queryset(self): |
| for host in self.hosts: |
| self.assertFalse(host.invalid) |
| |
| hosts = models.Host.objects.all() |
| hosts.delete() |
| self.assertEqual(hosts.count(), len(self.hosts)) |
| |
| for host in hosts: |
| self.assertTrue(host.invalid) |
| |
| |
| def test_cloned_queryset_delete(self): |
| """ |
| Make sure that a cloned queryset maintains the custom delete() |
| """ |
| to_delete = ('host1', 'host2') |
| |
| for host in self.hosts: |
| self.assertFalse(host.invalid) |
| |
| hosts = models.Host.objects.all().filter(hostname__in=to_delete) |
| hosts.delete() |
| all_hosts = models.Host.objects.all() |
| self.assertEqual(all_hosts.count(), len(self.hosts)) |
| |
| for host in all_hosts: |
| if host.hostname in to_delete: |
| self.assertTrue( |
| host.invalid, |
| '%s.invalid expected to be True' % host.hostname) |
| else: |
| self.assertFalse( |
| host.invalid, |
| '%s.invalid expected to be False' % host.hostname) |
| |
| |
| def test_normal_delete(self): |
| job = self._create_job(hosts=[1]) |
| self.assertEqual(1, models.Job.objects.all().count()) |
| |
| job.delete() |
| self.assertEqual(0, models.Job.objects.all().count()) |
| |
| |
| def test_normal_delete_queryset(self): |
| self._create_job(hosts=[1]) |
| self._create_job(hosts=[2]) |
| |
| self.assertEqual(2, models.Job.objects.all().count()) |
| |
| models.Job.objects.all().delete() |
| self.assertEqual(0, models.Job.objects.all().count()) |
| |
| |
| class SerializationTest(unittest.TestCase, |
| frontend_test_utils.FrontendTestMixin): |
| def setUp(self): |
| self._frontend_common_setup(fill_data=False) |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def _get_example_response(self): |
| return {'hosts': [{'aclgroup_set': [{'description': '', |
| 'id': 1, |
| 'name': 'Everyone', |
| 'users': [{ |
| 'access_level': 100, |
| 'id': 1, |
| 'login': 'autotest_system', |
| 'reboot_after': 0, |
| 'reboot_before': 1, |
| 'show_experimental': False}]}], |
| 'dirty': True, |
| 'hostattribute_set': [], |
| 'hostname': '100.107.2.163', |
| 'id': 2, |
| 'invalid': False, |
| 'labels': [{'id': 7, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'power:battery', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 9, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'hw_video_acc_h264', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 10, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'hw_video_acc_enc_h264', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 11, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'webcam', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 12, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'touchpad', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 13, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'spring', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 14, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'board:daisy', |
| 'only_if_needed': False, |
| 'platform': True}, |
| {'id': 15, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'board_freq_mem:daisy_1.7GHz', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 16, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'bluetooth', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 17, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'gpu_family:mali', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 19, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'ec:cros', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 20, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'storage:mmc', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 21, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'hw_video_acc_vp8', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 22, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'video_glitch_detection', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 23, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'pool:suites', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 25, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'daisy-board-name', |
| 'only_if_needed': False, |
| 'platform': False}], |
| 'leased': False, |
| 'lock_reason': '', |
| 'lock_time': None, |
| 'locked': False, |
| 'protection': 0, |
| 'shard': {'hostname': '1', 'id': 1}, |
| 'status': 'Ready', |
| 'synch_id': None}], |
| 'jobs': [{'control_file': 'some control file\n\n\n', |
| 'control_type': 2, |
| 'created_on': '2014-09-04T13:09:35', |
| 'dependency_labels': [{'id': 14, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'board:daisy', |
| 'only_if_needed': False, |
| 'platform': True}, |
| {'id': 23, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'pool:suites', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 25, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'daisy-board-name', |
| 'only_if_needed': False, |
| 'platform': False}], |
| 'email_list': '', |
| 'hostqueueentry_set': [{'aborted': False, |
| 'active': False, |
| 'complete': False, |
| 'deleted': False, |
| 'execution_subdir': '', |
| 'finished_on': None, |
| 'id': 5, |
| 'meta_host': { |
| 'id': 14, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'board:daisy', |
| 'only_if_needed': False, |
| 'platform': True}, |
| 'host_id': None, |
| 'started_on': None, |
| 'status': 'Queued'}], |
| 'id': 5, |
| 'jobkeyval_set': [{'id': 10, |
| 'job_id': 5, |
| 'key': 'suite', |
| 'value': 'dummy'}, |
| {'id': 11, |
| 'job_id': 5, |
| 'key': 'build', |
| 'value': 'daisy-release'}, |
| {'id': 12, |
| 'job_id': 5, |
| 'key': 'experimental', |
| 'value': 'False'}], |
| 'max_runtime_hrs': 72, |
| 'max_runtime_mins': 1440, |
| 'name': 'daisy-experimental', |
| 'owner': 'autotest', |
| 'parse_failed_repair': True, |
| 'priority': 40, |
| 'reboot_after': 0, |
| 'reboot_before': 1, |
| 'run_reset': True, |
| 'run_verify': False, |
| 'shard': {'hostname': '1', 'id': 1}, |
| 'synch_count': 1, |
| 'test_retry': 0, |
| 'timeout': 24, |
| 'timeout_mins': 1440, |
| 'require_ssp': None}, |
| {'control_file': 'some control file\n\n\n', |
| 'control_type': 2, |
| 'created_on': '2014-09-04T13:09:35', |
| 'dependency_labels': [{'id': 14, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'board:daisy', |
| 'only_if_needed': False, |
| 'platform': True}, |
| {'id': 23, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'pool:suites', |
| 'only_if_needed': False, |
| 'platform': False}, |
| {'id': 25, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'daisy-board-name', |
| 'only_if_needed': False, |
| 'platform': False}], |
| 'email_list': '', |
| 'hostqueueentry_set': [{'aborted': False, |
| 'active': False, |
| 'complete': False, |
| 'deleted': False, |
| 'execution_subdir': '', |
| 'finished_on': None, |
| 'id': 7, |
| 'meta_host': { |
| 'id': 14, |
| 'invalid': False, |
| 'kernel_config': '', |
| 'name': 'board:daisy', |
| 'only_if_needed': False, |
| 'platform': True}, |
| 'host_id': None, |
| 'started_on': None, |
| 'status': 'Queued'}], |
| 'id': 7, |
| 'jobkeyval_set': [{'id': 16, |
| 'job_id': 7, |
| 'key': 'suite', |
| 'value': 'dummy'}, |
| {'id': 17, |
| 'job_id': 7, |
| 'key': 'build', |
| 'value': 'daisy-release'}, |
| {'id': 18, |
| 'job_id': 7, |
| 'key': 'experimental', |
| 'value': 'False'}], |
| 'max_runtime_hrs': 72, |
| 'max_runtime_mins': 1440, |
| 'name': 'daisy-experimental', |
| 'owner': 'autotest', |
| 'parse_failed_repair': True, |
| 'priority': 40, |
| 'reboot_after': 0, |
| 'reboot_before': 1, |
| 'run_reset': True, |
| 'run_verify': False, |
| 'shard': {'hostname': '1', 'id': 1}, |
| 'synch_count': 1, |
| 'test_retry': 0, |
| 'timeout': 24, |
| 'timeout_mins': 1440, |
| 'require_ssp': None}]} |
| |
| |
| def test_response(self): |
| heartbeat_response = self._get_example_response() |
| hosts_serialized = heartbeat_response['hosts'] |
| jobs_serialized = heartbeat_response['jobs'] |
| |
| # Persisting is automatically done inside deserialize |
| hosts = [models.Host.deserialize(host) for host in hosts_serialized] |
| jobs = [models.Job.deserialize(job) for job in jobs_serialized] |
| |
| generated_heartbeat_response = { |
| 'hosts': [host.serialize() for host in hosts], |
| 'jobs': [job.serialize() for job in jobs] |
| } |
| example_response = self._get_example_response() |
| # For attribute-like objects, we don't care about its id. |
| for r in [generated_heartbeat_response, example_response]: |
| for job in r['jobs']: |
| for keyval in job['jobkeyval_set']: |
| keyval.pop('id') |
| for host in r['hosts']: |
| for attribute in host['hostattribute_set']: |
| keyval.pop('id') |
| self.assertEqual(generated_heartbeat_response, example_response) |
| |
| |
| def test_update(self): |
| job = self._create_job(hosts=[1]) |
| serialized = job.serialize(include_dependencies=False) |
| serialized['owner'] = 'some_other_owner' |
| |
| job.update_from_serialized(serialized) |
| self.assertEqual(job.owner, 'some_other_owner') |
| |
| serialized = job.serialize() |
| self.assertRaises( |
| ValueError, |
| job.update_from_serialized, serialized) |
| |
| |
| def test_sync_aborted(self): |
| job = self._create_job(hosts=[1]) |
| serialized = job.serialize() |
| |
| serialized['hostqueueentry_set'][0]['aborted'] = True |
| serialized['hostqueueentry_set'][0]['status'] = 'Running' |
| |
| models.Job.deserialize(serialized) |
| |
| job = models.Job.objects.get(pk=job.id) |
| self.assertTrue(job.hostqueueentry_set.all()[0].aborted) |
| self.assertEqual(job.hostqueueentry_set.all()[0].status, 'Queued') |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
