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

Python core.compose函数代码示例

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

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



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

示例1: testAnyOrCallCallsFirstImplementer

    def testAnyOrCallCallsFirstImplementer(self):
        class A(object):
            def f(self):
                raise StopIteration(A.f)
                yield
            def f_sync(self):
                return A.f
        class B(object):
            def f(self):
                raise StopIteration(B.f)
                yield
            def f_sync(self):
                return B.f
            def g(self):
                raise StopIteration(B.g)
                yield
            def g_sync(self):
                return B.g
        root = be((Observable(), (A(),), (B(),)))

        try:
            compose(root.any.f()).next()
            self.fail('Should not happen')
        except StopIteration, e:
            self.assertEquals((A.f,), e.args)
开发者ID:blep,项目名称:weightless-core,代码行数:25,代码来源:observabletest.py


示例2: testDeleteRecord

    def testDeleteRecord(self):
        rdf = """<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
        <rdf:Description>
            <rdf:type>uri:testDelete</rdf:type>
        </rdf:Description>
    </rdf:RDF>"""
        jenaClient = HttpClient(host='localhost', port=self.jenaPort, synchronous=True)
        list(compose(jenaClient.add('uri:identifier', rdf)))

        json = self.query('SELECT ?x WHERE {?x ?y "uri:testDelete"}')
        self.assertEquals(1, len(json['results']['bindings']))

        rdfUpdated = """<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
        <rdf:Description>
            <rdf:type>uri:testDeleteUpdated</rdf:type>
        </rdf:Description>
    </rdf:RDF>"""
        list(compose(jenaClient.add('uri:identifier', rdfUpdated)))

        json = self.query('SELECT ?x WHERE {?x ?y "uri:testDelete"}')
        self.assertEquals(0, len(json['results']['bindings']))
        json = self.query('SELECT ?x WHERE {?x ?y "uri:testDeleteUpdated"}')
        self.assertEquals(1, len(json['results']['bindings']))

        list(compose(jenaClient.delete('uri:identifier')))
        json = self.query('SELECT ?x WHERE {?x ?y "uri:testDelete"}')
        self.assertEquals(0, len(json['results']['bindings']))
        json = self.query('SELECT ?x WHERE {?x ?y "uri:testDeleteUpdated"}')
        self.assertEquals(0, len(json['results']['bindings']))

        list(compose(jenaClient.add('uri:record', rdf)))
        json = self.query('SELECT ?x WHERE {?x ?y "uri:testDelete"}')
        self.assertEquals(1, len(json['results']['bindings']))
开发者ID:seecr,项目名称:meresco-jena,代码行数:33,代码来源:jenatest.py


示例3: testValidXml

    def testValidXml(self):
        component = SruParser()
        sruHandler = SruHandler()
        component.addObserver(sruHandler)
        observer = CallTrace('observer')
        sruHandler.addObserver(observer)
        response = Response(total=2, hits=[Hit('id0'), Hit('id1')])
        def executeQuery(**kwargs):
            raise StopIteration(response)
            yield
        def retrieveData(**kwargs):
            raise StopIteration('<bike/>')
            yield
        observer.methods['executeQuery'] = executeQuery
        observer.returnValues['echoedExtraRequestData'] = (f for f in [])
        observer.returnValues['extraResponseData'] = (f for f in [])
        observer.methods['extraRecordData'] = lambda hit: (f for f in [])
        observer.methods['retrieveData'] = retrieveData

        result = ''.join(compose(component.handleRequest(arguments={'version':['1.1'], 'query': ['aQuery'], 'operation':['searchRetrieve']})))
        header, body = result.split('\r\n'*2)
        assertValid(body, join(schemasPath, 'srw-types1.2.xsd'))
        self.assertTrue('<bike/>' in body, body)

        result = ''.join(compose(component.handleRequest(arguments={'version':['1.1'], 'operation':['searchRetrieve']})))
        header, body = result.split('\r\n'*2)
        assertValid(body, join(schemasPath, 'srw-types1.2.xsd'))
        self.assertTrue('diagnostic' in body, body)
开发者ID:seecr,项目名称:meresco-components,代码行数:28,代码来源:sruhandlertest.py


