1 #!/usr/bin/python 2 3 import common 4 import unittest 5 6 # This has to be done very early. 7 from autotest_lib.client.common_lib import global_config 8 global_config.global_config.override_config_value( 9 'HOSTS', 'default_protection', 10 'NO_PROTECTION') 11 from autotest_lib.client.common_lib import host_protections 12 13 from autotest_lib.frontend import setup_django_environment 14 from autotest_lib.frontend import setup_test_environment 15 from django.test import client 16 from autotest_lib.frontend.shared import resource_test_utils 17 from autotest_lib.frontend.afe import control_file, models, model_attributes 18 from autotest_lib.client.common_lib import control_data 19 20 class 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 36 class 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 74 class MiscellaneousTest(AfeResourceTestCase): 75 def test_trailing_slash(self): 76 response = self.request('get', 'hosts/host1/') 77 self.assertEquals(response['hostname'], 'host1') 78 79 80 class 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 98 class 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 119 class 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 149 class 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 170 class 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 258 class 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 277 class 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 293 class 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 313 class 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 391 class 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 400 if __name__ == '__main__': 401 unittest.main() 402
