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