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

Python twistedsupport.succeeded函数代码示例

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

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



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

示例1: test_start_responding

    def test_start_responding(self, token, subdomain, zone_name):
        """
        Calling ``start_responding`` causes an appropriate TXT record to be
        created.
        """
        challenge = self._challenge_factory(token=token)
        response = challenge.response(RSA_KEY_512)
        responder = self._responder_factory(zone_name=zone_name)
        server_name = u'{}.{}'.format(subdomain, zone_name)
        zone = responder._driver.list_zones()[0]

        self.assertThat(zone.list_records(), HasLength(0))
        d = responder.start_responding(server_name, challenge, response)
        self._perform()
        self.assertThat(d, succeeded(Always()))
        self.assertThat(
            zone.list_records(),
            MatchesListwise([
                MatchesStructure(
                    name=EndsWith(u'.' + subdomain),
                    type=Equals('TXT'),
                    )]))

        # Starting twice before stopping doesn't break things
        d = responder.start_responding(server_name, challenge, response)
        self._perform()
        self.assertThat(d, succeeded(Always()))
        self.assertThat(zone.list_records(), HasLength(1))

        d = responder.stop_responding(server_name, challenge, response)
        self._perform()
        self.assertThat(d, succeeded(Always()))
        self.assertThat(zone.list_records(), HasLength(0))
开发者ID:mithrandi,项目名称:txacme,代码行数:33,代码来源:test_challenges.py


示例2: test_cancellation

    def test_cancellation(self, server_name):
        """
        Cancelling the deferred returned by ``issue_cert`` cancels the actual
        issuing process.
        """
        with AcmeFixture() as fixture:
            fixture.service.startService()
            self.assertThat(
                fixture.cert_store.as_dict(),
                succeeded(
                    Not(Contains(server_name))))

            fixture.controller.pause()
            d1 = fixture.service.issue_cert(server_name)
            self.assertThat(d1, has_no_result())
            d2 = fixture.service.issue_cert(server_name)
            self.assertThat(d2, has_no_result())
            self.assertThat(fixture.controller.count(), Equals(1))
            d2.cancel()

            fixture.controller.resume()
            self.assertThat(d1, failed_with(IsInstance(CancelledError)))
            self.assertThat(d2, failed_with(IsInstance(CancelledError)))
            self.assertThat(
                fixture.cert_store.as_dict(),
                succeeded(
                    Not(Contains(server_name))))
开发者ID:mithrandi,项目名称:txacme,代码行数:27,代码来源:test_service.py


示例3: test_sync_app_existing_cert

    def test_sync_app_existing_cert(self):
        """
        When a sync is run and Marathon has an app with a domain label but we
        already have a certificate for that app then a new certificate should
        not be fetched.
        """
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'MARATHON_ACME_0_DOMAIN': 'example.com'
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })
        self.cert_store.store('example.com', 'certcontent')

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(Equals([])))

        # Existing cert unchanged, marathon-lb not notified, but Marathon
        # checked
        assert_that(
            self.fake_marathon_api.check_called_get_apps(), Equals(True))
        assert_that(self.cert_store.as_dict(), succeeded(
            Equals({'example.com': 'certcontent'})))
        assert_that(self.fake_marathon_lb.check_signalled_usr1(),
                    Equals(False))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:30,代码来源:test_service.py


示例4: test_sync_acme_server_failure_acceptable

    def test_sync_acme_server_failure_acceptable(self):
        """
        When a sync is run and we try to issue a certificate for a domain but
        the ACME server returns an error, if that error is of an acceptable
        type then it should be ignored.
        """
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'MARATHON_ACME_0_DOMAIN': 'example.com'
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })
        acme_error = acme_Error(typ='urn:acme:error:rateLimited', detail='bar')
        # Server error takes an ACME error and a treq response...but we don't
        # have a response
        self.txacme_client.issuance_error = txacme_ServerError(
            acme_error, None)

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()

        assert_that(d, succeeded(Equals([None])))
        # Nothing stored, nothing notified
        assert_that(self.cert_store.as_dict(), succeeded(Equals({})))
        assert_that(self.fake_marathon_lb.check_signalled_usr1(),
                    Equals(False))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:30,代码来源:test_service.py