示例4: testOnceCalledMethodsMustResultInAGeneratorOrComposeOrNone

    def testOnceCalledMethodsMustResultInAGeneratorOrComposeOrNone(self):
        callLog = []
        class MyObserver(Observable):
            def __init__(self):
                Observable.__init__(self)
                self.generatorReturningCallable = partial(lambda arg: (x for x in 'a'), arg='partialed')

            def noGeneratorFunction(self):
                callLog.append('function called')

            def valueReturningFunction(self):
                return 42

            def composedGenerator(self):
                return compose(x for x in 'a')

        once = MyObserver()
        dna = \
            (Observable(),
                (once,),
            )
        root = be(dna)
        list(compose(root.once.noGeneratorFunction()))
        self.assertEquals(['function called'], callLog)

        try:
            list(compose(root.once.valueReturningFunction()))
            self.fail("Should have gotten AssertionError because of unexpected return value")
        except AssertionError, e:
            self.assertEquals("<bound method MyObserver.valueReturningFunction of MyObserver(name=None)> returned '42'", str(e))
开发者ID:blep,项目名称:weightless-core,代码行数:30,代码来源:observabletest.py


示例5: testUndirectedObserverMessagingIsUnaffectedByObserverName

    def testUndirectedObserverMessagingIsUnaffectedByObserverName(self):
        observable = Observable()
        called = []
        class A(Observable):
            def method(this):
                called.append(("A", this.observable_name()))
                return
                yield
        
        class B(Observable):
            def method(this):
                called.append(("B", this.observable_name()))
                return
                yield

        observable.addObserver(A("name"))
        observable.addObserver(A().observable_setName("anothername"))
        observable.addObserver(B("anothername"))
        observable.addObserver(B())

        list(compose(observable.all.method()))
        
        self.assertEquals([("A", "name"), 
            ("A", "anothername"), 
            ("B", "anothername"), 
            ("B", None)], called)
        del called[:]

        list(compose(observable.all["name"].method()))
        self.assertEquals([("A", "name")], called)
开发者ID:blep,项目名称:weightless-core,代码行数:30,代码来源:observabledirectedmessagingtest.py


示例6: testOneTransactionPerGenerator

 def testOneTransactionPerGenerator(self):
     txId = []
     class MyTxParticipant(Observable):
         def doSomething(self):
             txId.append(self.ctx.tx.getId())
             yield 'A'
             txId.append(self.ctx.tx.getId())
             yield 'B'
     dna = \
         (Observable(),
             (TransactionScope('name'),
                 (MyTxParticipant(),)
             )
         )
     body = be(dna)
     scope1 = compose(body.all.doSomething())
     scope2 = compose(body.all.doSomething())
     scope1.next()
     scope2.next()
     scope1.next()
     scope2.next()
     self.assertTrue(txId[0] != txId[1])
     self.assertTrue(txId[1] > 0)
     self.assertTrue(txId[0] > 0)
     self.assertEquals(txId[0], txId[2])
     self.assertEquals(txId[1], txId[3])
开发者ID:seecr,项目名称:meresco-core,代码行数:26,代码来源:transactiontest.py


示例7: testOnceInternalsNotOnTracebackUnlessAssertsAndThenOnlyOnce

    def testOnceInternalsNotOnTracebackUnlessAssertsAndThenOnlyOnce(self):
        class OnceRaiser(object):
            def raisesOnCall(self):
                raise BaseException('Boom')
            def raisesOnCallGenerator(self):
                raise BaseException('Ka-Boom')
                yield

        dna = (Observable(),              # called-from
            (Observable(),                # 1
                (Observable(),            # 2
                    (Observable(),        # 3
                        (OnceRaiser(),),  # target
                    )
                )
            )
        )
        root = be(dna)

        try:
            list(compose(root.once.raisesOnCallGenerator()))
        except BaseException:
            self.assertFunctionsOnTraceback('testOnceInternalsNotOnTracebackUnlessAssertsAndThenOnlyOnce', 'raisesOnCallGenerator')
        else:
            self.fail('Should not happen')

        try:
            list(compose(root.once.raisesOnCall()))
        except BaseException:
            self.assertFunctionsOnTraceback('testOnceInternalsNotOnTracebackUnlessAssertsAndThenOnlyOnce', 'raisesOnCall')
        else:
            self.fail('Should not happen')
开发者ID:blep,项目名称:weightless-core,代码行数:32,代码来源:observabletest.py


