• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号
登陆 注册

Python pulsar.get_application函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中pulsar.get_application函数的典型用法代码示例。如果您正苦于以下问题：Python get_application函数的具体用法？Python get_application怎么用？Python get_application使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了get_application函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_run_new_task_asynchronous_wait_on_test

 def test_run_new_task_asynchronous_wait_on_test(self):
     app = yield get_application(self.name())
     r = yield self.proxy.run_new_task(jobname='asynchronous', lag=3)
     r = yield app.backend.wait_for_task(r)
     self.assertEqual(r.status, tasks.SUCCESS)
     self.assertTrue(r.result['loops'])
     self.assertTrue(r.result['time'] > 3)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:7，代码来源:test_local.py


示例2: test_sorted_tags

 def test_sorted_tags(self):
     app = yield from get_application('test')
     loader = TestLoader(app.root_dir, app.cfg.modules, app.runner)
     modules = list(loader.testmodules())
     self.assertTrue(modules)
     tags = [m[0] for m in modules]
     self.assertEqual(tags, sorted(tags))
开发者ID:Danzeer，项目名称:pulsar，代码行数:7，代码来源:loader.py


示例3: test_id_not_overlap

 def test_id_not_overlap(self):
     '''Check `generate_task_ids` when `can_overlap` attribute is set to
     False.'''
     app = yield get_application(self.name())
     job = app.backend.registry['notoverlap']
     self.assertEqual(job.type, 'regular')
     self.assertFalse(job.can_overlap)
     #
     id1, oid1 = job.generate_task_ids((), {})
     self.assertTrue(oid1)
     id2, oid2 = job.generate_task_ids((), {})
     self.assertTrue(oid2)
     self.assertNotEqual(id2, id1)
     self.assertEqual(oid2, oid1)
     #
     id3, oid3 = job.generate_task_ids((10, 'bla'), {'p': 45})
     self.assertTrue(oid3)
     self.assertNotEqual(id3, id2)
     self.assertNotEqual(oid3, oid2)
     id4, oid4 = job.generate_task_ids((10, 'bla'), {'p': 45})
     self.assertNotEqual(id4, id3)
     self.assertEqual(oid4, oid3)
     #
     id5, oid5 = job.generate_task_ids((), {'p': 45, 'c': 'bla'})
     self.assertTrue(oid5)
     id6, oid6 = job.generate_task_ids((), {'p': 45, 'c': 'bla'})
     id7, oid7 = job.generate_task_ids((), {'p': 45, 'd': 'bla'})
     id8, oid8 = job.generate_task_ids((), {'p': 45, 'c': 'blas'})
     #
     self.assertEqual(oid5, oid6)
     self.assertNotEqual(oid5, oid7)
     self.assertNotEqual(oid5, oid8)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:32，代码来源:test_local.py


示例4: testRegistry

 def testRegistry(self):
     app = get_application(self.name_tq())
     self.assertTrue(isinstance(app.registry, dict))
     regular = app.registry.regular()
     periodic = app.registry.periodic()
     self.assertTrue(regular)
     self.assertTrue(periodic)
开发者ID:cyberj，项目名称:pulsar，代码行数:7，代码来源:tests.py


示例5: test_registry

 def test_registry(self):
     app = yield get_application(self.name())
     self.assertTrue(isinstance(app.backend.registry, dict))
     regular = app.backend.registry.regular()
     periodic = app.backend.registry.periodic()
     self.assertTrue(regular)
     self.assertTrue(periodic)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:7，代码来源:test_local.py


示例6: testMeta

 def testMeta(self):
     app = yield get_application(self.name())
     self.assertEqual(app.name, self.name())
     monitor = get_actor().get_actor(app.name)
     self.assertTrue(monitor.is_running())
     self.assertEqual(app, monitor.app)
     self.assertEqual(str(app), app.name)
     self.assertEqual(app.cfg.bind, '127.0.0.1:0')
开发者ID:JinsongBian，项目名称:pulsar，代码行数:8，代码来源:tests.py


示例7: testRpcMeta

 def testRpcMeta(self):
     app = get_application(self.name_rpc())
     self.assertTrue(app)
     self.assertEqual(app.name, self.name_rpc())
     rpc = app.callable
     root = rpc.handler
     tq = root.taskqueue
     self.assertEqual(tq, self.name_tq())
