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.