本文整理汇总了Python中seecr.test.SeecrTestCase类的典型用法代码示例。如果您正苦于以下问题:Python SeecrTestCase类的具体用法?Python SeecrTestCase怎么用?Python SeecrTestCase使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了SeecrTestCase类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: 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
示例2: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.observer1 = CallTrace(
'observer1',
emptyGeneratorMethods=['message'],
returnValues={
'function': 41,
'gen': (i for i in [41]),
'noop': None
}
)
self.observer2 = object()
self.observer3 = CallTrace(
'observer3',
emptyGeneratorMethods=['message'],
returnValues={
'function': 42,
'gen': (i for i in [42]),
'noop': None
}
)
self.dna = be((Observable(),
(FilterMessages(disallowed=['message', 'function', 'gen', 'noop']),
(self.observer1,)
),
(FilterMessages(allowed=['message', 'function', 'gen', 'noop']),
(self.observer2,)
),
(FilterMessages(allowed=['message', 'function', 'gen', 'noop']),
(self.observer3,)
)
))
开发者ID:seecr,项目名称:meresco-components,代码行数:32,代码来源:filtermessagestest.py
示例3: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.srw = Srw()
self.sruParser = SruParser()
self.sruHandler = SruHandler()
self.srw.addObserver(self.sruParser)
self.sruParser.addObserver(self.sruHandler)
self.response = StopIteration(Response(total=1, hits=[Hit('0')]))
def executeQuery(**kwargs):
raise self.response
yield
def retrieveData(**kwargs):
raise StopIteration('data')
yield
self.observer = CallTrace(
methods={
'executeQuery': executeQuery,
'retrieveData': retrieveData
},
emptyGeneratorMethods=[
'extraResponseData',
'echoedExtraRequestData',
'additionalDiagnosticDetails',
'extraRecordData'
])
self.sruHandler.addObserver(self.observer)
开发者ID:seecr,项目名称:meresco-components,代码行数:27,代码来源:srwtest.py
示例4: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.stateDir = join(self.tempdir, "state")
self.logDir = join(self.tempdir, "log")
self.domainId = "adomain"
makedirs(join(self.stateDir, self.domainId))
repoId1LogDir = join(self.logDir, self.domainId, "invalid", "repoId1")
repoId2LogDir = join(self.logDir, self.domainId, "invalid", escapeFilename("repoId/2"))
makedirs(repoId1LogDir)
makedirs(repoId2LogDir)
open(join(repoId1LogDir, "invalidId1"), 'w').write("<diagnostic>ERROR1</diagnostic>")
open(join(repoId1LogDir, "invalidId&2"), 'w').write("<diagnostic>ERROR2</diagnostic>")
open(join(repoId2LogDir, escapeFilename("invalidId/3")), 'w').write("<diagnostic>ERROR3</diagnostic>")
open(join(self.stateDir, self.domainId, "repoId1_invalid.ids"), 'w').write("invalidId1\ninvalidId&2")
open(join(self.stateDir, self.domainId, escapeFilename("repoId/2_invalid.ids")), 'w').write("invalidId/3")
open(join(self.stateDir, self.domainId, "repoId3_invalid.ids"), 'w').write("")
self.status = RepositoryStatus(self.logDir, self.stateDir)
observer = CallTrace("HarvesterData")
observer.returnValues["getRepositoryGroupIds"] = ["repoGroupId1", "repoGroupId2"]
def getRepositoryIds(domainId, repositoryGroupId):
if repositoryGroupId == "repoGroupId1":
return ["repoId1", "repoId/2"]
return ["repoId3", "anotherRepoId"]
observer.methods["getRepositoryIds"] = getRepositoryIds
def getRepositoryGroupId(domainId, repositoryId):
return 'repoGroupId1' if repositoryId in ['repoId1', 'repoId/2'] else 'repoGroupId2'
observer.methods["getRepositoryGroupId"] = getRepositoryGroupId
self.status.addObserver(observer)
开发者ID:seecr,项目名称:meresco-harvester,代码行数:28,代码来源:repositorystatustest.py
示例5: setUp
def setUp(self):
SeecrTestCase.setUp(self)
ts = [(1 + i*0.1) for i in xrange(100)]
def timeNow():
return ts.pop(0)
self._timeNow = timeNow
class MockHit(object):
def __init__(self, id):
self.id = id
self.total = 2
self.hits = [1,2]
self.drilldownData = None
def executeQuery(*args, **kwargs):
result = LuceneResponse(
total=self.total,
hits=[MockHit('id:%s' % i) for i in self.hits],
queryTime=30
)
if self.drilldownData:
result.drilldownData = self.drilldownData
raise StopIteration(result)
yield
def retrieveData(identifier, name):
raise StopIteration({'identifier':identifier, 'name': name})
yield
self.observer = CallTrace(methods=dict(
executeQuery=executeQuery,
retrieveData=retrieveData))
self._buildDna()
开发者ID:seecr,项目名称:meresco-components,代码行数:32,代码来源:jsonsearchtest.py
示例6: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.info = UserInfo(join(self.tempdir, 'users'))
self.form = UserInfoForm(action='/action')
self.form.addObserver(self.info)
self.info.addUserInfo('normal', fullname='Full Username')
self.adminUser = BasicHtmlLoginForm.User('admin')
self.normalUser = BasicHtmlLoginForm.User('normal')
开发者ID:seecr,项目名称:meresco-html,代码行数:8,代码来源:userinfoformtest.py
示例7: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.observer = CallTrace('observer')
self.top = be((Observable(),
(LogComponent('x'),
(self.observer,),
)
))
开发者ID:seecr,项目名称:meresco-components,代码行数:8,代码来源:logcomponenttest.py
示例8: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.mock_createUpload_exception = ''
self._testpath = os.path.realpath(os.path.curdir)
self.output = StringIO()
self.mappingData = dict(identifier="mappingId", code=DEFAULT_DC_CODE)
self.mapping = Mapping('mappingId')
self.mapping.code = DEFAULT_DC_CODE
self.harvest = OnlineHarvest(self.output)
开发者ID:seecr,项目名称:meresco-harvester,代码行数:9,代码来源:onlineharvesttest.py
示例9: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self._timeNow = 1257161136.0 # 2009-11-02 11:30:00
def time():
self._timeNow += 1.0
return self._timeNow
directoryLog = DirectoryLog(self.tempdir)
self.queryLog = QueryLog(log=directoryLog, loggedPaths=['/path/sru', '/path/srw'])
self.queryLog._time = time
开发者ID:seecr,项目名称:meresco-components,代码行数:9,代码来源:querylogtest.py
示例10: setUp
def setUp(self):
SeecrTestCase.setUp(self)
def httpRequest(**kwargs):
raise StopIteration(self.response)
yield
httpclient.httpget = httpRequest
httpclient.httppost = httpRequest
httpclient.httpsget = httpRequest
httpclient.httpspost = httpRequest
开发者ID:seecr,项目名称:meresco-components,代码行数:9,代码来源:httpclienttest.py
示例11: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.matchesVersion = MatchesVersion(minVersion='1', untilVersion='3')
self.observer = CallTrace('observer', methods=dict(somemessage=lambda: (x for x in ['result'])), emptyGeneratorMethods=['updateConfig'])
self.top = be((Observable(),
(self.matchesVersion,
(self.observer,)
)
))
开发者ID:seecr,项目名称:meresco-distributed,代码行数:9,代码来源:matchesversiontest.py
示例12: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.queryLogWriter = QueryLogWriter(DirectoryLog(self.tempdir))
self.handleRequestLog = HandleRequestLog()
self._timeNow = 1257161136.0 # 2009-11-02 11:30:00
def time():
self._timeNow += 1.0
return self._timeNow
self.handleRequestLog._time = time
开发者ID:seecr,项目名称:meresco-components,代码行数:9,代码来源:srulogtest.py
示例13: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.observer = CallTrace('Observer', ignoredAttributes=['start'])
self.observable = be(
(Observable(),
(XmlParseLxml(fromKwarg='data', toKwarg='lxmlNode'),
(self.observer, )
)
)
)
开发者ID:seecr,项目名称:meresco-components,代码行数:10,代码来源:xmlpumptest.py
示例14: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.observer = CallTrace(emptyGeneratorMethods=['add'])
self.dna = be(
(Observable(),
(ReindexIdentifier(),
(self.observer, )
)
)
)
开发者ID:CatchPlus,项目名称:OAS-Open-Annotation-Service,代码行数:10,代码来源:reindexidentifiertest.py
示例15: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.observer = CallTrace(emptyGeneratorMethods=['process', "delete"])
self.dna = be(
(Observable(),
(Normalize(),
(self.observer,)
)
)
)
开发者ID:CatchPlus,项目名称:OAS-Open-Annotation-Service,代码行数:10,代码来源:normalizetest.py
示例16: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.observer = CallTrace()
self.dna = be(
(Observable(),
(RdfTypeToField(),
(self.observer, )
)
)
)
开发者ID:CatchPlus,项目名称:OAS-Open-Annotation-Service,代码行数:10,代码来源:rdftypetofieldtest.py
示例17: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.reactor = Reactor()
self.observer = CallTrace('Observer')
self.inboxDirectory = join(self.tempdir, 'inbox')
self.doneDirectory = join(self.tempdir, 'done')
makedirs(self.inboxDirectory)
makedirs(self.doneDirectory)
self.inbox = Inbox(self.reactor, inboxDirectory=self.inboxDirectory, doneDirectory=self.doneDirectory)
self.inbox.addObserver(self.observer)
开发者ID:seecr,项目名称:meresco-components,代码行数:10,代码来源:inboxtest.py
示例18: setUp
def setUp(self):
SeecrTestCase.setUp(self)
self.sruLimit = SruLimitStartRecord(limitBeyond=1000)
def searchRetrieve(**kwargs):
yield "<result/>"
self.observer = CallTrace("Observer", methods={"searchRetrieve": searchRetrieve})
self.dna = be((Observable(), (self.sruLimit, (self.observer,))))
开发者ID:seecr,项目名称:meresco-components,代码行数:10,代码来源:srulimitstartrecordtest.py
示例19: tearDown
def tearDown(self):
try:
self._reactor.calledMethods.reset() # don't keep any references.
self.lucene.close()
self.lucene = None
gc.collect()
diff = self._getJavaObjects() - self._javaObjects
self.assertEquals(0, len(diff), diff)
finally:
SeecrTestCase.tearDown(self)
开发者ID:FashtimeDotCom,项目名称:meresco-lucene,代码行数:10,代码来源:lucenetestcase.py
示例20: setUp
def setUp(self):
SeecrTestCase.setUp(self)
addDeleteToMultiSequential = AddDeleteToMultiSequential()
self.multiSequentialStorage = MultiSequentialStorage(self.tempdir)
self.top = be(
(Observable(),
(addDeleteToMultiSequential,
(self.multiSequentialStorage,)
)
)
)
开发者ID:seecr,项目名称:meresco-sequentialstore,代码行数:11,代码来源:adddeletetomultisequentialtest.py
注:本文中的seecr.test.SeecrTestCase类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论