示例8: testRequestScopeIsPerRequest

    def testRequestScopeIsPerRequest(self):
        class MyObserver(Observable):
            def handleRequest(self, key, value, *args, **kwargs):
                self.do.setArg(key, value)
                yield self.call.getArg()
        class SetArgObserver(Observable):
            def setArg(self, key, value):
                self.ctx.requestScope[key] = value
        class GetArgObserver(Observable):
            def getArg(self):
                return ';'.join('%s=%s' % (k,v) for k,v in self.ctx.requestScope.items())

        dna = be((Observable(),
            (RequestScope(),
                (MyObserver(),
                    (SetArgObserver(),),
                    (GetArgObserver(),)
                )
            )
        ))

        result0 = list(compose(dna.all.handleRequest("key0", "value0")))
        result1 = list(compose(dna.all.handleRequest("key1", "value1")))

        self.assertEquals(['key0=value0'], result0)
        self.assertEquals(['key1=value1'], result1)
开发者ID:seecr,项目名称:meresco-components,代码行数:26,代码来源:requestscopetest.py


示例9: testDirectedMessagesCanAlsoBeAcceptedByObjects

    def testDirectedMessagesCanAlsoBeAcceptedByObjects(self):
        observable = Observable()
        called = []
        class Y(object):
            def method(this):
                called.append("Y")
                return
                yield
            def observable_name(this):
                return 'name'
        class Z(object):
            def method(this):
                called.append("Z")
                return
                yield
        observable.addObserver(Y())
        observable.addObserver(Z())

        list(compose(observable.all["name"].method()))
        
        self.assertEquals(['Y'], called)

        del called[:]

        list(compose(observable.all.method()))

        self.assertEquals(['Y', "Z"], called)

        del called[:]

        list(compose(observable.all["other"].method()))

        self.assertEquals([], called)
开发者ID:blep,项目名称:weightless-core,代码行数:33,代码来源:observabledirectedmessagingtest.py


示例10: testShouldValidateHashOnUpdate

    def testShouldValidateHashOnUpdate(self):
        updatePath = '/service/v2/update'
        hash = serviceUpdateHash(secret='guessme!', identifier='cc635329-c089-41a8-91be-2a4554851515', type='srv', ipAddress='127.0.0.1', infoport=1234)
        bodyArgs = {
            'identifier': 'cc635329-c089-41a8-91be-2a4554851515',
            'type': 'srv',
            'ipAddress': '127.0.0.1',
            'infoport': '1234',
            'hash': hash,
            'data': dumps({'VERSION': '2.718281828'}),
        }
        postBody = urlencode(bodyArgs)
        result = ''.join(compose(self.dna.all.handleRequest(
            path=updatePath,
            Method='POST',
            arguments={},
            Body=postBody,
        )))
        header, body = httpSplit(result)
        self.assertTrue('HTTP/1.0 200', header)

        bodyArgs['hash'] = 'wrong'
        postBody = urlencode(bodyArgs)
        result = ''.join(compose(self.dna.all.handleRequest(
            path=updatePath,
            Method='POST',
            arguments={},
            Body=postBody,
        )))
        header, body = httpSplit(result)
        self.assertTrue('HTTP/1.0 400', header)
        self.assertEquals('Hash does not match expected hash.', body)
开发者ID:seecr,项目名称:meresco-distributed,代码行数:32,代码来源:servicehandlertest.py


示例11: testWithSorting

    def testWithSorting(self):
        mockData = {
            'yearAndMonth': [('2008-01',1),('2008-02',2),('2007-12',11)],
            'year': [('2008',13),('2003',10),('2005',9), ('2007', 15)]
        }
        drilldown = CallTrace('Drilldown')
        def doDrilldown(bitMatrixRow, fieldNamesAndMaxResults):
            levelField, levelMax, levelSorted = fieldNamesAndMaxResults[0]
            data = mockData[levelField]
            if levelSorted:
                data = sorted(data, cmp=lambda (term0, card0), (term1, card1): cmp(card1, card0))
            if levelMax > 0:
                data = data[:levelMax]
            raise StopIteration(iter([(levelField, iter(data))]))
            yield
        drilldown.drilldown = doDrilldown

        multi = MultiLevelDrilldown({'date':[('yearAndMonth', 2, False), ('year', 3, True)]})
        multi.addObserver(drilldown)
        result = list(compose(multi.multiLevelDrilldown('bitMatrixRow', ['date'])))
        self.assertEquals([(('date', 'year'), [('2007', 15), ('2008', 13), ('2003', 10)])], result)

        multi = MultiLevelDrilldown({'date':[('yearAndMonth', 4, False), ('year', 3, False)]})
        multi.addObserver(drilldown)

        result = list(compose(multi.multiLevelDrilldown('bitMatrixRow', ['date'])))
        self.assertEquals([(('date', 'yearAndMonth'), [('2008-01',1),('2008-02',2),('2007-12',11)])], result)