示例5: test_sync_app_label_but_no_domains

    def test_sync_app_label_but_no_domains(self):
        """
        When a sync is run and Marathon has an app and that app has a domain
        label but that label has no domains, then no certificates should be
        fetched and marathon-lb should not be notified.
        """
        # Store an app in Marathon with a marathon-acme domain
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'MARATHON_ACME_0_DOMAIN': '',
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(Equals([])))

        # Nothing stored, nothing notified, but Marathon checked
        assert_that(
            self.fake_marathon_api.check_called_get_apps(), Equals(True))
        assert_that(self.cert_store.as_dict(), succeeded(Equals({})))
        assert_that(self.fake_marathon_lb.check_signalled_usr1(),
                    Equals(False))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:28,代码来源:test_service.py


示例6: test_sync_app_port_group_mismatch

    def test_sync_app_port_group_mismatch(self):
        """
        When a sync is run and Marathon has an app and that app has a matching
        group but mismatching port group, then no certificates should be
        fetched and marathon-lb should not be notified.
        """
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'HAPROXY_0_GROUP': 'internal',
                'HAPROXY_0_VHOST': 'example.com',
                'MARATHON_ACME_0_DOMAIN': 'example.com',
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(Equals([])))

        # Nothing stored, nothing notified, but Marathon checked
        assert_that(
            self.fake_marathon_api.check_called_get_apps(), Equals(True))
        assert_that(self.cert_store.as_dict(), succeeded(Equals({})))
        assert_that(self.fake_marathon_lb.check_signalled_usr1(),
                    Equals(False))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:29,代码来源:test_service.py


示例7: test_sync_app_multiple_domains_multiple_certs_allowed

    def test_sync_app_multiple_domains_multiple_certs_allowed(self):
        """
        When a sync is run and there is an app with a domain label containing
        multiple domains, and ``allow_multiple_certs`` is True, all domains
        are considered.
        """
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'MARATHON_ACME_0_DOMAIN': 'example.com,example2.com'
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })

        marathon_acme = self.mk_marathon_acme(allow_multiple_certs=True)
        d = marathon_acme.sync()
        assert_that(d, succeeded(MatchesListwise([  # Per domain
            is_marathon_lb_sigusr_response,
            is_marathon_lb_sigusr_response,
        ])))

        assert_that(self.cert_store.as_dict(), succeeded(MatchesDict({
            'example.com': Not(Is(None)),
            'example2.com': Not(Is(None)),
        })))

        assert_that(self.fake_marathon_lb.check_signalled_usr1(), Equals(True))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:30,代码来源:test_service.py


示例8: test_sync_app_no_domains

    def test_sync_app_no_domains(self):
        """
        When a sync is run and Marathon has an app but that app has no
        marathon-acme domains, then no certificates should be fetched and
        marathon-lb should not be notified.
        """
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_0_VHOST': 'example.com'
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(Equals([])))

        # Nothing stored, nothing notified, but Marathon checked
        assert_that(
            self.fake_marathon_api.check_called_get_apps(), Equals(True))
        assert_that(self.cert_store.as_dict(), succeeded(Equals({})))
        assert_that(self.fake_marathon_lb.check_signalled_usr1(),
                    Equals(False))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:26,代码来源:test_service.py


示例9: test_sync_app_multiple_ports

    def test_sync_app_multiple_ports(self):
        """
        When a sync is run and there is an app with domain labels for multiple
        ports, then certificates should be fetched for each port.
        """
        # Store an app in Marathon with a marathon-acme domain
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'MARATHON_ACME_0_DOMAIN': 'example.com',
                'MARATHON_ACME_1_DOMAIN': 'example2.com'
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}},
                {'port': 9001, 'protocol': 'tcp', 'labels': {}}
            ]
        })

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(MatchesListwise([  # Per domain
            is_marathon_lb_sigusr_response,
            is_marathon_lb_sigusr_response
        ])))

        assert_that(self.cert_store.as_dict(), succeeded(MatchesDict({
            'example.com': Not(Is(None)),
            'example2.com': Not(Is(None))
        })))

        assert_that(self.fake_marathon_lb.check_signalled_usr1(), Equals(True))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:32,代码来源:test_service.py


示例10: test_sync_app

    def test_sync_app(self):
        """
        When a sync is run and there is an app with a domain label and no
        existing certificate, then a new certificate should be issued for the
        domain. The certificate should be stored in the certificate store and
        marathon-lb should be notified.
        """
        # Store an app in Marathon with a marathon-acme domain
        self.fake_marathon.add_app({
            'id': '/my-app_1',
            'labels': {
                'HAPROXY_GROUP': 'external',
                'MARATHON_ACME_0_DOMAIN': 'example.com'
            },
            'portDefinitions': [
                {'port': 9000, 'protocol': 'tcp', 'labels': {}}
            ]
        })

        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(MatchesListwise([  # Per domain
            is_marathon_lb_sigusr_response
        ])))

        assert_that(self.cert_store.as_dict(), succeeded(MatchesDict({
            'example.com': Not(Is(None))
        })))

        assert_that(self.fake_marathon_lb.check_signalled_usr1(), Equals(True))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:30,代码来源:test_service.py


