Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! 
How to use new_event_loop method in pytest-asyncio
Best Python code snippet using pytest-asyncio_python
test_workq.py
Source:test_workq.py 
...6from workq.error import WorkqTimeout, WorkqJobIdNotFound7from workq.job import BackgroundJob, ScheduledJob, ForegroundJob8class TestWorkq(unittest.TestCase):9 def test_add(self):10 loop = asyncio.new_event_loop()11 client = WorkqClient('127.0.0.1', 9922, loop)12 jobid = uuid.uuid4()13 job = BackgroundJob(jobid, "ping1", 5000, 60000, "hello bg job")14 try:15 loop.run_until_complete(client.connect())16 loop.run_until_complete(client.add(job))17 leased_job = loop.run_until_complete(client.lease(("ping1", ), 10000))18 self.assertEqual(len(leased_job), 1)19 self.assertEqual(leased_job[0].id, "%s" % jobid)20 self.assertEqual(leased_job[0].name, "ping1")21 finally:22 loop.close()23 def test_schedule(self):24 loop = asyncio.new_event_loop()25 client = WorkqClient('127.0.0.1', 9922, loop)26 jobid = uuid.uuid4()27 t = datetime.utcnow() + timedelta(seconds=1)28 job = ScheduledJob(jobid, "ping1", 5000, 60000, t, "hello bg job")29 try:30 loop.run_until_complete(client.connect())31 loop.run_until_complete(client.add(job))32 leased_job = loop.run_until_complete(client.lease(("ping1", ), 10000))33 self.assertEqual(len(leased_job), 1)34 self.assertEqual(leased_job[0].id, "%s" % jobid)35 self.assertEqual(leased_job[0].name, "ping1")36 finally:37 loop.close()38 def test_run(self):39 self.skipTest("broken")40 loop = asyncio.new_event_loop()41 client = WorkqClient('127.0.0.1', 9922, loop)42 jobid = uuid.uuid4()43 job = ForegroundJob(jobid, "fg1", 5000, 60000, "hello fg job")44 @asyncio.coroutine45 def parallel(client, loop):46 run_task = asyncio.ensure_future(client.run(job), loop=loop)47 lease_task = asyncio.ensure_future(client.lease(("fg1", ), 10000), loop=loop)48 tasks = [run_task, lease_task]49 results = yield from asyncio.gather(*tasks, loop=loop)50 return results51 try:52 loop.run_until_complete(client.connect())53 loop.run_until_complete(parallel(client, loop))54 finally:55 loop.close()56 def test_lease_timeout(self):57 loop = asyncio.new_event_loop()58 client = WorkqClient('127.0.0.1', 9922, loop)59 try:60 loop.run_until_complete(client.connect())61 with self.assertRaises(WorkqTimeout):62 leased_job = loop.run_until_complete(client.lease(("ping1", ), 100))63 print(leased_job)64 finally:65 loop.close()66 def test_complete(self):67 loop = asyncio.new_event_loop()68 client = WorkqClient('127.0.0.1', 9922, loop)69 jobid = uuid.uuid4()70 job = BackgroundJob(jobid, "test.complete1", 5000, 60000, "hello bg job")71 try:72 loop.run_until_complete(client.connect())73 loop.run_until_complete(client.add(job))74 leased_job = loop.run_until_complete(client.lease(("test.complete1", ), 10000))75 self.assertEqual(len(leased_job), 1)76 ljob = leased_job[0]77 loop.run_until_complete(client.complete(ljob.id, "ok"))78 finally:79 loop.close()80 def test_fail(self):81 loop = asyncio.new_event_loop()82 client = WorkqClient('127.0.0.1', 9922, loop)83 jobid = uuid.uuid4()84 job = BackgroundJob(jobid, "test.complete1", 5000, 60000, "hello bg job")85 try:86 loop.run_until_complete(client.connect())87 loop.run_until_complete(client.add(job))88 leased_job = loop.run_until_complete(client.lease(("test.complete1", ), 10000))89 self.assertEqual(len(leased_job), 1)90 ljob = leased_job[0]91 loop.run_until_complete(client.fail(ljob.id, "ng"))92 finally:93 loop.close()94 def test_delete(self):95 loop = asyncio.new_event_loop()96 client = WorkqClient('127.0.0.1', 9922, loop)97 jobid = uuid.uuid4()98 job = BackgroundJob(jobid, "test.complete1", 5000, 60000, "hello bg job")99 try:100 loop.run_until_complete(client.connect())101 loop.run_until_complete(client.add(job))102 loop.run_until_complete(client.delete(job.id))103 finally:104 loop.close()105 def test_delete_with_error(self):106 loop = asyncio.new_event_loop()107 client = WorkqClient('127.0.0.1', 9922, loop)108 jobid = uuid.uuid4()109 try:110 loop.run_until_complete(client.connect())111 with self.assertRaises(WorkqJobIdNotFound):112 loop.run_until_complete(client.delete(jobid))113 finally:114 loop.close()115if __name__ == '__main__':...
test_cron.py
Source:test_cron.py
...6import pytest7class CustomError(Exception):8 pass9def test_str():10 loop = asyncio.new_event_loop()11 @crontab("* * * * * *", loop=loop)12 def t():13 pass14 assert "* * * * *" in str(t)15def test_sleep():16 loop = asyncio.new_event_loop()17 future = asyncio.Future(loop=loop)18 @crontab("* * * * * *", loop=loop)19 async def t():20 print(datetime.datetime.now(), "t.begin")21 await asyncio.sleep(3)22 print(datetime.datetime.now(), "t.end")23 # t.start()24 # loop.run_until_complete(future)25def test_cron():26 loop = asyncio.new_event_loop()27 future = asyncio.Future(loop=loop)28 @crontab("* * * * * *", start=False, loop=loop)29 def t():30 # è°ç¨ä¸æ¬¡å°±ç»æ31 future.set_result(1)32 t.start()33 loop.run_until_complete(future)34 t.stop()35 assert future.result() == 136def test_raise():37 loop = asyncio.new_event_loop()38 future = asyncio.Future(loop=loop)39 @crontab("* * * * * *", start=False, loop=loop)40 def t():41 loop.call_later(1, future.set_result, 1)42 raise ValueError()43 t.start()44 loop.run_until_complete(future)45 t.stop()46 assert future.result() == 147def test_next():48 loop = asyncio.new_event_loop()49 def t():50 return 151 t = crontab("* * * * * *", func=t, loop=loop)52 future = asyncio.ensure_future(t.next(), loop=loop)53 loop.run_until_complete(future)54 assert future.result() == 155def test_null_callback():56 loop = asyncio.new_event_loop()57 t = crontab("* * * * * *", loop=loop)58 assert t.handle is None # not started59 future = asyncio.ensure_future(t.next(4), loop=loop)60 loop.run_until_complete(future)61 assert future.result() == (4,)62def test_next_raise():63 loop = asyncio.new_event_loop()64 @crontab("* * * * * *", loop=loop)65 def t():66 raise CustomError()67 future = asyncio.ensure_future(t.next(), loop=loop)68 with pytest.raises(CustomError):69 loop.run_until_complete(future)70def test_coro_next():71 loop = asyncio.new_event_loop()72 @crontab("* * * * * *", loop=loop)73 async def t():74 return 175 future = asyncio.ensure_future(t.next(), loop=loop)76 loop.run_until_complete(future)77 assert future.result() == 178def test_coro_next_raise():79 loop = asyncio.new_event_loop()80 @crontab("* * * * * *", loop=loop)81 async def t():82 raise CustomError()83 future = asyncio.ensure_future(t.next(), loop=loop)84 with pytest.raises(CustomError):85 loop.run_until_complete(future)86def test_next_dst(monkeypatch):87 class mydatetime:88 @classmethod89 def now(cls, tzinfo=None):90 return datetime.datetime(2017, 10, 29, 2, 58, 58, tzinfo=tzinfo)91 monkeypatch.setattr("aiocron.datetime", mydatetime)92 monkeypatch.setattr("dateutil.tz.time.timezone", -3600)93 monkeypatch.setattr("dateutil.tz.time.altzone", -7200)94 monkeypatch.setattr("dateutil.tz.time.daylight", 1)95 monkeypatch.setattr("dateutil.tz.time.tzname", ("CET", "CEST"))96 loop = asyncio.new_event_loop()97 t = crontab("* * * * *", loop=loop)98 t.initialize()99 # last hit in DST100 a = t.get_next()101 time.sleep(3)102 # first hit after DST103 b = t.get_next()...
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 pytest-asyncio 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?
