| #!/usr/bin/python |
| |
| import common |
| import unittest |
| |
| # This has to be done very early. |
| from autotest_lib.client.common_lib import global_config |
| global_config.global_config.override_config_value( |
| 'HOSTS', 'default_protection', |
| 'NO_PROTECTION') |
| from autotest_lib.client.common_lib import host_protections |
| |
| from autotest_lib.frontend import setup_django_environment |
| from autotest_lib.frontend import setup_test_environment |
| from django.test import client |
| from autotest_lib.frontend.shared import resource_test_utils |
| from autotest_lib.frontend.afe import control_file, models, model_attributes |
| |
| class AfeResourceTestCase(resource_test_utils.ResourceTestCase): |
| URI_PREFIX = 'http://testserver/afe/server/resources' |
| |
| CONTROL_FILE_CONTENTS = 'my control file contents' |
| |
| def setUp(self): |
| super(AfeResourceTestCase, self).setUp() |
| self._add_additional_data() |
| |
| |
| def _add_additional_data(self): |
| models.Test.objects.create(name='mytest', |
| test_type=model_attributes.TestTypes.SERVER, |
| path='/path/to/mytest') |
| |
| |
| class FilteringPagingTest(AfeResourceTestCase): |
| # we'll arbitarily choose to use hosts for this |
| |
| def setUp(self): |
| super(FilteringPagingTest, self).setUp() |
| |
| self.labels[0].host_set = [self.hosts[0], self.hosts[1]] |
| for host in self.hosts[:3]: |
| host.locked = True |
| host.save() |
| |
| def test_simple_filtering(self): |
| response = self.request('get', 'hosts?locked=true&has_label=label1') |
| self.check_collection(response, 'hostname', ['host1', 'host2']) |
| |
| |
| def test_in_filtering(self): |
| response = self.request('get', 'hosts?hostname:in=host1,host2') |
| self.check_collection(response, 'hostname', ['host1', 'host2']) |
| |
| |
| def test_paging(self): |
| response = self.request('get', 'hosts?start_index=1&items_per_page=2') |
| self.check_collection(response, 'hostname', ['host2', 'host3']) |
| self.assertEquals(response['total_results'], 9) |
| self.assertEquals(response['items_per_page'], 2) |
| self.assertEquals(response['start_index'], 1) |
| |
| |
| def test_full_representations(self): |
| response = self.request( |
| 'get', 'hosts?hostname=host1&full_representations=true') |
| self.check_collection(response, 'hostname', ['host1']) |
| host = response['members'][0] |
| # invalid only included in full representation |
| self.assertEquals(host['invalid'], False) |
| |
| |
| class MiscellaneousTest(AfeResourceTestCase): |
| def test_trailing_slash(self): |
| response = self.request('get', 'hosts/host1/') |
| self.assertEquals(response['hostname'], 'host1') |
| |
| |
| class AtomicGroupClassTest(AfeResourceTestCase): |
| def test_collection(self): |
| response = self.request('get', 'atomic_group_classes') |
| self.check_collection(response, 'name', ['atomic1', 'atomic2'], |
| length=2) |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'atomic_group_classes/atomic1') |
| self.assertEquals(response['name'], 'atomic1') |
| self.assertEquals(response['max_number_of_machines'], 2) |
| |
| |
| def test_labels(self): |
| self.check_relationship('atomic_group_classes/atomic1', 'labels', |
| 'label', 'name', ['label4', 'label5']) |
| |
| |
| class LabelTest(AfeResourceTestCase): |
| def test_collection(self): |
| response = self.request('get', 'labels') |
| self.check_collection(response, 'name', ['label1', 'label2'], length=9, |
| check_number=2) |
| label1 = self.sorted_by(response['members'], 'name')[0] |
| self.assertEquals(label1['is_platform'], False) |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'labels/label1') |
| self.assertEquals(response['name'], 'label1') |
| self.assertEquals(response['is_platform'], False) |
| self.assertEquals(response['atomic_group_class'], None) |
| |
| |
| def test_hosts(self): |
| self.check_relationship('labels/label1', 'hosts', 'host', 'hostname', |
| ['host1']) |
| |
| |
| class UserTest(AfeResourceTestCase): |
| def test_collection(self): |
| response = self.request('get', 'users') |
| self.check_collection(response, 'username', |
| ['autotest_system', 'debug_user']) |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'users/debug_user') |
| self.assertEquals(response['username'], 'debug_user') |
| |
| me_response = self.request('get', 'users/@me') |
| self.assertEquals(response, me_response) |
| |
| |
| def test_acls(self): |
| self.check_relationship('users/debug_user', 'acls', 'acl', 'name', |
| ['Everyone', 'my_acl']) |
| |
| |
| def test_accessible_hosts(self): |
| group = models.AclGroup.objects.create(name='mygroup') |
| models.User.objects.get(login='debug_user').aclgroup_set = [group] |
| self.hosts[0].aclgroup_set = [group] |
| |
| user = self.request('get', 'users/debug_user') |
| response = self.request('get', user['accessible_hosts']['href']) |
| self.check_collection(response, 'hostname', ['host1']) |
| |
| |
| class AclTest(AfeResourceTestCase): |
| def test_collection(self): |
| response = self.request('get', 'acls') |
| self.check_collection(response, 'name', ['Everyone', 'my_acl']) |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'acls/my_acl') |
| self.assertEquals(response['name'], 'my_acl') |
| |
| |
| def test_users(self): |
| self.check_relationship('acls/my_acl', 'users', 'user', 'username', |
| ['autotest_system', 'debug_user']) |
| |
| |
| def test_hosts(self): |
| self.check_relationship('acls/my_acl', 'hosts', 'host', 'hostname', |
| ['host1', 'host2'], length=9, check_number=2) |
| |
| |
| class HostTest(AfeResourceTestCase): |
| def test_collection(self): |
| response = self.request('get', 'hosts') |
| self.check_collection(response, 'hostname', ['host1', 'host2'], |
| length=9, check_number=2) |
| host1 = self.sorted_by(response['members'], 'hostname')[0] |
| self.assertEquals(host1['platform']['name'], 'myplatform') |
| self.assertEquals(host1['locked'], False) |
| self.assertEquals(host1['status'], 'Ready') |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'hosts/host1') |
| self.assertEquals(response['protection_level'], 'No protection') |
| |
| |
| def test_labels(self): |
| self.check_relationship('hosts/host1', 'labels', 'label', 'name', |
| ['label1', 'myplatform']) |
| |
| |
| def test_acls(self): |
| self.check_relationship('hosts/host1', 'acls', 'acl', 'name', |
| ['my_acl']) |
| |
| |
| def test_queue_entries(self): |
| self._create_job(hosts=[1]) |
| host = self.request('get', 'hosts/host1') |
| entries = self.request('get', host['queue_entries']['href']) |
| self.check_collection(entries, ['job', 'id'], [1]) |
| |
| |
| def test_health_tasks(self): |
| models.SpecialTask.schedule_special_task( |
| host=self.hosts[0], task=models.SpecialTask.Task.VERIFY) |
| host = self.request('get', 'hosts/host1') |
| tasks = self.request('get', host['health_tasks']['href']) |
| self.check_collection(tasks, 'task_type', ['Verify']) |
| |
| |
| def test_put(self): |
| response = self.request('put', 'hosts/host1', data={'locked': True}) |
| self.assertEquals(response['locked'], True) |
| response = self.request('get', 'hosts/host1') |
| self.assertEquals(response['locked'], True) |
| self.assertEquals(response['locked_by']['username'], 'debug_user') |
| |
| |
| def test_post(self): |
| data = {'hostname': 'newhost', |
| 'platform': {'href': self.URI_PREFIX + '/labels/myplatform'}, |
| 'protection_level': 'Do not verify'} |
| response = self.request('post', 'hosts', data=data) |
| self.assertEquals(response, self.URI_PREFIX + '/hosts/newhost') |
| |
| host = models.Host.objects.get(hostname='newhost') |
| self.assertEquals(host.platform().name, 'myplatform') |
| self.assertEquals(host.protection, models.Host.Protection.DO_NOT_VERIFY) |
| |
| |
| def _check_labels(self, host, expected_labels): |
| label_names = sorted(label.name for label in host.labels.all()) |
| self.assertEquals(label_names, sorted(expected_labels)) |
| |
| |
| def test_add_label(self): |
| labels_href = self.request('get', 'hosts/host1')['labels']['href'] |
| data = {'label': self.URI_PREFIX + '/labels/label2'} |
| response = self.request('post', labels_href, data=data) |
| self._check_labels(self.hosts[0], ['label1', 'label2', 'myplatform']) |
| |
| |
| def test_remove_label(self): |
| labels_href = self.request('get', 'hosts/host1')['labels']['href'] |
| labels_href += '&label=label1' |
| labelings = self.request('get', labels_href)['members'] |
| self.assertEquals(len(labelings), 1) |
| self.request('delete', labelings[0]['href']) |
| self._check_labels(self.hosts[0], ['myplatform']) |
| |
| |
| def test_delete(self): |
| self.request('delete', 'hosts/host1') |
| hosts = models.Host.valid_objects.filter(hostname='host1') |
| self.assertEquals(len(hosts), 0) |
| |
| |
| class TestTest(AfeResourceTestCase): # yes, we're testing the "tests" resource |
| def test_collection(self): |
| response = self.request('get', 'tests') |
| self.check_collection(response, 'name', ['mytest']) |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'tests/mytest') |
| self.assertEquals(response['name'], 'mytest') |
| self.assertEquals(response['control_file_type'], 'Server') |
| self.assertEquals(response['control_file_path'], '/path/to/mytest') |
| |
| |
| def test_dependencies(self): |
| models.Test.objects.get(name='mytest').dependency_labels = [self.label3] |
| self.check_relationship('tests/mytest', 'dependencies', 'label', 'name', |
| ['label3']) |
| |
| |
| class ExecutionInfoTest(AfeResourceTestCase): |
| def setUp(self): |
| super(ExecutionInfoTest, self).setUp() |
| |
| def mock_read_control_file(test): |
| return self.CONTROL_FILE_CONTENTS |
| self.god.stub_with(control_file, 'read_control_file', |
| mock_read_control_file) |
| def test_get(self): |
| response = self.request('get', 'execution_info?tests=mytest') |
| info = response['execution_info'] |
| self.assert_(self.CONTROL_FILE_CONTENTS in info['control_file']) |
| self.assertEquals(info['is_server'], True) |
| self.assertEquals(info['machines_per_execution'], 1) |
| |
| |
| class QueueEntriesRequestTest(AfeResourceTestCase): |
| def test_get(self): |
| response = self.request( |
| 'get', |
| 'queue_entries_request?hosts=host1,host2&meta_hosts=label1') |
| |
| # choose an arbitrary but consistent ordering to ease checking |
| def entry_href(entry): |
| if 'host' in entry: |
| return entry['host']['href'] |
| return entry['meta_host']['href'] |
| entries = sorted(response['queue_entries'], key=entry_href) |
| |
| expected = [ |
| {'host': {'href': self.URI_PREFIX + '/hosts/host1'}}, |
| {'host': {'href': self.URI_PREFIX + '/hosts/host2'}}, |
| {'meta_host': {'href': self.URI_PREFIX + '/labels/label1'}}] |
| self.assertEquals(entries, expected) |
| |
| |
| class JobTest(AfeResourceTestCase): |
| def setUp(self): |
| super(JobTest, self).setUp() |
| |
| for _ in xrange(2): |
| self._create_job(hosts=[1, 2]) |
| |
| job = models.Job.objects.get(id=1) |
| job.control_file = self.CONTROL_FILE_CONTENTS |
| job.save() |
| |
| models.JobKeyval.objects.create(job=job, key='mykey', value='myvalue') |
| |
| |
| def test_collection(self): |
| response = self.request('get', 'jobs') |
| self.check_collection(response, 'id', [1, 2]) |
| |
| |
| def test_keyval_filtering(self): |
| response = self.request('get', 'jobs?has_keyval=mykey=myvalue') |
| self.check_collection(response, 'id', [1]) |
| |
| |
| def test_entry(self): |
| response = self.request('get', 'jobs/1') |
| self.assertEquals(response['id'], 1) |
| self.assertEquals(response['name'], 'test') |
| self.assertEquals(response['keyvals'], {'mykey': 'myvalue'}) |
| info = response['execution_info'] |
| self.assertEquals(info['control_file'], self.CONTROL_FILE_CONTENTS) |
| self.assertEquals(info['is_server'], False) |
| self.assertEquals(info['cleanup_before_job'], 'Never') |
| self.assertEquals(info['cleanup_after_job'], 'Always') |
| self.assertEquals(info['machines_per_execution'], 1) |
| self.assertEquals(info['run_verify'], True) |
| |
| |
| def test_queue_entries(self): |
| job = self.request('get', 'jobs/1') |
| entries = self.request('get', job['queue_entries']['href']) |
| self.check_collection(entries, ['host', 'hostname'], ['host1', 'host2']) |
| |
| |
| def _test_post_helper(self, owner): |
| data = {'name': 'myjob', |
| 'execution_info': {'control_file': self.CONTROL_FILE_CONTENTS, |
| 'is_server': True}, |
| 'owner': owner, |
| 'drone_set': models.DroneSet.default_drone_set_name(), |
| 'queue_entries': |
| [{'host': {'href': self.URI_PREFIX + '/hosts/host1'}}, |
| {'host': {'href': self.URI_PREFIX + '/hosts/host2'}}]} |
| response = self.request('post', 'jobs', data=data) |
| self.assertEquals(response, self.URI_PREFIX + '/jobs/3') |
| job = models.Job.objects.get(id=3) |
| self.assertEquals(job.name, 'myjob') |
| self.assertEquals(job.control_file, self.CONTROL_FILE_CONTENTS) |
| self.assertEquals(job.control_type, models.Job.ControlType.SERVER) |
| entries = job.hostqueueentry_set.order_by('host__hostname') |
| self.assertEquals(entries[0].host.hostname, 'host1') |
| self.assertEquals(entries[1].host.hostname, 'host2') |
| |
| owner_test = owner |
| if not owner_test: |
| owner_test = models.User.current_user().login |
| self.assertEquals(job.owner, owner_test) |
| |
| |
| def test_post_no_owner(self): |
| self._test_post_helper(None) |
| |
| |
| def test_post_with_owner(self): |
| self._test_post_helper('job_owner') |
| |
| |
| class DirectoryTest(AfeResourceTestCase): |
| def test_get(self): |
| response = self.request('get', '') |
| for key in ('atomic_group_classes', 'labels', 'users', 'acl_groups', |
| 'hosts', 'tests', 'jobs', 'execution_info', |
| 'queue_entries_request'): |
| self.assert_(key in response) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
