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

Python tests.expect函数代码示例

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

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



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

示例1: test_illformed_id

    def test_illformed_id(self):
        # any ill-formed IDs should be regarded as non-existing ones.

        self.controller.delete(self.queue_name, "illformed", project=self.project)

        with testing.expect(errors.DoesNotExist):
            self.controller.get(self.queue_name, "illformed", project=self.project)

        with testing.expect(errors.DoesNotExist):
            self.controller.update(self.queue_name, "illformed", {"ttl": 40}, project=self.project)
开发者ID:kangliqiang,项目名称:marconi,代码行数:10,代码来源:base.py


示例2: test_expired_claim

    def test_expired_claim(self):
        meta = {"ttl": 0, "grace": 60}

        claim_id, messages = self.controller.create(self.queue_name, meta, project=self.project)

        with testing.expect(storage.errors.DoesNotExist):
            self.controller.get(self.queue_name, claim_id, project=self.project)

        with testing.expect(storage.errors.DoesNotExist):
            self.controller.update(self.queue_name, claim_id, meta, project=self.project)
开发者ID:kangliqiang,项目名称:marconi,代码行数:10,代码来源:base.py


示例3: test_expired_claim

    def test_expired_claim(self):
        meta = {'ttl': 0, 'grace': 60}

        claim_id, messages = self.controller.create(self.queue_name, meta,
                                                    project=self.project)

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get(self.queue_name, claim_id,
                                project=self.project)

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.update(self.queue_name, claim_id,
                                   meta, project=self.project)
开发者ID:chadlung,项目名称:marconi,代码行数:13,代码来源:base.py


示例4: test_claim_effects

    def test_claim_effects(self):
        client_uuid = uuid.uuid4()

        _insert_fixtures(self.controller, self.queue_name, project=self.project, client_uuid=client_uuid, num=12)

        def list_messages(include_claimed=None):
            kwargs = {"project": self.project, "client_uuid": client_uuid, "echo": True}

            # Properly test default value
            if include_claimed is not None:
                kwargs["include_claimed"] = include_claimed

            interaction = self.controller.list(self.queue_name, **kwargs)

            messages = next(interaction)
            return [msg["id"] for msg in messages]

        messages_before = list_messages(True)

        meta = {"ttl": 70, "grace": 60}
        another_cid, _ = self.claim_controller.create(self.queue_name, meta, project=self.project)

        messages_after = list_messages(True)
        self.assertEqual(messages_before, messages_after)

        messages_excluding_claimed = list_messages()
        self.assertNotEqual(messages_before, messages_excluding_claimed)
        self.assertEqual(2, len(messages_excluding_claimed))

        cid, msgs = self.claim_controller.create(self.queue_name, meta, project=self.project)
        [msg1, msg2] = msgs

        # A wrong claim does not ensure the message deletion
        with testing.expect(storage.errors.NotPermitted):
            self.controller.delete(self.queue_name, msg1["id"], project=self.project, claim=another_cid)

        # Make sure a message can be deleted with a claim
        self.controller.delete(self.queue_name, msg1["id"], project=self.project, claim=cid)

        with testing.expect(storage.errors.DoesNotExist):
            self.controller.get(self.queue_name, msg1["id"], project=self.project)

        # Make sure such a deletion is idempotent
        self.controller.delete(self.queue_name, msg1["id"], project=self.project, claim=cid)

        # A non-existing claim does not ensure the message deletion
        self.claim_controller.delete(self.queue_name, cid, project=self.project)

        with testing.expect(storage.errors.NotPermitted):
            self.controller.delete(self.queue_name, msg2["id"], project=self.project, claim=cid)
开发者ID:kangliqiang,项目名称:marconi,代码行数:50,代码来源:base.py


示例5: test_message_lifecycle

    def test_message_lifecycle(self):
        queue_name = self.queue_name

        messages = [
            {
                'ttl': 60,
                'body': {
                    'event': 'BackupStarted',
                    'backupId': 'c378813c-3f0b-11e2-ad92-7823d2b0f3ce'
                }
            },
        ]

        # Test Message Creation
        created = list(self.controller.post(queue_name, messages,
                                            project=self.project,
                                            client_uuid=uuid.uuid4()))
        self.assertEqual(len(created), 1)

        # Test Message Get
        self.controller.get(queue_name, created[0], project=self.project)

        # Test Message Deletion
        self.controller.delete(queue_name, created[0], project=self.project)

        # Test does not exist
        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get(queue_name, created[0], project=self.project)
