Best Python code snippet using autotest_python
rpc_interface_unittest.py
Source:rpc_interface_unittest.py
...150 self.assertEquals(len(queue_entries), 1)151 self.assertEquals(queue_entries[0].host, None)152 self.assertEquals(queue_entries[0].meta_host, None)153 self.assertEquals(queue_entries[0].atomic_group, None)154 def _setup_special_tasks(self):155 host = self.hosts[0]156 job1 = self._create_job(hosts=[1])157 job2 = self._create_job(hosts=[1])158 entry1 = job1.hostqueueentry_set.all()[0]159 entry1.update_object(started_on=datetime.datetime(2009, 1, 2),160 execution_subdir='host1')161 entry2 = job2.hostqueueentry_set.all()[0]162 entry2.update_object(started_on=datetime.datetime(2009, 1, 3),163 execution_subdir='host1')164 self.task1 = models.SpecialTask.objects.create(165 host=host, task=models.SpecialTask.Task.VERIFY,166 time_started=datetime.datetime(2009, 1, 1), # ran before job 1167 is_complete=True, requested_by=models.User.current_user())168 self.task2 = models.SpecialTask.objects.create(169 host=host, task=models.SpecialTask.Task.VERIFY,170 queue_entry=entry2, # ran with job 2171 is_active=True, requested_by=models.User.current_user())172 self.task3 = models.SpecialTask.objects.create(173 host=host, task=models.SpecialTask.Task.VERIFY,174 requested_by=models.User.current_user()) # not yet run175 def test_get_special_tasks(self):176 self._setup_special_tasks()177 tasks = rpc_interface.get_special_tasks(host__hostname='host1',178 queue_entry__isnull=True)179 self.assertEquals(len(tasks), 2)180 self.assertEquals(tasks[0]['task'], models.SpecialTask.Task.VERIFY)181 self.assertEquals(tasks[0]['is_active'], False)182 self.assertEquals(tasks[0]['is_complete'], True)183 def test_get_latest_special_task(self):184 # a particular usage of get_special_tasks()185 self._setup_special_tasks()186 self.task2.time_started = datetime.datetime(2009, 1, 2)187 self.task2.save()188 tasks = rpc_interface.get_special_tasks(189 host__hostname='host1', task=models.SpecialTask.Task.VERIFY,190 time_started__isnull=False, sort_by=['-time_started'],191 query_limit=1)192 self.assertEquals(len(tasks), 1)193 self.assertEquals(tasks[0]['id'], 2)194 def _common_entry_check(self, entry_dict):195 self.assertEquals(entry_dict['host']['hostname'], 'host1')196 self.assertEquals(entry_dict['job']['id'], 2)197 def test_get_host_queue_entries_and_special_tasks(self):198 self._setup_special_tasks()199 entries_and_tasks = (200 rpc_interface.get_host_queue_entries_and_special_tasks('host1'))201 paths = [entry['execution_path'] for entry in entries_and_tasks]202 self.assertEquals(paths, ['hosts/host1/3-verify',203 '2-autotest_system/host1',204 'hosts/host1/2-verify',205 '1-autotest_system/host1',206 'hosts/host1/1-verify'])207 verify2 = entries_and_tasks[2]208 self._common_entry_check(verify2)209 self.assertEquals(verify2['type'], 'Verify')210 self.assertEquals(verify2['status'], 'Running')211 self.assertEquals(verify2['execution_path'], 'hosts/host1/2-verify')212 entry2 = entries_and_tasks[1]...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!