Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.