1#!/usr/bin/python
2
3import common
4import unittest
5
6# This has to be done very early.
7from autotest_lib.client.common_lib import global_config
8global_config.global_config.override_config_value(
9    'HOSTS', 'default_protection',
10    'NO_PROTECTION')
11from autotest_lib.client.common_lib import host_protections
12
13from autotest_lib.frontend import setup_django_environment
14from autotest_lib.frontend import setup_test_environment
15from django.test import client
16from autotest_lib.frontend.shared import resource_test_utils
17from autotest_lib.frontend.afe import control_file, models, model_attributes
18from autotest_lib.client.common_lib import control_data
19
20class AfeResourceTestCase(resource_test_utils.ResourceTestCase):
21    URI_PREFIX = 'http://testserver/afe/server/resources'
22
23    CONTROL_FILE_CONTENTS = 'my control file contents'
24
25    def setUp(self):
26        super(AfeResourceTestCase, self).setUp()
27        self._add_additional_data()
28
29
30    def _add_additional_data(self):
31        models.Test.objects.create(name='mytest',
32                                   test_type=control_data.CONTROL_TYPE.SERVER,
33                                   path='/path/to/mytest')
34
35
36class FilteringPagingTest(AfeResourceTestCase):
37    # we'll arbitarily choose to use hosts for this
38
39    def setUp(self):
40        super(FilteringPagingTest, self).setUp()
41
42        self.labels[0].host_set = [self.hosts[0], self.hosts[1]]
43        for host in self.hosts[:3]:
44            host.locked = True
45            host.save()
46
47    def test_simple_filtering(self):
48        response = self.request('get', 'hosts?locked=true&has_label=label1')
49        self.check_collection(response, 'hostname', ['host1', 'host2'])
50
51
52    def test_in_filtering(self):
53        response = self.request('get', 'hosts?hostname:in=host1,host2')
54        self.check_collection(response, 'hostname', ['host1', 'host2'])
55
56
57    def test_paging(self):
58        response = self.request('get', 'hosts?start_index=1&items_per_page=2')
59        self.check_collection(response, 'hostname', ['host2', 'host3'])
60        self.assertEquals(response['total_results'], 9)
61        self.assertEquals(response['items_per_page'], 2)
62        self.assertEquals(response['start_index'], 1)
63
64
65    def test_full_representations(self):
66        response = self.request(
67                'get', 'hosts?hostname=host1&full_representations=true')
68        self.check_collection(response, 'hostname', ['host1'])
69        host = response['members'][0]
70        # invalid only included in full representation
71        self.assertEquals(host['invalid'], False)
72
73
74class MiscellaneousTest(AfeResourceTestCase):
75    def test_trailing_slash(self):
76        response = self.request('get', 'hosts/host1/')
77        self.assertEquals(response['hostname'], 'host1')
78
79
80class AtomicGroupClassTest(AfeResourceTestCase):
81    def test_collection(self):
82        response = self.request('get', 'atomic_group_classes')
83        self.check_collection(response, 'name', ['atomic1', 'atomic2'],
84                              length=2)
85
86
87    def test_entry(self):
88        response = self.request('get', 'atomic_group_classes/atomic1')
89        self.assertEquals(response['name'], 'atomic1')
90        self.assertEquals(response['max_number_of_machines'], 2)
91
92
93    def test_labels(self):
94        self.check_relationship('atomic_group_classes/atomic1', 'labels',
95                                'label', 'name', ['label4', 'label5'])
96
97
98class LabelTest(AfeResourceTestCase):
99    def test_collection(self):
100        response = self.request('get', 'labels')
101        self.check_collection(response, 'name', ['label1', 'label2'], length=10,
102                              check_number=2)
103        label1 = self.sorted_by(response['members'], 'name')[0]
104        self.assertEquals(label1['is_platform'], False)
105
106
107    def test_entry(self):
108        response = self.request('get', 'labels/label1')
109        self.assertEquals(response['name'], 'label1')
110        self.assertEquals(response['is_platform'], False)
111        self.assertEquals(response['atomic_group_class'], None)
112
113
114    def test_hosts(self):
115        self.check_relationship('labels/label1', 'hosts', 'host', 'hostname',
116                                ['host1'])
117
118
119class UserTest(AfeResourceTestCase):
120    def test_collection(self):
121        response = self.request('get', 'users')
122        self.check_collection(response, 'username',
123                              ['autotest_system', 'debug_user'])
124
125
126    def test_entry(self):
127        response = self.request('get', 'users/debug_user')
128        self.assertEquals(response['username'], 'debug_user')
129
130        me_response = self.request('get', 'users/@me')
131        self.assertEquals(response, me_response)
132
133
134    def test_acls(self):
135        self.check_relationship('users/debug_user', 'acls', 'acl', 'name',
136                                ['Everyone', 'my_acl'])
137
138
139    def test_accessible_hosts(self):
140        group = models.AclGroup.objects.create(name='mygroup')
141        models.User.objects.get(login='debug_user').aclgroup_set = [group]
142        self.hosts[0].aclgroup_set = [group]
143
144        user = self.request('get', 'users/debug_user')
145        response = self.request('get', user['accessible_hosts']['href'])
146        self.check_collection(response, 'hostname', ['host1'])
147
148
149class AclTest(AfeResourceTestCase):
150    def test_collection(self):
151        response = self.request('get', 'acls')
152        self.check_collection(response, 'name', ['Everyone', 'my_acl'])
153
154
155    def test_entry(self):
156        response = self.request('get', 'acls/my_acl')
157        self.assertEquals(response['name'], 'my_acl')
158
159
160    def test_users(self):
161        self.check_relationship('acls/my_acl', 'users', 'user', 'username',
162                                ['autotest_system', 'debug_user'])
163
164
165    def test_hosts(self):
166        self.check_relationship('acls/my_acl', 'hosts', 'host', 'hostname',
167                                ['host1', 'host2'], length=9, check_number=2)
168
169
170class HostTest(AfeResourceTestCase):
171    def test_collection(self):
172        response = self.request('get', 'hosts')
173        self.check_collection(response, 'hostname', ['host1', 'host2'],
174                              length=9, check_number=2)
175        host1 = self.sorted_by(response['members'], 'hostname')[0]
176        self.assertEquals(host1['platform']['name'], 'myplatform')
177        self.assertEquals(host1['locked'], False)
178        self.assertEquals(host1['status'], 'Ready')
179
180
181    def test_entry(self):
182        response = self.request('get', 'hosts/host1')
183        self.assertEquals(response['protection_level'], 'No protection')
184
185
186    def test_labels(self):
187        self.check_relationship('hosts/host1', 'labels', 'label', 'name',
188                                ['label1', 'myplatform'])
189
190
191    def test_acls(self):
192        self.check_relationship('hosts/host1', 'acls', 'acl', 'name',
193                                ['my_acl'])
194
195
196    def test_queue_entries(self):
197        self._create_job(hosts=[1])
198        host = self.request('get', 'hosts/host1')
199        entries = self.request('get', host['queue_entries']['href'])
200        self.check_collection(entries, ['job', 'id'], [1])
201
202
203    def test_health_tasks(self):
204        models.SpecialTask.schedule_special_task(
205                host=self.hosts[0], task=models.SpecialTask.Task.VERIFY)
206        host = self.request('get', 'hosts/host1')
207        tasks = self.request('get', host['health_tasks']['href'])
208        self.check_collection(tasks, 'task_type', ['Verify'])
209
210
211    def test_put(self):
212        response = self.request('put', 'hosts/host1', data={'locked': True})
213        self.assertEquals(response['locked'], True)
214        response = self.request('get', 'hosts/host1')
215        self.assertEquals(response['locked'], True)
216        self.assertEquals(response['locked_by']['username'], 'debug_user')
217
218
219    def test_post(self):
220        data = {'hostname': 'newhost',
221                'platform': {'href': self.URI_PREFIX + '/labels/myplatform'},
222                'protection_level': 'Do not verify'}
223        response = self.request('post', 'hosts', data=data)
224        self.assertEquals(response, self.URI_PREFIX + '/hosts/newhost')
225
226        host = models.Host.objects.get(hostname='newhost')
227        self.assertEquals(host.platform().name, 'myplatform')
228        self.assertEquals(host.protection, models.Host.Protection.DO_NOT_VERIFY)
229
230
231    def _check_labels(self, host, expected_labels):
232        label_names = sorted(label.name for label in host.labels.all())
233        self.assertEquals(label_names, sorted(expected_labels))
234
235
236    def test_add_label(self):
237        labels_href = self.request('get', 'hosts/host1')['labels']['href']
238        data = {'label': self.URI_PREFIX + '/labels/label2'}
239        response = self.request('post', labels_href, data=data)
240        self._check_labels(self.hosts[0], ['label1', 'label2', 'myplatform'])
241
242
243    def test_remove_label(self):
244        labels_href = self.request('get', 'hosts/host1')['labels']['href']
245        labels_href += '&label=label1'
246        labelings = self.request('get', labels_href)['members']
247        self.assertEquals(len(labelings), 1)
248        self.request('delete', labelings[0]['href'])
249        self._check_labels(self.hosts[0], ['myplatform'])
250
251
252    def test_delete(self):
253        self.request('delete', 'hosts/host1')
254        hosts = models.Host.valid_objects.filter(hostname='host1')
255        self.assertEquals(len(hosts), 0)
256
257
258class TestTest(AfeResourceTestCase): # yes, we're testing the "tests" resource
259    def test_collection(self):
260        response = self.request('get', 'tests')
261        self.check_collection(response, 'name', ['mytest'])
262
263
264    def test_entry(self):
265        response = self.request('get', 'tests/mytest')
266        self.assertEquals(response['name'], 'mytest')
267        self.assertEquals(response['control_file_type'], 'Server')
268        self.assertEquals(response['control_file_path'], '/path/to/mytest')
269
270
271    def test_dependencies(self):
272        models.Test.objects.get(name='mytest').dependency_labels = [self.label3]
273        self.check_relationship('tests/mytest', 'dependencies', 'label', 'name',
274                                ['label3'])
275
276
277class ExecutionInfoTest(AfeResourceTestCase):
278    def setUp(self):
279        super(ExecutionInfoTest, self).setUp()
280
281        def mock_read_control_file(test):
282            return self.CONTROL_FILE_CONTENTS
283        self.god.stub_with(control_file, 'read_control_file',
284                           mock_read_control_file)
285    def test_get(self):
286        response = self.request('get', 'execution_info?tests=mytest')
287        info = response['execution_info']
288        self.assert_(self.CONTROL_FILE_CONTENTS in info['control_file'])
289        self.assertEquals(info['is_server'], True)
290        self.assertEquals(info['machines_per_execution'], 1)
291
292
293class QueueEntriesRequestTest(AfeResourceTestCase):
294    def test_get(self):
295        response = self.request(
296                'get',
297                'queue_entries_request?hosts=host1,host2&meta_hosts=label1')
298
299        # choose an arbitrary but consistent ordering to ease checking
300        def entry_href(entry):
301            if 'host' in entry:
302                return entry['host']['href']
303            return entry['meta_host']['href']
304        entries = sorted(response['queue_entries'], key=entry_href)
305
306        expected = [
307                {'host': {'href': self.URI_PREFIX + '/hosts/host1'}},
308                {'host': {'href': self.URI_PREFIX + '/hosts/host2'}},
309                {'meta_host': {'href': self.URI_PREFIX + '/labels/label1'}}]
310        self.assertEquals(entries, expected)
311
312
313class JobTest(AfeResourceTestCase):
314    def setUp(self):
315        super(JobTest, self).setUp()
316
317        for _ in xrange(2):
318            self._create_job(hosts=[1, 2])
319
320        job = models.Job.objects.get(id=1)
321        job.control_file = self.CONTROL_FILE_CONTENTS
322        job.save()
323
324        models.JobKeyval.objects.create(job=job, key='mykey', value='myvalue')
325
326
327    def test_collection(self):
328        response = self.request('get', 'jobs')
329        self.check_collection(response, 'id', [1, 2])
330
331
332#    def test_keyval_filtering(self):
333#        response = self.request('get', 'jobs?has_keyval=mykey=myvalue')
334#        self.check_collection(response, 'id', [1])
335
336
337    def test_entry(self):
338        response = self.request('get', 'jobs/1')
339        self.assertEquals(response['id'], 1)
340        self.assertEquals(response['name'], 'test')
341        self.assertEquals(response['keyvals'], {'mykey': 'myvalue'})
342        info = response['execution_info']
343        self.assertEquals(info['control_file'], self.CONTROL_FILE_CONTENTS)
344        self.assertEquals(info['is_server'], False)
345        self.assertEquals(info['cleanup_before_job'], 'Never')
346        self.assertEquals(info['cleanup_after_job'], 'Never')
347        self.assertEquals(info['machines_per_execution'], 1)
348        self.assertEquals(info['run_verify'], False)
349        self.assertEquals(info['run_reset'], True)
350
351
352    def test_queue_entries(self):
353        job = self.request('get', 'jobs/1')
354        entries = self.request('get', job['queue_entries']['href'])
355        self.check_collection(entries, ['host', 'hostname'], ['host1', 'host2'])
356
357
358    def _test_post_helper(self, owner):
359        data = {'name': 'myjob',
360                'execution_info': {'control_file': self.CONTROL_FILE_CONTENTS,
361                                   'is_server': True},
362                'owner': owner,
363                'drone_set': models.DroneSet.default_drone_set_name(),
364                'queue_entries':
365                [{'host': {'href': self.URI_PREFIX + '/hosts/host1'}},
366                 {'host': {'href': self.URI_PREFIX + '/hosts/host2'}}]}
367        response = self.request('post', 'jobs', data=data)
368        self.assertEquals(response, self.URI_PREFIX + '/jobs/3')
369        job = models.Job.objects.get(id=3)
370        self.assertEquals(job.name, 'myjob')
371        self.assertEquals(job.control_file, self.CONTROL_FILE_CONTENTS)
372        self.assertEquals(job.control_type, control_data.CONTROL_TYPE.SERVER)
373        entries = job.hostqueueentry_set.order_by('host__hostname')
374        self.assertEquals(entries[0].host.hostname, 'host1')
375        self.assertEquals(entries[1].host.hostname, 'host2')
376
377        owner_test = owner
378        if not owner_test:
379            owner_test = models.User.current_user().login
380        self.assertEquals(job.owner, owner_test)
381
382
383    def test_post_no_owner(self):
384        self._test_post_helper(None)
385
386
387    def test_post_with_owner(self):
388        self._test_post_helper('job_owner')
389
390
391class DirectoryTest(AfeResourceTestCase):
392    def test_get(self):
393        response = self.request('get', '')
394        for key in ('atomic_group_classes', 'labels', 'users', 'acl_groups',
395                    'hosts', 'tests', 'jobs', 'execution_info',
396                    'queue_entries_request'):
397            self.assert_(key in response)
398
399
400if __name__ == '__main__':
401    unittest.main()
402