开发者ID:chadlung,项目名称:marconi,代码行数:28,代码来源:base.py


示例6: test_message_lifecycle

    def test_message_lifecycle(self):
        queue_name = self.queue_name

        message = {
            'ttl': 60,
            'body': {
                'event': 'BackupStarted',
                'backupId': 'c378813c-3f0b-11e2-ad92-7823d2b0f3ce'
            }
        }

        # Test Message Creation
        created = list(self.controller.post(queue_name, [message],
                                            project=self.project,
                                            client_uuid=uuid.uuid4()))
        self.assertEqual(len(created), 1)
        message_id = created[0]

        # Test Message Get
        message_out = self.controller.get(queue_name, message_id,
                                          project=self.project)
        self.assertEqual(set(message_out), set(('id', 'body', 'ttl', 'age')))
        self.assertEqual(message_out['id'], message_id)
        self.assertEqual(message_out['body'], message['body'])
        self.assertEqual(message_out['ttl'], message['ttl'])

        # Test Message Deletion
        self.controller.delete(queue_name, message_id, project=self.project)

        # Test does not exist
        with testing.expect(storage.errors.DoesNotExist):
            self.controller.get(queue_name, message_id, project=self.project)
开发者ID:TheSriram,项目名称:marconi-1,代码行数:32,代码来源:base.py


示例7: test_illformed_id

    def test_illformed_id(self):
        # any ill-formed IDs should be regarded as non-existing ones.

        self.controller.delete(self.queue_name,
                               'illformed',
                               project=self.project)

        with testing.expect(exceptions.DoesNotExist):
            self.controller.get(self.queue_name,
                                'illformed',
                                project=self.project)

        with testing.expect(exceptions.DoesNotExist):
            self.controller.update(self.queue_name,
                                   'illformed',
                                   {'ttl': 40},
                                   project=self.project)
开发者ID:chadlung,项目名称:marconi,代码行数:17,代码来源:base.py


示例8: test_bad_id

    def test_bad_id(self):
        # NOTE(cpp-cabrera): A malformed ID should result in an empty
        # query. Raising an exception for validating IDs makes the
        # implementation more verbose instead of taking advantage of
        # the Maybe/Optional protocol, particularly when dealing with
        # bulk operations.
        bad_message_id = "xyz"
        self.controller.delete(self.queue_name, bad_message_id, project=self.project)

        with testing.expect(errors.MessageDoesNotExist):
            self.controller.get(self.queue_name, bad_message_id, project=self.project)
开发者ID:kangliqiang,项目名称:marconi,代码行数:11,代码来源:base.py


示例9: test_expired_message

    def test_expired_message(self):
        messages = [{'body': 3.14, 'ttl': 0}]

        [msgid] = self.controller.post(self.queue_name, messages,
                                       project=self.project,
                                       client_uuid='my_uuid')

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get(self.queue_name, msgid,
                                project=self.project)

        countof = self.queue_controller.stats(self.queue_name,
                                              project=self.project)
        self.assertEquals(countof['messages']['free'], 0)
开发者ID:pombredanne,项目名称:marconi,代码行数:14,代码来源:base.py


示例10: test_multi_ids

    def test_multi_ids(self):
        messages_in = [{"ttl": 120, "body": 0}, {"ttl": 240, "body": 1}]
        ids = self.controller.post(self.queue_name, messages_in, project=self.project, client_uuid=uuid.uuid4())

        messages_out = self.controller.bulk_get(self.queue_name, ids, project=self.project)

        for idx, message in enumerate(messages_out):
            self.assertEqual(message["body"], idx)

        self.controller.bulk_delete(self.queue_name, ids, project=self.project)

        with testing.expect(StopIteration):
            result = self.controller.bulk_get(self.queue_name, ids, project=self.project)
            next(result)
开发者ID:kangliqiang,项目名称:marconi,代码行数:14,代码来源:base.py


示例11: test_expired_messages

    def test_expired_messages(self):
        messages = [{"body": 3.14, "ttl": 0}]
        client_uuid = uuid.uuid4()

        [msgid] = self.controller.post(self.queue_name, messages, project=self.project, client_uuid=client_uuid)

        [msgid] = self.controller.post(self.queue_name, messages, project=self.project, client_uuid=client_uuid)

        time.sleep(self.gc_interval)

        with testing.expect(storage.errors.DoesNotExist):
            self.controller.get(self.queue_name, msgid, project=self.project)

        stats = self.queue_controller.stats(self.queue_name, project=self.project)

        self.assertEqual(stats["messages"]["free"], 0)