示例11: test_issue_concurrently

    def test_issue_concurrently(self, server_name):
        """
        Invoking ``issue_cert`` multiple times concurrently for the same name
        will not start multiple issuing processes, only wait for the first
        process to complete.
        """
        with AcmeFixture() as fixture:
            fixture.service.startService()
            self.assertThat(
                fixture.cert_store.as_dict(),
                succeeded(
                    Not(Contains(server_name))))

            fixture.controller.pause()
            d1 = fixture.service.issue_cert(server_name)
            self.assertThat(d1, has_no_result())
            d2 = fixture.service.issue_cert(server_name)
            self.assertThat(d2, has_no_result())
            self.assertThat(fixture.controller.count(), Equals(1))

            fixture.controller.resume()
            self.assertThat(d1, succeeded(Always()))
            self.assertThat(d2, succeeded(Always()))

            self.assertThat(
                fixture.cert_store.as_dict(),
                succeeded(
                    MatchesDict({server_name: Not(Equals([]))})))
开发者ID:mithrandi,项目名称:txacme,代码行数:28,代码来源:test_service.py


示例12: test_cancel_job

    def test_cancel_job(self):
        """
        Tests that scheduler is stopped whenever a port is failing.
        """
        port_out = object()
        port_in = Mock(spec=_port_callback)
        self.flowmap[port_out] = port_in

        run_deferred = self.scheduler.run(self.clock)
        self.scheduler.send('some item', port_out)

        self.assertEquals(len(list(self.scheduler.pending)), 1)

        self.scheduler.stop('bye!')

        self.assertEquals(len(list(self.scheduler.pending)), 1)

        join_deferred = self.scheduler.join()
        self.clock.advance(self.epsilon)
        assert_that(join_deferred, twistedsupport.succeeded(matchers.Always()))
        assert_that(run_deferred, twistedsupport.succeeded(matchers.Equals('bye!')))

        self.assertEquals(len(list(self.scheduler.pending)), 0)

        self.assertEquals(port_in.call_count, 0)
开发者ID:spreadflow,项目名称:spreadflow-core,代码行数:25,代码来源:test_scheduler.py


示例13: test_acme_challenge_ping

 def test_acme_challenge_ping(self):
     """
     When a GET request is made to the ACME challenge path ping endpoint,
     a pong message should be returned.
     """
     response = self.client.get(
         'http://localhost/.well-known/acme-challenge/ping')
     assert_that(response, succeeded(MatchesAll(
         IsJsonResponseWithCode(200),
         After(json_content, succeeded(Equals({'message': 'pong'})))
     )))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:11,代码来源:test_server.py


示例14: test_unicode_keys

 def test_unicode_keys(self, server_name, pem_objects):
     """
     The keys of the dict returned by ``as_dict`` are ``unicode``.
     """
     self.assertThat(
         self.cert_store.store(server_name, pem_objects),
         succeeded(Is(None)))
     self.assertThat(
         self.cert_store.as_dict(),
         succeeded(AfterPreprocessing(
             methodcaller('keys'),
             AllMatch(IsInstance(unicode)))))
开发者ID:mithrandi,项目名称:txacme,代码行数:12,代码来源:test_store.py


示例15: test_health_handler_unset

 def test_health_handler_unset(self):
     """
     When a GET request is made to the health endpoint, and the health
     handler hasn't been set, a 501 status code should be returned together
     with a JSON message that explains that the handler is not set.
     """
     response = self.client.get('http://localhost/health')
     assert_that(response, succeeded(MatchesAll(
         IsJsonResponseWithCode(501),
         After(json_content, succeeded(Equals({
             'error': 'Cannot determine service health: no handler set'
         })))
     )))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:13,代码来源:test_server.py