开发者ID:cyberj，项目名称:pulsar，代码行数:8，代码来源:tests.py


示例8: test_testsuite

 def test_testsuite(self):
     app = yield from get_application('test')
     self.assertTrue(app.cfg.script)
     # self.assertEqual(app.script, sys.argv[0])
     self.assertEqual(os.path.dirname(app.cfg.script), app.root_dir)
     self.assertEqual(app.cfg.modules, ('tests',
                                        ('examples', 'tests'),
                                        ('examples', 'test_*')))
开发者ID:Danzeer，项目名称:pulsar，代码行数:8，代码来源:loader.py


示例9: test_load_exclude

 def test_load_exclude(self):
     app = yield from get_application('test')
     modules = dict(app.loader.test_files(
         exclude=('stores', 'apps.pubsub')))
     self.assertFalse('stores.lock' in modules)
     self.assertFalse('stores.parser' in modules)
     self.assertFalse('apps.pubsub' in modules)
     self.assertTrue('apps.greenio' in modules)
开发者ID:juanignaciocatalano，项目名称:pulsar，代码行数:8，代码来源:test_loader.py


示例10: test_run_new_task_asynchronous_from_test

 def test_run_new_task_asynchronous_from_test(self):
     app = yield get_application(self.name())
     r = yield app.backend.run('asynchronous', lag=3)
     r = yield self.proxy.wait_for_task(r)
     self.assertEqual(r['status'], tasks.SUCCESS)
     result = r['result']
     self.assertTrue(result['loops'])
     self.assertTrue(result['time'] > 3)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:8，代码来源:test_local.py


示例11: test_load_exclude

 def test_load_exclude(self):
     app = yield from get_application('test')
     loader = TestLoader(app.root_dir, app.cfg.modules, app.runner)
     modules = dict(loader.testmodules(
         exclude_tags=('taskqueue', 'apps.pubsub')))
     self.assertTrue(modules)
     for module in modules:
         self.assertTrue('taskqueue' not in module)
         self.assertTrue('apps.pubsub' not in module)
开发者ID:Danzeer，项目名称:pulsar，代码行数:9，代码来源:loader.py


示例12: test_pubsub

 def test_pubsub(self):
     '''Tests meta attributes of taskqueue'''
     app = yield get_application(self.name())
     self.assertFalse(app.backend.local.pubsub)
     pubsub = app.backend.pubsub
     self.assertEqual(app.backend.local.pubsub, pubsub)
     # the pubsub name is the same as the task queue application
     self.assertEqual(app.backend.name, app.name)
     self.assertEqual(pubsub.name, app.name)
     self.pubsub_test(app)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:10，代码来源:test_local.py


示例13: __test_delete_task

 def __test_delete_task(self):
     #ISSUE #56
     app = yield get_application(self.name())
     id = yield app.backend.run('addition', 1, 4)
     r1 = yield app.backend.wait_for_task(id)
     self.assertEqual(r1.result, 5)
     deleted = yield app.backend.delete_tasks([r1.id, 'kjhbkjb'])
     self.assertEqual(len(deleted), 1)
     r1 = yield app.backend.get_task(r1.id)
     self.assertFalse(r1)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:10，代码来源:test_local.py


示例14: test_load_pulsar_tests

 def test_load_pulsar_tests(self):
     app = yield from get_application('test')
     modules = dict(app.loader.test_files())
     self.assertTrue(modules)
     self.assertFalse('httpbin' in modules)
     self.assertTrue('echo' in modules)
     self.assertFalse('djchat' in modules)
     self.assertTrue('djchat.app' in modules)
     self.assertTrue('djchat.pulse' in modules)
     self.assertFalse('async' in modules)
     self.assertTrue('async.actor' in modules)
开发者ID:juanignaciocatalano，项目名称:pulsar，代码行数:11，代码来源:test_loader.py


示例15: testDeleteTask

 def testDeleteTask(self):
     app = get_application(self.name_tq())
     r1 = app.scheduler.queue_task(app.monitor, 'addition', (1,4), {})
     id = r1.id
     get_task = app.scheduler.get_task
     while get_task(id).status in tasks.UNREADY_STATES:
         yield NOT_DONE
     r2 = get_task(id)
     self.assertEqual(r1.id, r2.id)
     r2 = get_task(id, remove=True)
     self.assertEqual(get_task(id), None)
     self.assertEqual(get_task(r1), r1)
     app.scheduler.delete_tasks()
