KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use test_queue_entries method in autotest
Best Python code snippet using autotest_python
resources_test.py
Source:resources_test.py 
...131 ['label1', 'myplatform'])132 def test_acls(self):133 self.check_relationship('hosts/host1', 'acls', 'acl', 'name',134 ['my_acl'])135 def test_queue_entries(self):136 self._create_job(hosts=[1])137 host = self.request('get', 'hosts/host1')138 entries = self.request('get', host['queue_entries']['href'])139 self.check_collection(entries, ['job', 'id'], [1])140 def test_health_tasks(self):141 models.SpecialTask.schedule_special_task(142 host=self.hosts[0], task=models.SpecialTask.Task.VERIFY)143 host = self.request('get', 'hosts/host1')144 tasks = self.request('get', host['health_tasks']['href'])145 self.check_collection(tasks, 'task_type', ['Verify'])146 def test_put(self):147 response = self.request('put', 'hosts/host1', data={'locked': True})148 self.assertEquals(response['locked'], True)149 response = self.request('get', 'hosts/host1')150 self.assertEquals(response['locked'], True)151 self.assertEquals(response['locked_by']['username'], 'debug_user')152 def test_post(self):153 data = {'hostname': 'newhost',154 'platform': {'href': self.URI_PREFIX + '/labels/myplatform'},155 'protection_level': 'Do not verify'}156 response = self.request('post', 'hosts', data=data)157 self.assertEquals(response, self.URI_PREFIX + '/hosts/newhost')158 host = models.Host.objects.get(hostname='newhost')159 self.assertEquals(host.platform().name, 'myplatform')160 self.assertEquals(host.protection, models.Host.Protection.DO_NOT_VERIFY)161 def _check_labels(self, host, expected_labels):162 label_names = sorted(label.name for label in host.labels.all())163 self.assertEquals(label_names, sorted(expected_labels))164 def test_add_label(self):165 labels_href = self.request('get', 'hosts/host1')['labels']['href']166 data = {'label': self.URI_PREFIX + '/labels/label2'}167 response = self.request('post', labels_href, data=data)168 self._check_labels(self.hosts[0], ['label1', 'label2', 'myplatform'])169 def test_remove_label(self):170 labels_href = self.request('get', 'hosts/host1')['labels']['href']171 labels_href += '&label=label1'172 labelings = self.request('get', labels_href)['members']173 self.assertEquals(len(labelings), 1)174 self.request('delete', labelings[0]['href'])175 self._check_labels(self.hosts[0], ['myplatform'])176 def test_delete(self):177 self.request('delete', 'hosts/host1')178 hosts = models.Host.valid_objects.filter(hostname='host1')179 self.assertEquals(len(hosts), 0)180class TestTest(AfeResourceTestCase): # yes, we're testing the "tests" resource181 def test_collection(self):182 response = self.request('get', 'tests')183 self.check_collection(response, 'name', ['mytest'])184 def test_entry(self):185 response = self.request('get', 'tests/mytest')186 self.assertEquals(response['name'], 'mytest')187 self.assertEquals(response['control_file_type'], 'Server')188 self.assertEquals(response['control_file_path'], '/path/to/mytest')189 def test_dependencies(self):190 models.Test.objects.get(name='mytest').dependency_labels = [self.label3]191 self.check_relationship('tests/mytest', 'dependencies', 'label', 'name',192 ['label3'])193class ExecutionInfoTest(AfeResourceTestCase):194 def setUp(self):195 super(ExecutionInfoTest, self).setUp()196 def mock_read_control_file(test):197 return self.CONTROL_FILE_CONTENTS198 self.god.stub_with(control_file, 'read_control_file',199 mock_read_control_file)200 def test_get(self):201 response = self.request('get', 'execution_info?tests=mytest')202 info = response['execution_info']203 self.assert_(self.CONTROL_FILE_CONTENTS in info['control_file'])204 self.assertEquals(info['is_server'], True)205 self.assertEquals(info['machines_per_execution'], 1)206class QueueEntriesRequestTest(AfeResourceTestCase):207 def test_get(self):208 response = self.request(209 'get',210 'queue_entries_request?hosts=host1,host2&meta_hosts=label1')211 # choose an arbitrary but consistent ordering to ease checking212 def entry_href(entry):213 if 'host' in entry:214 return entry['host']['href']215 return entry['meta_host']['href']216 entries = sorted(response['queue_entries'], key=entry_href)217 expected = [218 {'host': {'href': self.URI_PREFIX + '/hosts/host1'}},219 {'host': {'href': self.URI_PREFIX + '/hosts/host2'}},220 {'meta_host': {'href': self.URI_PREFIX + '/labels/label1'}}]221 self.assertEquals(entries, expected)222class JobTest(AfeResourceTestCase):223 def setUp(self):224 super(JobTest, self).setUp()225 for _ in xrange(2):226 self._create_job(hosts=[1, 2])227 job = models.Job.objects.get(id=1)228 job.control_file = self.CONTROL_FILE_CONTENTS229 job.save()230 models.JobKeyval.objects.create(job=job, key='mykey', value='myvalue')231 def test_collection(self):232 response = self.request('get', 'jobs')233 self.check_collection(response, 'id', [1, 2])234# def test_keyval_filtering(self):235# response = self.request('get', 'jobs?has_keyval=mykey=myvalue')236# self.check_collection(response, 'id', [1])237 def test_entry(self):238 response = self.request('get', 'jobs/1')239 self.assertEquals(response['id'], 1)240 self.assertEquals(response['name'], 'test')241 self.assertEquals(response['keyvals'], {'mykey': 'myvalue'})242 info = response['execution_info']243 self.assertEquals(info['control_file'], self.CONTROL_FILE_CONTENTS)244 self.assertEquals(info['is_server'], False)245 self.assertEquals(info['cleanup_before_job'], 'Never')246 self.assertEquals(info['cleanup_after_job'], 'Never')247 self.assertEquals(info['machines_per_execution'], 1)248 self.assertEquals(info['run_verify'], False)249 self.assertEquals(info['run_reset'], True)250 def test_queue_entries(self):251 job = self.request('get', 'jobs/1')252 entries = self.request('get', job['queue_entries']['href'])253 self.check_collection(entries, ['host', 'hostname'], ['host1', 'host2'])254 def _test_post_helper(self, owner):255 data = {'name': 'myjob',256 'execution_info': {'control_file': self.CONTROL_FILE_CONTENTS,257 'is_server': True},258 'owner': owner,259 'drone_set': models.DroneSet.default_drone_set_name(),260 'queue_entries':261 [{'host': {'href': self.URI_PREFIX + '/hosts/host1'}},262 {'host': {'href': self.URI_PREFIX + '/hosts/host2'}}]}263 response = self.request('post', 'jobs', data=data)264 self.assertEquals(response, self.URI_PREFIX + '/jobs/3')...
resources_unittest.py
Source:resources_unittest.py
...129 ['label1', 'myplatform'])130 def test_acls(self):131 self.check_relationship('hosts/host1', 'acls', 'acl', 'name',132 ['my_acl'])133 def test_queue_entries(self):134 self._create_job(hosts=[1])135 host = self.request('get', 'hosts/host1')136 entries = self.request('get', host['queue_entries']['href'])137 self.check_collection(entries, ['job', 'id'], [1])138 def test_health_tasks(self):139 models.SpecialTask.schedule_special_task(140 host=self.hosts[0], task=models.SpecialTask.Task.VERIFY)141 host = self.request('get', 'hosts/host1')142 tasks = self.request('get', host['health_tasks']['href'])143 self.check_collection(tasks, 'task_type', ['Verify'])144 def test_put(self):145 response = self.request('put', 'hosts/host1', data={'locked': True})146 self.assertEquals(response['locked'], True)147 response = self.request('get', 'hosts/host1')148 self.assertEquals(response['locked'], True)149 self.assertEquals(response['locked_by']['username'], 'debug_user')150 def test_post(self):151 data = {'hostname': 'newhost',152 'platform': {'href': self.URI_PREFIX + '/labels/myplatform'},153 'protection_level': 'Do not verify'}154 response = self.request('post', 'hosts', data=data)155 self.assertEquals(response, self.URI_PREFIX + '/hosts/newhost')156 host = models.Host.objects.get(hostname='newhost')157 self.assertEquals(host.platform().name, 'myplatform')158 self.assertEquals(host.protection, models.Host.Protection.DO_NOT_VERIFY)159 def _check_labels(self, host, expected_labels):160 label_names = sorted(label.name for label in host.labels.all())161 self.assertEquals(label_names, sorted(expected_labels))162 def test_add_label(self):163 labels_href = self.request('get', 'hosts/host1')['labels']['href']164 data = {'label': self.URI_PREFIX + '/labels/label2'}165 response = self.request('post', labels_href, data=data)166 self._check_labels(self.hosts[0], ['label1', 'label2', 'myplatform'])167 def test_remove_label(self):168 labels_href = self.request('get', 'hosts/host1')['labels']['href']169 labels_href += '&label=label1'170 labelings = self.request('get', labels_href)['members']171 self.assertEquals(len(labelings), 1)172 self.request('delete', labelings[0]['href'])173 self._check_labels(self.hosts[0], ['myplatform'])174 def test_delete(self):175 self.request('delete', 'hosts/host1')176 hosts = models.Host.valid_objects.filter(hostname='host1')177 self.assertEquals(len(hosts), 0)178class TestTest(AfeResourceTestCase): # yes, we're testing the "tests" resource179 def test_collection(self):180 response = self.request('get', 'tests')181 self.check_collection(response, 'name', ['mytest'])182 def test_entry(self):183 response = self.request('get', 'tests/mytest')184 self.assertEquals(response['name'], 'mytest')185 self.assertEquals(response['control_file_type'], 'Server')186 self.assertEquals(response['control_file_path'], '/path/to/mytest')187 def test_dependencies(self):188 models.Test.objects.get(name='mytest').dependency_labels = [self.label3]189 self.check_relationship('tests/mytest', 'dependencies', 'label', 'name',190 ['label3'])191class ExecutionInfoTest(AfeResourceTestCase):192 def setUp(self):193 super(ExecutionInfoTest, self).setUp()194 def mock_read_control_file(test):195 return self.CONTROL_FILE_CONTENTS196 self.god.stub_with(control_file, 'read_control_file',197 mock_read_control_file)198 def test_get(self):199 response = self.request('get', 'execution_info?tests=mytest')200 info = response['execution_info']201 self.assert_(self.CONTROL_FILE_CONTENTS in info['control_file'])202 self.assertEquals(info['is_server'], True)203 self.assertEquals(info['machines_per_execution'], 1)204class QueueEntriesRequestTest(AfeResourceTestCase):205 def test_get(self):206 response = self.request(207 'get',208 'queue_entries_request?hosts=host1,host2&meta_hosts=label1')209 # choose an arbitrary but consistent ordering to ease checking210 def entry_href(entry):211 if 'host' in entry:212 return entry['host']['href']213 return entry['meta_host']['href']214 entries = sorted(response['queue_entries'], key=entry_href)215 expected = [216 {'host': {'href': self.URI_PREFIX + '/hosts/host1'}},217 {'host': {'href': self.URI_PREFIX + '/hosts/host2'}},218 {'meta_host': {'href': self.URI_PREFIX + '/labels/label1'}}]219 self.assertEquals(entries, expected)220class JobTest(AfeResourceTestCase):221 def setUp(self):222 super(JobTest, self).setUp()223 for _ in xrange(2):224 self._create_job(hosts=[1, 2])225 job = models.Job.objects.get(id=1)226 job.control_file = self.CONTROL_FILE_CONTENTS227 job.save()228 models.JobKeyval.objects.create(job=job, key='mykey', value='myvalue')229 def test_collection(self):230 response = self.request('get', 'jobs')231 self.check_collection(response, 'id', [1, 2])232 def test_keyval_filtering(self):233 response = self.request('get', 'jobs?has_keyval=mykey=myvalue')234 self.check_collection(response, 'id', [1])235 def test_entry(self):236 response = self.request('get', 'jobs/1')237 self.assertEquals(response['id'], 1)238 self.assertEquals(response['name'], 'test')239 self.assertEquals(response['keyvals'], {'mykey': 'myvalue'})240 info = response['execution_info']241 self.assertEquals(info['control_file'], self.CONTROL_FILE_CONTENTS)242 self.assertEquals(info['is_server'], False)243 self.assertEquals(info['cleanup_before_job'], 'Never')244 self.assertEquals(info['cleanup_after_job'], 'Always')245 self.assertEquals(info['machines_per_execution'], 1)246 self.assertEquals(info['run_verify'], True)247 def test_queue_entries(self):248 job = self.request('get', 'jobs/1')249 entries = self.request('get', job['queue_entries']['href'])250 self.check_collection(entries, ['host', 'hostname'], ['host1', 'host2'])251 def _test_post_helper(self, owner):252 data = {'name': 'myjob',253 'execution_info': {'control_file': self.CONTROL_FILE_CONTENTS,254 'is_server': True},255 'owner': owner,256 'drone_set': models.DroneSet.default_drone_set_name(),257 'queue_entries':258 [{'host': {'href': self.URI_PREFIX + '/hosts/host1'}},259 {'host': {'href': self.URI_PREFIX + '/hosts/host2'}}]}260 response = self.request('post', 'jobs', data=data)261 self.assertEquals(response, self.URI_PREFIX + '/jobs/3')...
test_queuenames.py
Source:test_queuenames.py
...14 self.assertTrue(QueueType.publish == 'publish')15 def test_queue_entry(self):16 que = QueueEntry('', '', '')17 self.assertTrue(que)18 def test_queue_entries(self):19 que = QueueEntries()20 self.assertTrue(que)21 que.add('test', 'test')22 que.addbroadcast('qbroad', 'test')23 que.addalert('msg')24 self.assertTrue(que.hasentries())25if __name__ == '__main__':...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