开发者ID:kangliqiang,项目名称:marconi,代码行数:16,代码来源:base.py


示例12: test_message_lifecycle

    def test_message_lifecycle(self):
        queue_name = self.queue_name

        messages = [{"ttl": 60, "body": {"event": "BackupStarted", "backupId": "c378813c-3f0b-11e2-ad92-7823d2b0f3ce"}}]

        # Test Message Creation
        created = list(self.controller.post(queue_name, messages, project=self.project, client_uuid=uuid.uuid4()))
        self.assertEqual(len(created), 1)

        # Test Message Get
        self.controller.get(queue_name, created[0], project=self.project)

        # Test Message Deletion
        self.controller.delete(queue_name, created[0], project=self.project)

        # Test does not exist
        with testing.expect(storage.errors.DoesNotExist):
            self.controller.get(queue_name, created[0], project=self.project)
开发者ID:kangliqiang,项目名称:marconi,代码行数:18,代码来源:base.py


示例13: test_multi_ids

    def test_multi_ids(self):
        messages_in = [{'ttl': 120, 'body': 0}, {'ttl': 240, 'body': 1}]
        ids = self.controller.post(self.queue_name, messages_in,
                                   project=self.project,
                                   client_uuid='my_uuid')

        messages_out = self.controller.bulk_get(self.queue_name, ids,
                                                project=self.project)

        for idx, message in enumerate(messages_out):
            self.assertEquals(message['body'], idx)

        self.controller.bulk_delete(self.queue_name, ids,
                                    project=self.project)

        with testing.expect(StopIteration):
            result = self.controller.bulk_get(self.queue_name, ids,
                                              project=self.project)
            next(result)
开发者ID:pombredanne,项目名称:marconi,代码行数:19,代码来源:base.py


示例14: test_expired_messages

    def test_expired_messages(self):
        messages = [{'body': 3.14, 'ttl': 0}]
        client_uuid = uuid.uuid4()

        [msgid] = self.controller.post(self.queue_name, messages,
                                       project=self.project,
                                       client_uuid=client_uuid)

        [msgid] = self.controller.post(self.queue_name, messages,
                                       project=self.project,
                                       client_uuid=client_uuid)

        time.sleep(self.gc_interval)

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get(self.queue_name, msgid,
                                project=self.project)

        stats = self.queue_controller.stats(self.queue_name,
                                            project=self.project)

        self.assertEqual(stats['messages']['free'], 0)
开发者ID:chadlung,项目名称:marconi,代码行数:22,代码来源:base.py


示例15: test_claim_effects

    def test_claim_effects(self):
        _insert_fixtures(self.controller, self.queue_name,
                         project=self.project, client_uuid='my_uuid', num=12)

        def list_messages(include_claimed=None):
            kwargs = {
                'project': self.project,
                'client_uuid': 'my_uuid',
                'echo': True,
            }

            # Properly test default value
            if include_claimed is not None:
                kwargs['include_claimed'] = include_claimed

            interaction = self.controller.list(self.queue_name, **kwargs)

            messages = next(interaction)
            return [msg['id'] for msg in messages]

        messages_before = list_messages(True)

        meta = {'ttl': 70, 'grace': 60}
        another_cid, _ = self.claim_controller.create(self.queue_name, meta,
                                                      project=self.project)

        messages_after = list_messages(True)
        self.assertEqual(messages_before, messages_after)

        messages_excluding_claimed = list_messages()
        self.assertNotEqual(messages_before, messages_excluding_claimed)
        self.assertEqual(2, len(messages_excluding_claimed))

        cid, msgs = self.claim_controller.create(self.queue_name, meta,
                                                 project=self.project)
        [msg1, msg2] = msgs

        # A wrong claim does not ensure the message deletion
        with testing.expect(storage.exceptions.NotPermitted):
            self.controller.delete(self.queue_name, msg1['id'],
                                   project=self.project,
                                   claim=another_cid)

        # Make sure a message can be deleted with a claim
        self.controller.delete(self.queue_name, msg1['id'],
                               project=self.project,
                               claim=cid)

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get(self.queue_name, msg1['id'],
                                project=self.project)

        # Make sure such a deletion is idempotent
        self.controller.delete(self.queue_name, msg1['id'],
                               project=self.project,
                               claim=cid)

        # A non-existing claim does not ensure the message deletion
        self.claim_controller.delete(self.queue_name, cid,
                                     project=self.project)

        with testing.expect(storage.exceptions.NotPermitted):
            self.controller.delete(self.queue_name, msg2['id'],
                                   project=self.project,
                                   claim=cid)
开发者ID:pombredanne,项目名称:marconi,代码行数:65,代码来源:base.py


示例16: test_queue_lifecycle

    def test_queue_lifecycle(self):
        # Test queue creation
        created = self.controller.create('test', project=self.project)
        self.assertTrue(created)

        # Test queue existence
        self.assertTrue(self.controller.exists('test', project=self.project))

        # Test queue retrieval
        interaction = self.controller.list(project=self.project)
        queue = list(next(interaction))[0]
        self.assertEqual(queue['name'], 'test')

        # Test queue metadata retrieval
        metadata = self.controller.get_metadata('test', project=self.project)
        self.assertEqual(metadata, {})

        # Test queue update
        created = self.controller.set_metadata('test', project=self.project,
                                               metadata=dict(meta='test_meta'))

        metadata = self.controller.get_metadata('test', project=self.project)
        self.assertEqual(metadata['meta'], 'test_meta')

        # Touching an existing queue does not affect metadata
        created = self.controller.create('test', project=self.project)
        self.assertFalse(created)

        metadata = self.controller.get_metadata('test', project=self.project)
        self.assertEqual(metadata['meta'], 'test_meta')

        client_uuid = uuid.uuid4()

        # Test queue statistic
        _insert_fixtures(self.message_controller, 'test',
                         project=self.project, client_uuid=client_uuid,
                         num=6)

        # NOTE(kgriffs): We can't get around doing this, because
        # we don't know how the storage drive may be calculating
        # message timestamps (and may not be monkey-patchable).
        time.sleep(1.2)

        _insert_fixtures(self.message_controller, 'test',
                         project=self.project, client_uuid=client_uuid,
                         num=6)

        stats = self.controller.stats('test', project=self.project)
        message_stats = stats['messages']

        self.assertEqual(message_stats['free'], 12)
        self.assertEqual(message_stats['claimed'], 0)
        self.assertEqual(message_stats['total'], 12)

        oldest = message_stats['oldest']
        newest = message_stats['newest']

        self.assertNotEqual(oldest, newest)

        age = oldest['age']
        self.assertThat(age, matchers.GreaterThan(0))

        # NOTE(kgriffs): Ensure is different enough
        # for the next comparison to work.
        soon = timeutils.utcnow() + datetime.timedelta(seconds=60)

        for message_stat in (oldest, newest):
            created_iso = message_stat['created']
            created = timeutils.parse_isotime(created_iso)
            self.assertThat(timeutils.normalize_time(created),
                            matchers.LessThan(soon))

            self.assertIn('id', message_stat)

        self.assertThat(oldest['created'],
                        matchers.LessThan(newest['created']))

        # Test queue deletion
        self.controller.delete('test', project=self.project)

        # Test queue existence
        self.assertFalse(self.controller.exists('test', project=self.project))

        # Test DoesNotExist exception
        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get_metadata('test', project=self.project)

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.set_metadata('test', '{}', project=self.project)
开发者ID:chadlung,项目名称:marconi,代码行数:89,代码来源:base.py