开发者ID:cyberj，项目名称:pulsar，代码行数:13，代码来源:tests.py


示例16: test_not_overlap

 def test_not_overlap(self):
     sec = 2 + random()
     app = yield get_application(self.name())
     self.assertEqual(app.name, app.backend.name)
     self.assertTrue('notoverlap' in app.backend.registry)
     r1 = yield app.backend.run('notoverlap', sec)
     self.assertTrue(r1)
     r2 = yield app.backend.run('notoverlap', sec)
     self.assertFalse(r2)
     # We need to make sure the first task is completed
     r1 = yield app.backend.wait_for_task(r1)
     self.assertEqual(r1.status, tasks.SUCCESS)
     self.assertTrue(r1.result > sec)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:13，代码来源:test_local.py


示例17: testMeta

 def testMeta(self):
     '''Tests meta attributes of taskqueue'''
     app = get_application(self.name_tq())
     self.assertTrue(app)
     self.assertEqual(app.name, self.name_tq())
     self.assertTrue(app.registry)
     scheduler = app.scheduler
     self.assertTrue(scheduler.entries)
     job = app.registry['runpycode']
     self.assertEqual(job.type,'regular')
     self.assertTrue(job.can_overlap)
     id = job.make_task_id((),{})
     self.assertTrue(id)
     self.assertNotEqual(id,job.make_task_id((),{}))
开发者ID:cyberj，项目名称:pulsar，代码行数:14，代码来源:tests.py


示例18: test_rpc_meta

 def test_rpc_meta(self):
     app = yield get_application(self.rpc_name())
     self.assertTrue(app)
     self.assertEqual(app.name, self.rpc_name())
     self.assertEqual(app.cfg.address, ('127.0.0.1', 0))
     self.assertNotEqual(app.cfg.address, app.address)
     self.assertEqual(app.cfg.concurrency, self.concurrency)
     wsgi_handler = app.callable.handler
     self.assertEqual(len(wsgi_handler.middleware), 1)
     router = wsgi_handler.middleware[0]
     self.assertTrue(router.post)
     root = router.post
     tq = root.taskqueue
     self.assertEqual(tq, self.name())
开发者ID:BazookaShao，项目名称:pulsar，代码行数:14，代码来源:test_local.py


示例19: start_pulsar_ds

def start_pulsar_ds(arbiter, host, workers=0):
    lock = getattr(arbiter, 'lock', None)
    if lock is None:
        arbiter.lock = lock = asyncio.Lock()
    yield from lock.acquire()
    try:
        app = yield from get_application('pulsards')
        if not app:
            app = PulsarDS(bind=host, workers=workers, load_config=False)
            cfg = yield from app(arbiter)
        else:
            cfg = app.cfg
        return cfg
    finally:
        lock.release()
开发者ID:LJS109，项目名称:pulsar，代码行数:15，代码来源:startds.py


示例20: testNotOverlap

 def testNotOverlap(self):
     app = get_application(self.name_tq())
     self.assertTrue('notoverlap' in app.registry)
     r1 = app.scheduler.queue_task(app.monitor, 'notoverlap', (1,), {})
     self.assertEqual(str(r1), 'notoverlap(%s)' % r1.id)
     self.assertTrue(r1._queued)
     r2 = app.scheduler.queue_task(app.monitor, 'notoverlap', (1,), {})
     self.assertFalse(r2._queued)
     id = r1.id
     self.assertEqual(id, r2.id)
     # We need to make sure the first task is completed
     get_task = app.scheduler.get_task
     while get_task(id).status in tasks.UNREADY_STATES:
         yield NOT_DONE
     self.assertEqual(get_task(id).status, tasks.SUCCESS)
开发者ID:cyberj，项目名称:pulsar，代码行数:15，代码来源:tests.py



注：本文中的pulsar.get_application函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Python pulsar.get_event_loop函数代码示例发布时间：2022-05-25
下一篇：
Python pulsar.get_actor函数代码示例发布时间：2022-05-25
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap