本文整理汇总了Python中weightless.core.be函数的典型用法代码示例。如果您正苦于以下问题:Python be函数的具体用法?Python be怎么用?Python be使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了be函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: testApacheLog
def testApacheLog(self):
requestHandler = CallTrace("handler", ignoredAttributes=["writeLog", "do_unknown"])
requestHandler.returnValues["handleRequest"] = (f for f in [Yield, okPlainText, "te", callable, "xt"])
stream = StringIO()
handleRequestLog = HandleRequestLog()
handleRequestLog._time = lambda: 1395409143.0
observable = be(
(Observable(), (LogCollector(), (handleRequestLog, (requestHandler,)), (ApacheLogWriter(stream),)))
)
result = asList(
observable.all.handleRequest(
Method="GET",
Client=("127.0.0.1", 1234),
RequestURI="http://example.org/path?key=value",
query="key=value",
path="/path",
Headers={"Referer": "http://meresco.org", "User-Agent": "Meresco-Components Test"},
otherKwarg="value",
)
)
self.assertEquals([Yield, okPlainText, "te", callable, "xt"], result)
self.assertEquals(["handleRequest"], requestHandler.calledMethodNames())
logline = stream.getvalue()
self.assertEquals(
'127.0.0.1 - - [21/Mar/2014:13:39:03 +0000] "GET /path?key=value HTTP/1.0" 200 64 "http://meresco.org" "Meresco-Components Test"\n',
logline,
)
开发者ID:seecr,项目名称:meresco-components,代码行数:30,代码来源:handlerequestlogtest.py
示例2: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.jazz = jazz = OaiJazz(join(self.tempdir, 'jazz'))
self.storage = MultiSequentialStorage(join(self.tempdir, 'sequential-store'))
self.oaipmh = self.getOaiPmh()
self.root = be((Observable(),
(self.oaipmh,
(jazz, ),
(RetrieveToGetDataAdapter(),
(self.storage,)
)
)
))
for i in xrange(20):
identifier = recordId = 'record:id:%02d' % i
metadataFormats = [('oai_dc', 'http://www.openarchives.org/OAI/2.0/oai_dc.xsd', 'http://www.openarchives.org/OAI/2.0/oai_dc/')]
if i >= 10:
metadataFormats.append(('prefix2', 'http://example.org/prefix2/?format=xsd&prefix=2','http://example.org/prefix2/'))
sets = []
if i >= 5:
sets.append(('setSpec%s' % ((i//5)*5), ('' if ((i//5)*5) == 10 else 'setName'))) # empty string becomes 'set <setSpec>'.
if 5 <= i < 10:
sets.append(('hierarchical:set', 'hierarchical set'))
if 10 <= i < 15:
sets.append(('hierarchical', 'hierarchical toplevel only'))
sleep(0.001) # avoid timestamps being equals on VMs
jazz.addOaiRecord(recordId, sets=sets, metadataFormats=metadataFormats)
if i % 5 == 0:
list(compose(jazz.delete(recordId)))
self.storage.addData(identifier=identifier, name='oai_dc', data='<oai_dc:dc xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/" xmlns:dc="http://purl.org/dc/elements/1.1/"><dc:identifier>%s</dc:identifier></oai_dc:dc>' % recordId)
if i >= 10:
self.storage.addData(identifier=identifier, name='prefix2', data='<oai_dc:dc xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/" xmlns:dc="http://purl.org/dc/elements/1.1/"><dc:subject>%s</dc:subject></oai_dc:dc>' % recordId)
开发者ID:seecr,项目名称:meresco-oai,代码行数:33,代码来源:oaipmhtest.py
示例3: testWorksWithRealTransactionScope
def testWorksWithRealTransactionScope(self):
intercept = CallTrace('Intercept', ignoredAttributes=['begin', 'commit', 'rollback'], methods={'add': add})
class MockVenturi(Observable):
def all_unknown(self, message, *args, **kwargs):
self.ctx.tx.locals['id'] = 'an:identifier'
yield self.all.unknown(message, *args, **kwargs)
class MockMultiFielder(Observable):
def add(self, *args, **kwargs):
self.do.addField('field.name', 'MyName')
self.do.addField('field.name', 'AnotherName')
self.do.addField('field.title', 'MyDocument')
yield 'ok'
root = be(
(Observable(),
(TransactionScope(transactionName="xmlDocument"),
(MockVenturi(),
(MockMultiFielder(),
(ResourceManager("xmlDocument"),
(Fields2Xml(partName="partname"),
(intercept,),
)
)
)
)
)
)
)
list(compose(root.all.add('some', 'arguments')))
self.assertEquals(['add'], [m.name for m in intercept.calledMethods])
method = intercept.calledMethods[0]
expectedXml = "<partname><field><name>MyName</name></field><field><name>AnotherName</name></field><field><title>MyDocument</title></field></partname>"
self.assertEquals(((), {'identifier': 'an:identifier', 'partname': 'partname', 'data': expectedXml}), (method.args, method.kwargs))
开发者ID:seecr,项目名称:meresco-components,代码行数:32,代码来源:fields2xmltest.py
示例4: testNonGeneratorMethodMayNeverRaiseGeneratorExceptionsOnMessages
def testNonGeneratorMethodMayNeverRaiseGeneratorExceptionsOnMessages(self):
# any, all, do, call and once
class OddObject(object):
def stopIter(self):
raise StopIteration('Stop!')
def genExit(self):
raise GeneratorExit('Exit!')
dna = (Observable(),
(Transparent(),
(OddObject(),),
)
)
root = be(dna)
# Verify traceback's and wrapped-exception text is ok
try: root.call.stopIter()
except AssertionError, e:
self.assertTrue(str(e).startswith('Non-Generator <bound method OddObject.stopIter of <core.observabletest.OddObject object at 0x'), str(e))
expected = ignoreLineNumbers('''>> should not have raised Generator-Exception:
Traceback (most recent call last):
File "%(__file__)s", line [#], in stopIter
raise StopIteration('Stop!')
StopIteration: Stop!
''' % fileDict)
self.assertTrue(ignoreLineNumbers(str(e)).endswith(expected), str(e))
self.assertFunctionsOnTraceback(
'testNonGeneratorMethodMayNeverRaiseGeneratorExceptionsOnMessages',
'call_unknown', # From Transparent, is supposed to be on the stack to aid retracing the path taken for a messages.
'handleNonGeneratorGeneratorExceptions')
开发者ID:blep,项目名称:weightless-core,代码行数:30,代码来源:observabletest.py
示例5: 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
示例6: testCallViaUnknown
def testCallViaUnknown(self):
class A(object):
def call_unknown(self, message, *args, **kwargs):
return message, args, kwargs
root = be((Observable(), (A(),)))
r = root.call.f(1, a=2)
self.assertEquals(('f', (1,), {'a': 2}), r)
开发者ID:blep,项目名称:weightless-core,代码行数:7,代码来源:observabletest.py
示例7: testLoggedPathsNewStyle
def testLoggedPathsNewStyle(self):
log = CallTrace('log')
def handleRequest(**kwargs):
yield okPlainText
yield 'result'
index = CallTrace('index', methods={'handleRequest':handleRequest})
observable = be((Observable(),
(LogCollector(),
(QueryLogWriter(log=log, scopeNames=('global', 'yesPath')),),
(LogCollectorScope('global'),
(HandleRequestLog(),
(PathFilter('/yes'),
(LogCollectorScope('yesPath'),
(index,),
)
),
(PathFilter('/no'),
(index,),
)
)
)
)
))
result = asString(observable.all.handleRequest(Client=('11.22.33.44', 1234), path='/yes'))
self.assertEquals(okPlainText+'result', result)
result = asString(observable.all.handleRequest(Client=('22.33.44.55', 2345), path='/no'))
self.assertEquals(okPlainText+'result', result)
result = asString(observable.all.handleRequest(Client=('33.44.55.66', 3456), path='/yes'))
self.assertEquals(okPlainText+'result', result)
self.assertEquals(['log', 'log'], log.calledMethodNames())
self.assertEquals(['/yes', '/yes'], [m.kwargs['path'] for m in log.calledMethods])
开发者ID:seecr,项目名称:meresco-components,代码行数:32,代码来源:querylogwritertest.py
示例8: testCreateUser
def testCreateUser(self):
observer = CallTrace()
action = UserActions(dataDir=self.tempdir)
session = {}
dna = be(
(Observable(),
(action,
(observer, )
),
))
self.assertEqual(1, len(action.listUsers()))
response = asString(dna.call.handleRequest(
Method="POST",
path="/user.action/create",
session=session,
Body=urlencode(dict(
redirectUri="/go_here_now",
username="johan",
domain="domein",
password1="password",
password2="password"))))
self.assertEqual(2, len(action.listUsers()))
self.assertTrue("Location: /go_here_now?identifier=johan" in response, response)
self.assertEqual(1, len(observer.calledMethods))
self.assertEqual({}, session)
开发者ID:seecr,项目名称:meresco-harvester,代码行数:26,代码来源:useractionstest.py
示例9: testUpdateUser
def testUpdateUser(self):
action = UserActions(dataDir=self.tempdir)
dna = be(
(Observable(),
(action, ),
))
users = action.listUsers()
users.append(User(username="johan"))
action.saveUsers(users)
self.assertEqual(2, len(action.listUsers()))
self.assertEqual("", action.getUser("johan").organization)
response = asString(dna.call.handleRequest(
Method="POST",
path="/user.action/update",
Body=urlencode(dict(
redirectUri="/go_here_now",
username="johan",
organization="Seecr"))))
self.assertEqual(2, len(action.listUsers()))
self.assertEqual("Seecr", action.getUser("johan").organization)
self.assertTrue("Location: /go_here_now?identifier=johan" in response, response)
response = asString(dna.call.handleRequest(
Method="POST",
path="/user.action/update",
Body=urlencode(dict(
redirectUri="/go_here_now",
username="johan",
organization=""))))
self.assertEqual("", action.getUser("johan").organization)
开发者ID:seecr,项目名称:meresco-harvester,代码行数:32,代码来源:useractionstest.py
示例10: startServer
def startServer(port, stateDir, luceneserverPort, gatewayPort, quickCommit=False, **ignored):
setSignalHandlers()
print 'Firing up Index Server.'
statePath = abspath(stateDir)
writerReactor = Reactor()
writer = writerMain(
writerReactor=writerReactor,
statePath=statePath,
luceneserverPort=luceneserverPort,
gatewayPort=gatewayPort,
quickCommit=quickCommit,
)
writerServer = be(writer)
# Start writer in main (this) thread:
consume(writerServer.once.observer_init())
registerShutdownHandler(statePath=statePath, server=writerServer, reactor=writerReactor, shutdownMustSucceed=False)
print "Ready to rumble at port %s" % port
stdout.flush()
writerReactor.loop()
开发者ID:ChrisMuller80,项目名称:narcis-meresco,代码行数:28,代码来源:indexserver.py
示例11: prepareOaiPmh
def prepareOaiPmh(dataDirs, tempDir, storage, batchSize):
print 'DATADIRS', dataDirs
oaiSuspendRegister = SuspendRegister()
oaiJazz = OaiJazz(tempDir)
oaiJazz.addObserver(oaiSuspendRegister)
oaiJazzOperations = {
'ADD': oaiJazz.addOaiRecord,
'DEL': oaiJazz.deleteOaiRecord
}
for dataDir in dataDirs:
for action, filename, setSpecs in iterOaiData(dataDir):
identifier, metadataPrefix = filename.rsplit('.', 1)
oaiJazzOperations[action](
identifier=identifier,
setSpecs=setSpecs,
metadataPrefixes=[metadataPrefix],
)
storage.addFile(filename, join(dataDir, escapeFilename(filename)))
sleep(0.000001)
oaiJazz.commit()
oaiPmh = be(
(IllegalFromFix(),
(OaiPmh(repositoryName='Mock', adminEmail='[email protected]', supportXWait=True, batchSize=batchSize),
# (LogComponent('OaiPmh'),),
(oaiJazz,),
(oaiSuspendRegister,),
(storage,),
)
)
)
return oaiPmh
开发者ID:seecr,项目名称:dc-erfgeo-enrich,代码行数:32,代码来源:mockoaiserver.py
示例12: testNoQueryInCaseOfGeoLatLong
def testNoQueryInCaseOfGeoLatLong(self):
queries = []
def queryErfGeoApi(query, expectedType=None):
queries.append(dict(query=query, expectedType=expectedType))
raise StopIteration(QUERY_RESULTS)
yield
def toAnnotation(pit, targetUri, query, geoCoordinates=None):
return PitToAnnotation().toAnnotation(pit=pit, targetUri=targetUri, query=query, geoCoordinates=geoCoordinates)
observer = CallTrace('observer', methods={'queryErfGeoApi': queryErfGeoApi, 'toAnnotation': toAnnotation})
top = be(
(Observable(),
(ErfGeoEnrichmentFromSummary(),
(observer,)
)
)
)
summary = makeSummary([], geoLatLong=('51.8', '5.0'))
result = retval(top.any.annotationFromSummary(summary))
self.assertEquals([], queries)
annotationUri, annotation = result
self.assertEquals(ERFGEO_ENRICHMENT_PROFILE.uriFor('uri:target'), annotationUri)
self.assertEquals('51.8', xpathFirst(annotation, '/rdf:RDF/oa:Annotation/oa:hasBody/rdf:Description/geo:lat/text()'))
开发者ID:seecr,项目名称:dc-erfgeo-enrich,代码行数:25,代码来源:erfgeoenrichmentfromsummarytest.py
示例13: testOpenSearchWithoutHtmlAndPort80
def testOpenSearchWithoutHtmlAndPort80(self):
queryTemplate = '/sru?version=1.1&operation=searchRetrieve&query={searchTerms}'
self.auto = be((Autocomplete(
host='localhost',
port=80,
path='/some/path',
templateQuery=queryTemplate,
shortname="Web Search",
description="Use this web search to search something",
defaultLimit=50,
defaultField='lom',
),
(self.observer,),
))
result = asString(self.auto.handleRequest(
path='/path/opensearchdescription.xml',
arguments={}))
header,body = result.split('\r\n'*2)
self.assertTrue("Content-Type: text/xml" in header, header)
self.assertEqualsWS("""<?xml version="1.0" encoding="UTF-8"?>
<OpenSearchDescription xmlns="http://a9.com/-/spec/opensearch/1.1/">
<ShortName>Web Search</ShortName>
<Description>Use this web search to search something</Description>
<Url type="text/xml" method="get" template="http://localhost/sru?version=1.1&operation=searchRetrieve&query={searchTerms}"/>
<Url type="application/x-suggestions+json" template="http://localhost/some/path?prefix={searchTerms}"/>
</OpenSearchDescription>""", body)
开发者ID:seecr,项目名称:meresco-components,代码行数:27,代码来源:autocompletetest.py
示例14: testOnceAssertsNoneReturnValues
def testOnceAssertsNoneReturnValues(self):
# OnceMessage assertion on None: #1a "normal object"
class AnObject(object):
def f(self):
raise StopIteration(1)
yield
# OnceMessage assertion on None: #1b "Observable"
class AnObservable(Observable):
def g(self):
raise StopIteration(1)
yield
root = be(
(Observable(),
(AnObject(),),
(AnObservable(),),
))
composed = compose(root.once.f())
try:
composed.next()
self.fail("Should not happen")
except AssertionError, e:
self.assertTrue("<bound method AnObject.f of <core.observabletest.AnObject object at 0x" in str(e), str(e))
self.assertTrue(">> returned '1'" in str(e), str(e))
开发者ID:blep,项目名称:weightless-core,代码行数:25,代码来源:observabletest.py
示例15: 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
示例16: testTailTakenCareOfWithoutAffectingOriginal
def testTailTakenCareOfWithoutAffectingOriginal(self):
observer = CallTrace('observer', methods={'test': lambda *args, **kwargs: (x for x in [])})
observable = be(
(Observable(),
(XmlXPath(
['/myns:root/myns:path'],
fromKwarg='lxmlNode',
namespaces={'myns': 'http://myns.org/'}
),
(observer, ),
)
)
)
XML = """\
<root xmlns:myns="http://myns.org/" xmlns="http://myns.org/">
<myns:path>
<to>me</to>
</myns:path>\n
</root>"""
lxmlNode = parse(StringIO(XML))
self.assertEquals(XML, lxmltostring(lxmlNode))
list(compose(observable.all.test('een tekst', lxmlNode=lxmlNode)))
self.assertEquals(1, len(observer.calledMethods))
method = observer.calledMethods[0]
self.assertEquals('test', method.name)
self.assertEqualsWS('<myns:path xmlns:myns="http://myns.org/" xmlns="http://myns.org/"><to>me</to></myns:path>', lxmltostring(method.kwargs['lxmlNode']))
self.assertEquals("""\
<myns:path xmlns:myns="http://myns.org/" xmlns="http://myns.org/">
<to>me</to>
</myns:path>""", lxmltostring(method.kwargs['lxmlNode']))
self.assertEquals(XML, lxmltostring(lxmlNode))
开发者ID:seecr,项目名称:meresco-components,代码行数:35,代码来源:xmlxpathtest.py
示例17: testBeTree
def testBeTree(self):
observable = Observable()
child0 = Observable()
child1 = Observable()
root = be((observable, (child0, (child1,))))
self.assertEquals([child0], root._observers)
self.assertEquals([child1], child0._observers)
开发者ID:blep,项目名称:weightless-core,代码行数:7,代码来源:observabletest.py
示例18: 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
示例19: testWorksWithRealTransactionScope
def testWorksWithRealTransactionScope(self):
intercept = CallTrace('Intercept', ignoredAttributes=['begin', 'commit', 'rollback'])
class MockVenturi(Observable):
def all_unknown(self, message, *args, **kwargs):
self.ctx.tx.locals['id'] = 'an:identifier'
yield self.all.unknown(message, *args, **kwargs)
class MockMultiFielder(Observable):
def add(self, *args, **kwargs):
self.do.addField('set', ('setSpec', 'setName'))
self.do.addField('metadataFormat', ('prefix', 'schema', 'namespace'))
yield 'ok'
root = be(
(Observable(),
(TransactionScope(transactionName="oaiRecord"),
(MockVenturi(),
(MockMultiFielder(),
(ResourceManager("oaiRecord"),
(Fields2OaiRecord(),
(intercept,),
)
)
)
)
)
)
)
list(compose(root.all.add('some', 'arguments')))
self.assertEquals(['addOaiRecord'], [m.name for m in intercept.calledMethods])
method = intercept.calledMethods[0]
self.assertEquals(((), {'identifier': 'an:identifier', 'metadataFormats': set([('prefix', 'schema', 'namespace')]), 'sets': set([('setSpec', 'setName')])}), (method.args, method.kwargs))
开发者ID:seecr,项目名称:meresco-oai,代码行数:30,代码来源:fields2oairecordtest.py
示例20: testQuery
def testQuery(self):
def executeQuery(**kwargs):
raise StopIteration(Response(total=42))
yield
index = CallTrace('index',
emptyGeneratorMethods=['echoedExtraRequestData', 'extraResponseData'],
methods=dict(executeQuery=executeQuery))
observable = be((Observable(),
(LogCollector(),
(self.handleRequestLog,
(SruParser(),
(SruHandler(enableCollectLog=True),
(index,)
)
)
),
(self.queryLogWriter,),
)
))
result = asString(observable.all.handleRequest(
Method='GET',
Client=('127.0.0.1', 1234),
arguments={
'version': ['1.2'],
'operation': ['searchRetrieve'],
'query': ['query'],
'maximumRecords': ['0'],
},
path='/path/sru',
otherKwarg='value'))
self.assertTrue('<srw:numberOfRecords>42</srw:numberOfRecords>' in result, result)
self.assertTrue(isfile(join(self.tempdir, '2009-11-02-query.log')))
self.assertEquals('2009-11-02T11:25:37Z 127.0.0.1 0.7K 1.000s 42hits /path/sru maximumRecords=0&operation=searchRetrieve&query=query&recordPacking=xml&recordSchema=dc&startRecord=1&version=1.2\n', open(join(self.tempdir, '2009-11-02-query.log')).read())
开发者ID:seecr,项目名称:meresco-components,代码行数:33,代码来源:srulogtest.py
注:本文中的weightless.core.be函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论