def testServiceConfig(self):
     c = ServiceConfig(type='api', minVersion="4.2", untilVersion="5.0")
     consume(c.updateConfig(**CONFIG_SERVICES()))
     self.assertEquals(['api.front.example.org', 'alias1', 'alias2'], asList(c.servernames()))
     self.assertEquals('', asString(c.zones()))
     self.assertEquals('    location / {\n        proxy_pass http://__var_af9b2fd9f1c7f17413223dc3c26aeee4_api;\n    }', asString(c.locations()))
     self.assertEquals('    listen 0.0.0.0:80;\n', asString(c.listenLines()))

 def testDeleteUnseenRecord(self):
     try:
         consume(self.dna.all.delete(identifier="identifier"))
     except:
         # The above delete should just be silently ignored and not raise an exception
         # (as it did on some point).
         self.fail()

    def testAdd(self):
        class Factory():
            def __init__(self, observable, untokenizedFieldnames):
                self.observable = observable
                self.untokenizedFieldnames = untokenizedFieldnames

            def fieldsFor(self, fieldname, value):
                raise StopIteration([(fieldname, value)])
                yield
        fieldFactory = Factory

        fieldRegistry = FieldRegistry(drilldownFields=[DrilldownField('drilldown.field')])
        index = FieldsListToLuceneDocument(fieldRegistry, untokenizedFieldnames=[], indexFieldFactory=fieldFactory)
        observer = CallTrace(emptyGeneratorMethods=['addDocument'])
        index.addObserver(observer)
        fields = [
            ("field1", "value1"),
            ("field2", "value2"),
            ("drilldown.field", "a drilldown value"),
            ("__key__.field", "a key value"),
            ("__key__.field1", 2),
        ]
        consume(index.add(identifier="", fieldslist=fields))
        self.assertEquals(['addDocument'], observer.calledMethodNames())
        fields = observer.calledMethods[0].kwargs['fields']
        self.assertEqual([
                {'name': 'field1', 'type': 'TextField', 'value': 'value1'},
                {'name': 'field2', 'type': 'TextField', 'value': 'value2'},
                {'name': 'drilldown.field', 'type': 'FacetField', 'path': ['a drilldown value']},
                {'name': '__key__.field', 'type': 'KeyField', 'value': 'a key value'},
                {'name': '__key__.field1', 'type': 'KeyField', 'value': 2},
            ], fields)

    def testAddTypeAndMissingValueToSortField(self):
        self.response = JsonDict({
                "total": 887,
                "queryTime": 6,
                "hits": [{"id": "record:1", "score": 0.1234}]
            }).dumps()

        cq = ComposedQuery('coreA')
        q = QueryExpressionToLuceneQueryDict([], LuceneSettings()).convert(cqlToExpression("field=value"))
        cq.setCoreQuery('coreB', q)
        cq.sortKeys = [dict(sortBy='sortField', core='coreA', sortDescending=True)]
        cq.addMatch(dict(core='coreA', uniqueKey='A'), dict(core='coreB', key='B'))
        consume(self._multiLucene.executeComposedQuery(cq))
        self.assertEqual({
                "_sortKeys": [{'core': 'coreA', 'sortBy': 'sortField', 'sortDescending': True, 'type': 'String', 'missingValue': 'STRING_FIRST'}],
                "resultsFrom": "coreA",
                '_matches': {'coreA->coreB': [{'core': 'coreA', 'uniqueKey': 'A'}, {'core': 'coreB', 'key': 'B'}]},
                "_facets": {},
                "_otherCoreFacetFilters": {},
                "_rankQueries": {},
                "_drilldownQueries": {},
                "_unites": [],
                '_queries': {'coreB': {'term': {'field': 'field', 'value': 'value'}, 'type': 'TermQuery'}},
                "cores": ["coreB", "coreA"],
                "_filterQueries": {}
            }, loads(self.post[0]['data']))

    def testUpdateRecordWithDifferentFragments(self):
        uri = "uri:someuri"
        rdfDescription = """<rdf:Description rdf:about="%s" xmlns:dcterms="http://purl.org/dc/terms/" xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
    <dc:title xmlns:dc="http://purl.org/dc/elements/1.1/" xml:lang="en">title</dc:title>
</rdf:Description>""" % uri

        lxmlNode = parse(StringIO("""<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
        %s
</rdf:RDF>""" % rdfDescription))
        consume(self.dna.all.add(identifier="identifier", partname="ignored", lxmlNode=lxmlNode))

        record = self.oaiJazz.getRecord(uri)
        data = self.storage.getData(identifier=record.identifier, name='rdf')
        self.assertTrue('<dc:title xmlns:dc="http://purl.org/dc/elements/1.1/" xml:lang="en">title</dc:title>' in data, data)

        # now add with new title
        rdfDescription = """<rdf:Description rdf:about="%s" xmlns:dcterms="http://purl.org/dc/terms/" xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
    <dc:title xmlns:dc="http://purl.org/dc/elements/1.1/" xml:lang="en">new title</dc:title>
</rdf:Description>""" % uri

        lxmlNode = parse(StringIO("""<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
        %s
</rdf:RDF>""" % rdfDescription))
        consume(self.dna.all.add(identifier="identifier", partname="ignored", lxmlNode=lxmlNode))

        record = self.oaiJazz.getRecord(uri)
        data = self.storage.getData(identifier=record.identifier, name='rdf')
        self.assertFalse('<dc:title xmlns:dc="http://purl.org/dc/elements/1.1/" xml:lang="en">title</dc:title>' in data, data)
        self.assertTrue('<dc:title xmlns:dc="http://purl.org/dc/elements/1.1/" xml:lang="en">new title</dc:title>' in data, data)

 def testMaxFiles(self):
     self.assertRaises(ValueError, lambda: PersistLog(self.tempdir, maxSize=10, maxFiles=2))
     persist = PersistLog(self.tempdir, maxSize=10, maxFiles=3)
     def dataDict(nr):
         d = LOGDATADICT.copy()
         d['arguments'] += '&line=%s' % nr
         return d
     consume(persist.logData(dataDict=dataDict(1)))
     self.assertEquals(1, len(listdir(self.tempdir)))
     consume(persist.logData(dataDict=dataDict(2)))
     persist._thread.join()
     self.assertEquals(2, len(listdir(self.tempdir)))
     consume(persist.logData(dataDict=dataDict(3)))
     persist._thread.join()
     self.assertEquals(3, len(listdir(self.tempdir)))
     consume(persist.logData(dataDict=dataDict(4)))
     persist._thread.join()
     self.assertEquals(3, len(listdir(self.tempdir)))
     consume(persist.logData(dataDict=dataDict(5)))
     persist._thread.join()
     self.assertEquals(3, len(listdir(self.tempdir)))
     persist.close()
     zipped, notzipped, current = sorted(listdir(self.tempdir))
     self.assertEquals('current', current)
     self.assertTrue(zipped.endswith('.gz'))
     self.assertTrue('line=5' in open(join(self.tempdir, current)).read())
     self.assertTrue('line=4' in open(join(self.tempdir, notzipped)).read())
     self.assertTrue('line=3' in gzip.open(join(self.tempdir, zipped)).read())

    def testExecuteQueryWithtriplestoreHostPortFromObserver(self):
        self.request = TriplestoreRequest()
        self.request.addObserver(self.observer)
        self.observer.returnValues['triplestoreServer'] = ('this.server.nl', 1234)
        self.responseData = RESULT_JSON
        consume(self.request.executeQuery("select ?x where {}"))
        self.assertEquals(['triplestoreServer', 'httprequest', 'handleQueryTimes'], self.observer.calledMethodNames())

        httprequestKwargs = self.observer.calledMethods[1].kwargs
        request = httprequestKwargs.pop('request')
        self.assertEquals({
                'headers': None,
                'method': 'GET',
                'host': 'this.server.nl',
                'port': 1234,
            }, httprequestKwargs)
        parsed = urlparse(request)
        self.assertEquals('/query', parsed.path)
        self.assertEquals({'query': ['''select ?x where {}''']}, parse_qs(parsed.query))

        handleQueryTimesKwargs = self.observer.calledMethods[2].kwargs
        self.assertEquals(['index', 'queryTime'], handleQueryTimesKwargs.keys())
        self.assertEquals(Decimal('0.042'), handleQueryTimesKwargs['index'])
        qt = float(handleQueryTimesKwargs['queryTime'])
        self.assertTrue(0.0 <= qt <0.1, qt)

 def testLuceneReadonly(self):
     self.setUpLucene(readonly=True)
     self._lucene.observer_init()
     self.assertEqual([], self.post)
     self.assertRaises(RuntimeError, lambda: consume(self._lucene.setSettings()))
     self.assertRaises(RuntimeError, lambda: consume(self._lucene.addDocument(fields=[])))
     self.assertRaises(RuntimeError, lambda: consume(self._lucene.delete('identifier')))

 def testFieldsFromSummaryAnnotation(self):
     self._createAnnotationToFieldsList()
     consume(self.annotationTofieldslist.add(lxmlNode=XML(ANNOTATION_SUMMARIZING)))
     fields = self. observer.calledMethods[0].kwargs['fieldslist']
     self.assertEquals([
             ('oa:annotatedBy.uri', "http://data.bibliotheek.nl/id/bnl"),
             ('oa:motivatedBy.uri', "http://data.bibliotheek.nl/ns/nbc/oa#summarizing"),
             ('oa:hasTarget.uri', "http://data.bibliotheek.nl/ggc/ppn/78240829X"),
             ('rdf:type.uri', "http://dbpedia.org/ontology/Book"),
             ('dcterms:type.uri', "http://dbpedia.org/ontology/Book"),
             ('dcterms:title', 'De Båèrkểnhuizen, Anno 1349'),
             ('dcterms:identifier.uri', 'http://data.bibliotheek.nl/ggc/ppn/78240829X'),
             ('dcterms:creator', 'Nieuwkerk Kramer, H G'),
             ('dcterms:creator.uri', 'http://data.bibliotheek.nl/ggc/ppn/987'),
             ('dcterms:creator.rdfs:label', 'Some Author'),
             ('dcterms:date', '1966'),
             ('dcterms:language.uri', 'urn:iso:std:iso:639:-2:dut'),
             ('dcterms:language.rdfs:label', 'Nederlands'),
             ('dcterms:extent', '15 p'),
             ('dcterms:isFormatOf.uri', "urn:a:work:123"),
             ('skos:note', 'BQM_14'),
             ('dcterms:spatial.uri', 'http://data.bibliotheek.nl/uitburo/location/8e71243e-abb0-407b-83a1-303db1f676e0'),
             ('dcterms:spatial.rdfs:label', 'Museum Boerhaave'),
             ('dcterms:spatial.geo:lat', '52.1613636'),
             ('dcterms:spatial.geo:long', '4.4891784'),
             ('dcterms:spatial.vcard:region', 'Leiden')
         ], fields)

 def testUpdatableConfig(self):
     config = UpdatableConfig()
     consume(config.updateConfig(services={'dont':'care'}, config=CONFIG))
     self.assertEqual(8000, config.get('port'))
     self.assertEqual(9000, config.get('otherPort', 9000))
     self.assertEqual('localhost', config['hostname'])
     self.assertRaises(KeyError, lambda: config['doesnotexist'])

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()
    

 def testDeleteAlreadAdded(self):
     self.observer.returnValues['getRecord'].isDeleted = False
     consume(self.top.all.delete(identifier='identifier'))
     self.assertEquals(['getRecord', 'delete'], self.observer.calledMethodNames())
     getRecordCall, deleteCall = self.observer.calledMethods
     self.assertEquals(dict(identifier='identifier'), getRecordCall.kwargs)
     self.assertEquals(dict(identifier='identifier'), deleteCall.kwargs)

 def testServiceConfigThrottling(self):
     c = ServiceConfig(type='api', minVersion="4.2", untilVersion="5.0", path='/main')
     configServices = CONFIG_SERVICES()
     configServices['config']['api.frontend']['throttling'] = {
         '/path': {'max_connections_per_ip' : 10, 'max_connections': 100},
         '/other': {'max_connections_per_ip' : 30, 'max_connections': 150}
     }
     consume(c.updateConfig(**configServices))
     self.assertEquals([
         'limit_conn_zone $binary_remote_addr zone=api-other-byip:10m;',
         'limit_conn_zone $server_name zone=api-other-total:10m;',
         'limit_conn_zone $binary_remote_addr zone=api-path-byip:10m;',
         'limit_conn_zone $server_name zone=api-path-total:10m;'
         ], asString(c.zones()).split('\n'))
     self.assertEquals([
         '    location /main {',
         '        proxy_pass http://__var_af9b2fd9f1c7f17413223dc3c26aeee4_api;',
         '    }',
         '    location /other {',
         '        proxy_pass http://__var_af9b2fd9f1c7f17413223dc3c26aeee4_api;',
         '        limit_conn api-other-byip 30;',
         '        limit_conn api-other-total 150;',
         '    }',
         '    location /path {',
         '        proxy_pass http://__var_af9b2fd9f1c7f17413223dc3c26aeee4_api;',
         '        limit_conn api-path-byip 10;',
         '        limit_conn api-path-total 100;',
         '    }',
         ], asString(c.locations()).split('\n'))

 def testExcludingPaths(self):
     f = PathFilter('/path', excluding=['/path/not/this'])
     f.addObserver(self.interceptor)
     consume(f.handleRequest(path='/path/not/this/path'))
     self.assertEquals(0, len(self.interceptor.calledMethods))
     consume(f.handleRequest(path='/path/other'))
     self.assertEquals(1, len(self.interceptor.calledMethods))

 def testShouldNotChangeCallsWithStartRecordLowerThanOrEqualTo1000(self):
     sruArguments = dict(
         version="1.1", recordSchema="schema", recordPacking="xml", startRecord=1, maximumRecords=10, query="query"
     )
     consume(self.dna.all.searchRetrieve(sruArguments=sruArguments, otherKwarg="otherKwarg", **sruArguments))
     self.assertEquals(["searchRetrieve"], self.observer.calledMethodNames())
     self.assertDictEquals(
         {
             "sruArguments": {
                 "recordSchema": "schema",
                 "version": "1.1",
                 "recordPacking": "xml",
                 "maximumRecords": 10,
                 "startRecord": 1,
                 "query": "query",
             },
             "otherKwarg": "otherKwarg",
             "limitBeyond": 1000,
             "recordSchema": "schema",
             "version": "1.1",
             "recordPacking": "xml",
             "maximumRecords": 10,
             "startRecord": 1,
             "query": "query",
         },
         self.observer.calledMethods[0].kwargs,
     )

 def testLogData(self):
     persist = PersistLog(join(self.tempdir, 'store'), dictToLine=dictToLogline)
     consume(persist.logData(dataDict=LOGDATADICT))
     persist.close()
     self.assertEquals(
             '2009-11-02T11:25:36Z 11.12.13.14 4.0K 12.340s 0hits /path key=value\n',
             open(join(self.tempdir, 'store', 'current')).read())

    def testCollectLog(self):
        handler = SruHandler(enableCollectLog=True)
        observer = CallTrace('observer', emptyGeneratorMethods=['echoedExtraRequestData', 'extraResponseData'])
        __callstack_var_logCollector__ = dict()
        times = [1, 2.5, 3.5]
        def timeNow():
            return times.pop(0)
        handler._timeNow = timeNow

        def executeQuery(**kwargs):
            response = Response(total=0, hits=[])
            response.queryTime=5
            raise StopIteration(response)
            yield
        observer.methods['executeQuery'] = executeQuery
        handler.addObserver(observer)
        arguments = dict(startRecord=11, maximumRecords=15, query='query', recordPacking='string', recordSchema='schema')
        consume(handler.searchRetrieve(sruArguments=arguments, **arguments))

        self.assertEquals({
            'sru': {
                'handlingTime': [Decimal('2.500')],
                'queryTime': [Decimal('1.500')],
                'indexTime': [Decimal('0.005')],
                'numberOfRecords': [0],
                'arguments': [{
                    'startRecord': 11,
                    'query': 'query',
                    'recordPacking': 'string',
                    'maximumRecords': 15,
                    'recordSchema': 'schema',
                }],
            }
        }, __callstack_var_logCollector__)

    def testRewrite(self):
        rewrite = MessageRewrite(fromMessage='this_message', toMessage='to_message')
        observer = CallTrace(emptyGeneratorMethods=['to_message'])
        tree = be((Observable(),
            (rewrite,
                (observer,),
            )
        ))
        consume(tree.all.this_message(aap='noot'))
        self.assertEqual(['to_message'], observer.calledMethodNames())
        self.assertEqual(dict(aap='noot'), observer.calledMethods[0].kwargs)

        observer.calledMethods.reset()
        consume(tree.any.this_message(aap='noot'))
        self.assertEqual(['to_message'], observer.calledMethodNames())
        self.assertEqual(dict(aap='noot'), observer.calledMethods[0].kwargs)

        del observer.emptyGeneratorMethods[:]
        observer.calledMethods.reset()
        tree.call.this_message(aap='noot')
        self.assertEqual(['to_message'], observer.calledMethodNames())
        self.assertEqual(dict(aap='noot'), observer.calledMethods[0].kwargs)

        observer.calledMethods.reset()
        tree.do.this_message(aap='noot')
        self.assertEqual(['to_message'], observer.calledMethodNames())
        self.assertEqual(dict(aap='noot'), observer.calledMethods[0].kwargs)

    def testCollectLogWhenIndexRaisesError(self):
        handler = SruHandler(enableCollectLog=True)
        observer = CallTrace('observer', emptyGeneratorMethods=['echoedExtraRequestData', 'extraResponseData', 'additionalDiagnosticDetails'])
        __callstack_var_logCollector__ = dict()
        times = [1]
        def timeNow():
            return times.pop(0)
        handler._timeNow = timeNow

        def executeQuery(**kwargs):
            raise Exception('Sorry')
            yield
        observer.methods['executeQuery'] = executeQuery
        handler.addObserver(observer)
        arguments = dict(startRecord=11, maximumRecords=15, query='query', recordPacking='string', recordSchema='schema')
        consume(handler.searchRetrieve(sruArguments=arguments, **arguments))

        self.assertEquals({
            'sru': {
                'arguments': [{
                    'startRecord': 11,
                    'query': 'query',
                    'recordPacking': 'string',
                    'maximumRecords': 15,
                    'recordSchema': 'schema',
                }],
            }
        }, __callstack_var_logCollector__)

    def assertMaxAndSort(self, maximumRecords, sortKey, sortDirection, rssArgs, sruArgs):
        rss = Rss(
            title = 'Test title',
            description = 'Test description',
            link = 'http://www.example.org',
            **rssArgs
        )
        recordIds = []
        def getRecord(identifier):
            recordIds.append(identifier)
            return '<item/>'

        def executeQuery(start, stop, *args, **kwargs):
            response = Response(total=50, hits=[Hit(i) for i in range(start, stop)])
            raise StopIteration(response)
            yield
        observer = CallTrace(
            methods={
                'executeQuery': executeQuery,
                'getRecord': getRecord,
            },
            ignoredAttributes=['extraResponseData', 'echoedExtraRequestData'])
        rss.addObserver(observer)

        consume(rss.handleRequest(RequestURI='/?query=aQuery&' + urlencode(sruArgs)))

        method = observer.calledMethods[0]
        self.assertEquals('executeQuery', method.name)
        if sortKey is not None:
            self.assertEquals([{'sortBy': sortKey, 'sortDescending': sortDirection}], method.kwargs['sortKeys'])
        else:
            self.assertEquals(None, method.kwargs['sortKeys'])
        self.assertEquals(maximumRecords, len(recordIds))