#!/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()