KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use push_job method in yandex-tank
Best Python code snippet using yandex-tank
test_project.py
Source:test_project.py 
...19 self.assertEqual(type(p1.projectid), six.text_type)20 self.assertEqual(type(p2.projectid), six.text_type)21 self.assertRaises(AssertionError, self.hsclient.get_project, '111/3')22 def test_get_job_from_key(self):23 job = self.project.push_job(self.spidername)24 parts = tuple(job.key.split('/'))25 self.assertEqual(len(parts), 3)26 self.assertEqual(parts[:2], (self.projectid, self.spiderid))27 samejob1 = self.hsclient.get_job(job.key)28 samejob2 = self.project.get_job(job.key)29 samejob3 = self.project.get_job(parts[1:])30 self.assertEqual(samejob1.key, job.key)31 self.assertEqual(samejob2.key, job.key)32 self.assertEqual(samejob3.key, job.key)33 def test_get_jobs(self):34 p = self.project35 j1 = p.push_job(self.spidername, testid=0)36 j2 = p.push_job(self.spidername, testid=1)37 j3 = p.push_job(self.spidername, testid=2)38 # global list must list at least one job39 self.assertTrue(list(p.get_jobs(count=1, state='pending')))40 # List all jobs for test spider41 r = list(p.get_jobs(spider=self.spidername, state='pending'))42 self.assertEqual([j.key for j in r], [j3.key, j2.key, j1.key])43 def test_get_jobs_with_legacy_filter(self):44 p = self.project45 j1 = p.push_job(self.spidername, state='finished',46 close_reason='finished', tags=['t2'])47 j2 = p.push_job(self.spidername, state='finished',48 close_reason='finished', tags=['t1'])49 j3 = p.push_job(self.spidername, state='pending')50 j4 = p.push_job(self.spidername, state='finished',51 close_reason='failed', tags=['t1'])52 j5 = p.push_job(self.spidername + 'skip', state='finished',53 close_reason='failed', tags=['t1'])54 filters = [['spider', '=', [self.spidername]],55 ['state', '=', ['finished']],56 ['close_reason', '=', ['finished']],57 ['tags', 'haselement', ['t1']],58 ['tags', 'hasnotelement', ['t2']]]59 jobs = p.get_jobs(filter=[json.dumps(x) for x in filters])60 assert [j.key for j in jobs] == [j2.key], jobs61 def test_push_job(self):62 job = self.project.push_job(self.spidername, state='running',63 priority=self.project.jobq.PRIO_HIGH,64 foo=u'bar')65 self.assertEqual(job.metadata.get('state'), u'running')66 self.assertEqual(job.metadata.get('foo'), u'bar')67 self.project.jobq.finish(job)68 self.project.jobq.delete(job)69 job.metadata.expire()70 self.assertEqual(job.metadata.get('state'), u'deleted')71 self.assertEqual(job.metadata.get('foo'), u'bar')72 def test_auth(self):73 # client without global auth set74 hsc = HubstorageClient(endpoint=self.hsclient.endpoint)75 self.assertEqual(hsc.auth, None)76 # check no-auth access77 try:78 hsc.push_job(self.projectid, self.spidername)79 except HTTPError as exc:80 self.assertTrue(exc.response.status_code, 401)81 else:82 self.assertTrue(False, '401 not raised')83 try:84 hsc.get_project(self.projectid).push_job(self.spidername)85 except HTTPError as exc:86 self.assertTrue(exc.response.status_code, 401)87 else:88 self.assertTrue(False, '401 not raised')89 try:90 hsc.get_job((self.projectid, 1, 1)).items.list()91 except HTTPError as exc:92 self.assertTrue(exc.response.status_code, 401)93 else:94 self.assertTrue(False, '401 not raised')95 try:96 hsc.get_project(self.projectid).get_job((self.projectid, 1, 1)).items.list()97 except HTTPError as exc:98 self.assertTrue(exc.response.status_code, 401)99 else:100 self.assertTrue(False, '401 not raised')101 # create project with auth102 auth = self.hsclient.auth103 project = hsc.get_project(self.projectid, auth)104 self.assertEqual(project.auth, auth)105 job = project.push_job(self.spidername)106 samejob = project.get_job(job.key)107 self.assertEqual(samejob.key, job.key)108 def test_broad(self):109 project = self.hsclient.get_project(self.projectid)110 # populate project with at least one job111 job = project.push_job(self.spidername)112 self.assertEqual(job.metadata.get('state'), 'pending')113 job = self.start_job()114 self.assertEqual(job.metadata.get('state'), 'running')115 job.items.write({'title': 'bar'})116 job.logs.info('nice to meet you')117 job.samples.write([1, 2, 3])118 job.close_writers()119 job.jobq.finish(job)120 # keep a jobid for get_job and unreference job121 jobid = job.key122 jobauth = job.auth123 del job124 self.assertTrue(list(project.items.list(self.spiderid, count=1)))125 self.assertTrue(list(project.logs.list(self.spiderid, count=1)))126 self.assertTrue(list(project.samples.list(self.spiderid, count=1)))127 job = project.client.get_job(jobid, jobauth=jobauth)128 job.purged()129 def test_settings(self):130 project = self.hsclient.get_project(self.projectid)131 settings = dict(project.settings)132 settings.pop('botgroups', None) # ignore testsuite botgroups133 self.assertEqual(settings, {})134 project.settings['created'] = created = millitime()135 project.settings['botgroups'] = ['g1']136 project.settings.save()137 self.assertEqual(project.settings.liveget('created'), created)138 self.assertEqual(project.settings.liveget('botgroups'), ['g1'])139 project.settings.expire()140 self.assertEqual(dict(project.settings), {141 'created': created,142 'botgroups': ['g1'],143 })144 def test_requests(self):145 ts = millitime()146 job = self.project.push_job(self.spidername, state='running')147 # top parent148 r1 = job.requests.add(url='http://test.com/', status=200, method='GET',149 rs=1337, duration=5, parent=None, ts=ts)150 # first child151 r2 = job.requests.add(url='http://test.com/2', status=400, method='POST',152 rs=0, duration=1, parent=r1, ts=ts + 1)153 # another child with fingerprint set154 r3 = job.requests.add(url='http://test.com/3', status=400, method='PUT',155 rs=0, duration=1, parent=r1, ts=ts + 2, fp='1234')156 job.requests.close()157 rr = job.requests.list()158 self.assertEqual(next(rr),159 {u'status': 200, u'rs': 1337,160 u'url': u'http://test.com/', u'time': ts,161 u'duration': 5, u'method': u'GET'})162 self.assertEqual(next(rr),163 {u'status': 400, u'parent': 0, u'rs': 0,164 u'url': u'http://test.com/2', u'time': ts + 1,165 u'duration': 1, u'method': u'POST'})166 self.assertEqual(next(rr),167 {u'status': 400, u'fp': u'1234', u'parent': 0,168 u'rs': 0, u'url': u'http://test.com/3',169 u'time': ts + 2, u'duration': 1,170 u'method': u'PUT'})171 self.assertRaises(StopIteration, next, rr)172 def test_samples(self):173 # no samples stored174 j1 = self.project.push_job(self.spidername, state='running')175 self.assertEqual(list(j1.samples.list()), [])176 # simple fill177 ts = millitime()178 j1.samples.write([ts, 1, 2, 3])179 j1.samples.write([ts + 1, 5, 9, 4])180 j1.samples.flush()181 o = list(j1.samples.list())182 self.assertEqual(len(o), 2)183 self.assertEqual(o[0], [ts, 1, 2, 3])184 self.assertEqual(o[1], [ts + 1, 5, 9, 4])185 # random fill186 j2 = self.project.push_job(self.spidername, state='running')187 samples = []188 ts = millitime()189 count = int(j2.samples.batch_size * (random() + randint(1, 5)))190 for _ in range(count):191 ts += randint(1, 2**16)192 row = [ts] + list(randint(0, 2**16) for _ in range(randint(0, 100)))193 samples.append(row)194 j2.samples.write(row)195 j2.samples.flush()196 o = list(j2.samples.list())197 self.assertEqual(len(o), count)198 for r1, r2 in zip(samples, o):199 self.assertEqual(r1, r2)200 def test_jobsummary(self):201 js = self.project.jobsummary()202 self.assertEqual(js.get('project'), int(self.project.projectid), js)203 self.assertEqual(js.get('has_capacity'), True, js)204 self.assertTrue('pending' in js, js)205 self.assertTrue('running' in js, js)206 def test_bulkdata(self):207 j = self.project.push_job(self.spidername, state='running')208 for i in range(20):209 j.logs.info("log line %d" % i)210 j.items.write(dict(field1="item%d" % i))211 j.requests.add("http://test.com/%d" % i,212 200, 'GET', 10, None, 10, 120)213 for resourcename in ('logs', 'items', 'requests'):214 resource = getattr(j, resourcename)215 resource.flush()216 # downloading resource, with simulated failures217 with failing_downloader(resource):218 downloaded = list(resource.iter_values())219 self.assertEqual(len(downloaded), 20)220 def test_output_string(self):221 project = self.hsclient.get_project(self.projectid)222 project.push_job(self.spidername)223 job = self.start_job()224 job.items.write({'foo': 'bar'})225 job.close_writers()226 items = self.hsclient.get_job(job.key).items.iter_json()227 self.assertEqual(type(next(items)), str)228@pytest.mark.parametrize('msgpack_available', [True, False])229@pytest.mark.parametrize('path,expected_result', [230 (None, True),231 ('33/1', True),232 ('33/1/', True),233 ((33, 1), True),234 ('stats', False),235 ('stats/', False),236 ('33/1/stats', False),...
test_job_queue.py
Source:test_job_queue.py
...21 queue = JobQueue()22 self.assertTrue(queue.empty)23 def test_empty_queue_single_push(self):24 queue = JobQueue()25 queue.push_job(Job())26 self.assertFalse(queue.empty)27 queue.pop_job()28 self.assertTrue(queue.empty)29 def test_empty_queue_multiple_pushes(self):30 queue = JobQueue()31 queue.push_job(Job())32 self.assertFalse(queue.empty)33 queue.push_job(Job())34 self.assertFalse(queue.empty)35 queue.pop_job()36 self.assertFalse(queue.empty)37 queue.pop_job()38 self.assertTrue(queue.empty)39 def test_push_pop_same_job(self):40 queue = JobQueue()41 job = Job()42 queue.push_job(job)43 self.assertIs(queue.pop_job(), job)44 def test_peek_job(self):45 queue = JobQueue()46 job = Job()47 queue.push_job(job)48 self.assertIs(queue.peek_job(), job)49 self.assertFalse(queue.empty)50 self.assertIs(queue.peek_job(), job)51 def test_pop_empty(self):52 queue = JobQueue()53 self.assertIs(queue.pop_job(), None)54 job = Job()55 queue.push_job(job)56 self.assertIs(queue.pop_job(), job)57 self.assertIs(queue.pop_job(), None)58 def test_job_pop_ordering(self):59 for combo in itertools.permutations([WeightedJob(1), WeightedJob(2), WeightedJob(3)]):60 queue = JobQueue()61 queue.add_priority_rule(weight_rule)62 for job in combo:63 queue.push_job(job)64 job = queue.pop_job()65 self.assertEqual(job.weight, 3)66 job = queue.pop_job()67 self.assertEqual(job.weight, 2)68 job = queue.pop_job()69 self.assertEqual(job.weight, 1)70 def test_job_reorder_queue(self):71 queue = JobQueue()72 for job in [WeightedJob(1), WeightedJob(2), WeightedJob(3)]:73 queue.push_job(job)74 rule = queue.add_priority_rule(weight_rule)75 self.assertEqual(queue.peek_job().weight, 3)76 queue.remove_priority_rule(rule)77 queue.add_priority_rule(reverse_weight_rule)...
init_job.py
Source:init_job.py
1from redis_cookies import RedisJob2from weibo_cn import JobType3# RedisJob.push_job(JobType.comment.value, {'url': 'https://weibo.cn/comment/FCoPpaIQp', 'tweetId': 'FCoPpaIQp'})4RedisJob.push_job(JobType.user.value, {'user_id': '2210643391'})...
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 yandex-tank 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?