示例16: test_delegates_to_agent_for_location

    def test_delegates_to_agent_for_location(self, agent):
        """
        When a request is made using the agent, the added agents are delegated
        to based on the URI location/authority.
        """
        agent.add_agent(b'foo:8080', DummyAgent())
        agent.add_agent(b'bar:8080', FailingAgent(RuntimeError('bar')))
        agent.add_agent(b'foo:9090', FailingAgent(RuntimeError('9090')))

        d = agent.request(b'GET', b'http://foo:8080')
        assert_that(d, succeeded(MatchesListwise([
            MatchesListwise([Equals(b'GET'), Equals(b'http://foo:8080')]),
            MatchesDict({'headers': Is(None), 'bodyProducer': Is(None)})
        ])))

        # Scheme doesn't matter
        d = agent.request(b'GET', b'https://foo:8080')
        assert_that(d, succeeded(MatchesListwise([
            MatchesListwise([Equals(b'GET'), Equals(b'https://foo:8080')]),
            MatchesDict({'headers': Is(None), 'bodyProducer': Is(None)})
        ])))

        # Path doesn't matter
        d = agent.request(b'GET', b'http://foo:8080/bar/baz')
        assert_that(d, succeeded(MatchesListwise([
            MatchesListwise([
                Equals(b'GET'), Equals(b'http://foo:8080/bar/baz')]),
            MatchesDict({'headers': Is(None), 'bodyProducer': Is(None)})
        ])))

        # Hostname *does* matter
        d = agent.request(b'GET', b'http://bar:8080')
        assert_that(d, failed(MatchesStructure(value=MatchesAll(
            IsInstance(RuntimeError),
            MatchesPredicate(str, Equals('bar'))
        ))))

        # Port *does* matter
        d = agent.request(b'GET', b'http://foo:9090')
        assert_that(d, failed(MatchesStructure(value=MatchesAll(
            IsInstance(RuntimeError),
            MatchesPredicate(str, Equals('9090'))
        ))))

        # Other args passed through
        d = agent.request(b'GET', b'http://foo:8080', 'bar', 'baz')
        assert_that(d, succeeded(MatchesListwise([
            MatchesListwise([Equals(b'GET'), Equals(b'http://foo:8080')]),
            MatchesDict(
                {'headers': Equals('bar'), 'bodyProducer': Equals('baz')})
        ])))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:51,代码来源:test_test_helpers.py


示例17: test_health_unhealthy

    def test_health_unhealthy(self):
        """
        When a GET request is made to the health endpoint, and the health
        handler reports that the service is unhealthy, a 503 status code should
        be returned together with the JSON message from the handler.
        """
        self.server.set_health_handler(
            lambda: Health(False, {'error': "I'm sad :("}))

        response = self.client.get('http://localhost/health')
        assert_that(response, succeeded(MatchesAll(
            IsJsonResponseWithCode(503),
            After(json_content, succeeded(Equals({'error': "I'm sad :("})))
        )))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:14,代码来源:test_server.py


示例18: test_blank_cert

 def test_blank_cert(self, server_name):
     """
     An empty certificate file will be treated like an expired certificate.
     """
     with AcmeFixture(certs={server_name: []}) as fixture:
         fixture.service.startService()
         self.assertThat(
             fixture.service.when_certs_valid(),
             succeeded(Always()))
         self.assertThat(
             fixture.cert_store.as_dict(),
             succeeded(
                 MatchesDict({server_name: Not(Equals([]))})))
         self.assertThat(fixture.responder.challenges, HasLength(0))
开发者ID:habnabit,项目名称:txacme,代码行数:14,代码来源:test_service.py


示例19: test_sync_no_apps

    def test_sync_no_apps(self):
        """
        When a sync is run and Marathon has no apps for us then no certificates
        should be fetched and marathon-lb should not be notified.
        """
        marathon_acme = self.mk_marathon_acme()
        d = marathon_acme.sync()
        assert_that(d, succeeded(Equals([])))

        # Nothing stored, nothing notified, but Marathon checked
        assert_that(
            self.fake_marathon_api.check_called_get_apps(), Equals(True))
        assert_that(self.cert_store.as_dict(), succeeded(Equals({})))
        assert_that(self.fake_marathon_lb.check_signalled_usr1(),
                    Equals(False))
开发者ID:praekeltfoundation,项目名称:marathon-acme,代码行数:15,代码来源:test_service.py


示例20: test_insert

 def test_insert(self, server_name, pem_objects):
     """
     Inserting an entry causes the same entry to be returned by ``get`` and
     ``as_dict``.
     """
     self.assertThat(
         self.cert_store.store(server_name, pem_objects),
         succeeded(Is(None)))
     self.assertThat(
         self.cert_store.get(server_name),
         succeeded(Equals(pem_objects)))
     self.assertThat(
         self.cert_store.as_dict(),
         succeeded(ContainsDict(
             {server_name: Equals(pem_objects)})))
开发者ID:mithrandi,项目名称:txacme,代码行数:15,代码来源:test_store.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python lockstatus.get_status函数代码示例发布时间:2022-05-27
下一篇:
Python helpers.get_output_path函数代码示例发布时间: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