开发者ID:seecr,项目名称:meresco-components,代码行数:27,代码来源:multileveldrilldowntest.py


示例12: testResolveCallStackVariables

    def testResolveCallStackVariables(self):
        do_result = []
        call_result = []
        class StackVarHolder(Observable):
            def all_unknown(self, message, *args, **kwargs):
                __callstack_var_myvar__ = []
                yield self.all.unknown(message, *args, **kwargs)
                yield __callstack_var_myvar__

            def any_unknown(self, message, *args, **kwargs):
                __callstack_var_myvar__ = []
                yield self.any.unknown(message, *args, **kwargs)
                raise StopIteration(__callstack_var_myvar__)

            def call_unknown(self, message, *args, **kwargs):
                __callstack_var_myvar__ = []
                call_result.append(__callstack_var_myvar__)
                return self.call.unknown(message, *args, **kwargs)

            def do_unknown(self, message, *args, **kwargs):
                __callstack_var_myvar__ = []
                do_result.append(__callstack_var_myvar__)
                self.do.unknown(message, *args, **kwargs)

        class StackVarUser(Observable):
            def useVariableAll(self):
                self.ctx.myvar.append('Thingy')
                yield 'stuffed'

            def useVariableAny(self):
                self.ctx.myvar.append('Thingy')
                return
                yield

            def useVariableCall(self):
                self.ctx.myvar.append('Thingy')
                return 'called'

            def useVariableDo(self):
                self.ctx.myvar.append('Thingy')

        dna = \
            (Observable(),
                (StackVarHolder(),
                    (Transparent(),
                        (StackVarUser(),)
                    )
                )
            )
        root = be(dna)
        self.assertEquals(['stuffed', ['Thingy']], list(compose(root.all.useVariableAll())))

        composed = compose(root.any.useVariableAny())
        try:
            while True:
                composed.next()
        except StopIteration, e:
            self.assertEquals((['Thingy'],), e.args)
开发者ID:seecr,项目名称:meresco-core,代码行数:58,代码来源:observabletest.py


示例13: testCallCompose

 def testCallCompose(self):
     try:
         compose()
         self.fail()
     except TypeError, e:
         self.assertTrue(
                 "compose() takes at least 1 argument (0 given)" in str(e)
                 or # (python 2.5/2.6 C-API differences)
                 "Required argument 'initial' (pos 1) not found" in str(e))
开发者ID:blep,项目名称:weightless-core,代码行数:9,代码来源:composetest.py


示例14: testAnyViaUnknown

 def testAnyViaUnknown(self):
     class A(object):
         def any_unknown(self, message, *args, **kwargs):
             raise StopIteration((message, args, kwargs), )
             yield
     root = be((Observable(), (A(),)))
     try: compose(root.any.f(1, a=2)).next()
     except StopIteration, e: r = e.args[0]
     self.assertEquals(('f', (1,), {'a': 2}), r)
开发者ID:blep,项目名称:weightless-core,代码行数:9,代码来源:observabletest.py


示例15: testShouldAllowPostsOnly

    def testShouldAllowPostsOnly(self):
        result = ''.join(compose(self.dna.all.handleRequest(path='/service/v2/update', arguments={}, Method='GET')))
        header, body = httpSplit(result)
        self.assertEquals('HTTP/1.0 405 Method Not Allowed', header)
        self.assertEquals('', body)

        result = ''.join(compose(self.dna.all.handleRequest(path='/service/v2/update', arguments={}, Method='HEAD')))
        header, body = httpSplit(result)
        self.assertEquals('HTTP/1.0 405 Method Not Allowed', header)
        self.assertEquals('', body)
开发者ID:seecr,项目名称:meresco-distributed,代码行数:10,代码来源:servicehandlertest.py


