xref: /aosp_15_r20/external/autotest/frontend/afe/models_test.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

#!/usr/bin/python3
# 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': 'stub'},
                                            {'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': 'stub'},
                                            {'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()