示例17: test_queue_lifecycle

    def test_queue_lifecycle(self):
        # Test Queue Creation
        created = self.controller.create('test', project=self.project)
        self.assertTrue(created)

        # Test Queue Existence
        self.assertTrue(self.controller.exists('test', project=self.project))

        # Test Queue retrieval
        metadata = self.controller.get_metadata('test', project=self.project)
        self.assertEqual(metadata, {})

        # Test Queue Update
        created = self.controller.set_metadata('test', project=self.project,
                                               metadata=dict(meta='test_meta'))

        metadata = self.controller.get_metadata('test', project=self.project)
        self.assertEqual(metadata['meta'], 'test_meta')

        # Touching an existing queue does not affect metadata
        created = self.controller.create('test', project=self.project)
        self.assertFalse(created)

        metadata = self.controller.get_metadata('test', project=self.project)
        self.assertEqual(metadata['meta'], 'test_meta')

        # Test Queue Statistic
        _insert_fixtures(self.message_controller, 'test',
                         project=self.project, client_uuid='my_uuid',
                         num=6)

        # NOTE(kgriffs): We can't get around doing this, because
        # we don't know how the storage drive may be calculating
        # message timestamps (and may not be monkey-patchable).
        time.sleep(1)

        _insert_fixtures(self.message_controller, 'test',
                         project=self.project, client_uuid='my_uuid',
                         num=6)

        stats = self.controller.stats('test', project=self.project)
        message_stats = stats['messages']

        self.assertEqual(message_stats['free'], 12)
        self.assertEqual(message_stats['claimed'], 0)
        self.assertEqual(message_stats['total'], 12)

        oldest = message_stats['oldest']
        newest = message_stats['newest']

        self.assertNotEqual(oldest, newest)

        # NOTE(kgriffs): Ensure "now" is different enough
        # for the next comparison to work.
        timeutils.set_time_override()
        timeutils.advance_time_seconds(10)

        for message_stat in (oldest, newest):
            created_iso = message_stat['created']
            created = timeutils.parse_isotime(created_iso)
            self.assertThat(timeutils.normalize_time(created),
                            matchers.LessThan(timeutils.utcnow()))

            self.assertIn('id', message_stat)

        self.assertThat(oldest['created'],
                        matchers.LessThan(newest['created']))

        # Test Queue Deletion
        self.controller.delete('test', project=self.project)

        # Test Queue Existence
        self.assertFalse(self.controller.exists('test', project=self.project))

        # Test DoesNotExist Exception
        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.get_metadata('test', project=self.project)

        with testing.expect(storage.exceptions.DoesNotExist):
            self.controller.set_metadata('test', '{}', project=self.project)
开发者ID:pombredanne,项目名称:marconi,代码行数:80,代码来源:base.py


示例18: test_race_condition_on_post

    def test_race_condition_on_post(self):
        queue_name = 'marker_test'
        self.queue_controller.create(queue_name)

        expected_messages = [
            {
                'ttl': 60,
                'body': {
                    'event': 'BackupStarted',
                    'backupId': 'c378813c-3f0b-11e2-ad92-7823d2b0f3ce',
                },
            },
            {
                'ttl': 60,
                'body': {
                    'event': 'BackupStarted',
                    'backupId': 'd378813c-3f0b-11e2-ad92-7823d2b0f3ce',
                },
            },
            {
                'ttl': 60,
                'body': {
                    'event': 'BackupStarted',
                    'backupId': 'e378813c-3f0b-11e2-ad92-7823d2b0f3ce',
                },
            },
        ]

        uuid = '97b64000-2526-11e3-b088-d85c1300734c'

        # NOTE(kgriffs): Patch _inc_counter so it is a noop, so that
        # the second time we post, we will get a collision. This simulates
        # what happens when we have parallel requests and the "winning"
        # requests hasn't gotten around to calling _inc_counter before the
        # "losing" request attempts to insert it's batch of messages.
        with mock.patch.object(mongodb.queues.QueueController,
                               '_inc_counter', autospec=True) as method:

            method.return_value = 2
            messages = expected_messages[:1]
            created = list(self.controller.post(queue_name, messages, uuid))
            self.assertEqual(len(created), 1)

            # Force infinite retries
            if testing.RUN_SLOW_TESTS:
                method.return_value = None

                with testing.expect(errors.MessageConflict):
                    self.controller.post(queue_name, messages, uuid)

        created = list(self.controller.post(queue_name,
                                            expected_messages[1:],
                                            uuid))

        self.assertEqual(len(created), 2)

        expected_ids = [m['body']['backupId'] for m in expected_messages]

        interaction = self.controller.list(queue_name, client_uuid=uuid,
                                           echo=True)

        actual_messages = list(next(interaction))
        self.assertEqual(len(actual_messages), len(expected_messages))
        actual_ids = [m['body']['backupId'] for m in actual_messages]

        self.assertEqual(actual_ids, expected_ids)
开发者ID:Sookyung,项目名称:marconi,代码行数:66,代码来源:test_impl_mongodb.py


示例19: test_wrong_type

 def test_wrong_type(self):
     ns = config.namespace('local')
     with testing.expect(config.cfg.Error):
         ns.from_options(opt={})
开发者ID:amitgandhinz,项目名称:marconi,代码行数:4,代码来源:test_config.py



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


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python http.get函数代码示例发布时间:2022-05-27
下一篇:
Python utils.to_json函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

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

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