示例16: testUpdateIpFilter

    def testUpdateIpFilter(self):
        observer = CallTrace(methods={'handleRequest': handleRequest})
        ipf = IpFilter(allowedIps=['192.168.1.1'], allowedIpRanges=[('10.0.0.1', '10.0.0.2'), '2001:41c8:10:7b::/64'])

        dna = be(
            (Observable(),
                (ipf,
                    (observer,)
                )
            )
        )

        list(compose(dna.all.handleRequest(Client=('127.0.0.1',), Headers={})))
        list(compose(dna.all.handleRequest(Client=('10.0.0.10',), Headers={})))
        self.assertEquals(0, len(observer.calledMethods))
        list(compose(dna.all.handleRequest(Client=('192.168.1.1',), Headers={})))
        self.assertEquals(1, len(observer.calledMethods))
        del observer.calledMethods[:]

        list(compose(dna.all.handleRequest(Client=('2001:41c8:10:7b:aa:6:0:2', ), Headers={})))
        self.assertEquals(1, len(observer.calledMethods))
        del observer.calledMethods[:]

        ipf.updateIps(ipAddresses=['127.0.0.1'], ipRanges=[('10.0.0.1', '10.0.0.255'), '2001:41c8:10:7c::/64'])
        list(compose(dna.all.handleRequest(Client=('192.168.1.1',), Headers={})))
        self.assertEquals(0, len(observer.calledMethods))
        list(compose(dna.all.handleRequest(Client=('127.0.0.1',), Headers={})))
        list(compose(dna.all.handleRequest(Client=('10.0.0.10',), Headers={})))
        self.assertEquals(2, len(observer.calledMethods))

        list(compose(dna.all.handleRequest(Client=('2001:41c8:10:7b:aa:6:0:2', ), Headers={})))
        self.assertEquals(2, len(observer.calledMethods))
        list(compose(dna.all.handleRequest(Client=('2001:41c8:10:7c:aa:6:0:2', ), Headers={})))
        self.assertEquals(3, len(observer.calledMethods))
开发者ID:seecr,项目名称:meresco-components,代码行数:34,代码来源:ipfiltertest.py


示例17: testParseAndSetAndRemoveSessionVars2

 def testParseAndSetAndRemoveSessionVars2(self):
     arguments = {}
     def handleRequest(session=None, *args, **kwargs):
         arguments.update(session)
         yield 'goodbye'
     self.observer.handleRequest = handleRequest
     session = {}
     list(compose(self.argsInSession.handleRequest(session=session, arguments={'aap': ["+'noot'"]})))
     self.assertEquals( ['noot'], arguments['aap'])
     list(compose(self.argsInSession.handleRequest(session=session, arguments={'aap': ["-'noot'"]})))
     self.assertEquals( [], arguments['aap'])
开发者ID:seecr,项目名称:meresco-components,代码行数:11,代码来源:argumentsinsessiontest.py


示例18: testValid

    def testValid(self):
        validXml = '<lom xmlns="http://ltsc.ieee.org/xsd/LOM"/>'
        self.assertEquals(['done'], list(compose(self.observable.all.someMethod(parse(StringIO(validXml))))))
        self.assertEquals(['done'], list(compose(self.observable.any.someMethod(parse(StringIO(validXml))))))

        self.interceptor.calledMethods.reset()
        self.observable.do.someMethod(parse(StringIO(validXml)))
        self.assertEquals(['do_unknown'], [m.name for m in self.interceptor.calledMethods])

        self.interceptor.calledMethods.reset()
        self.assertEquals('done', self.observable.call.someMethod(parse(StringIO(validXml))))
开发者ID:seecr,项目名称:meresco-components,代码行数:11,代码来源:validatetest.py


示例19: testParse2

    def testParse2(self):
        xmlString = """<tag><content>contents</content></tag>"""
        self.observable.do.add(identifier="id", partname="partName", data=xmlString)
        self.observable.call.add(identifier="id", partname="partName", data=xmlString)
        self.observer.methods['add'] = lambda **kwargs: (x for x in [])
        list(compose(self.observable.all.add(identifier="id", partname="partName", data=xmlString)))
        list(compose(self.observable.any.add(identifier="id", partname="partName", data=xmlString)))

        self.assertEquals(4, len(self.observer.calledMethods))
        for i in range(4):
            xmlNode = self.observer.calledMethods[i].kwargs['lxmlNode']
            self.assertEqualsLxml(XML(xmlString), xmlNode)
开发者ID:seecr,项目名称:meresco-components,代码行数:12,代码来源:xmlpumptest.py


示例20: testResourceManagerHandlesAttributeError

 def testResourceManagerHandlesAttributeError(self):
     class Resource(object):
         def beginTransaction(self):
             raise StopIteration(object())
             yield
     rm = ResourceManager('transaction')
     rm.addObserver(Resource())
     __callstack_var_tx__ = CallTrace('TransactionScope')
     list(compose(rm.begin('transaction')))
     try:
         list(compose(rm.all.unknown('doesnotexist')))
     except AttributeError:
         self.fail('ResourceManager must ignore unknown methods.')
开发者ID:seecr,项目名称:meresco-core,代码行数:13,代码来源:transactiontest.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python core.consume函数代码示例发布时间:2022-05-26
下一篇:
Python core.be函数代码示例发布时间:2022-05-26
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

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

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

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