1 # ====================================================================
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
13 # ====================================================================
15 import time, threading
16 from unittest import TestCase, main
20 class PyLuceneThreadTestCase(TestCase):
22 Test using threads in PyLucene with python threads
27 self.classLoader = Thread.currentThread().getContextClassLoader()
29 self.directory = RAMDirectory()
30 writer = IndexWriter(self.directory,
31 StandardAnalyzer(Version.LUCENE_CURRENT), True,
32 IndexWriter.MaxFieldLength.LIMITED)
38 doc1.add(Field("field", "one",
39 Field.Store.YES, Field.Index.ANALYZED))
40 doc2.add(Field("field", "two",
41 Field.Store.YES, Field.Index.ANALYZED))
42 doc3.add(Field("field", "three",
43 Field.Store.YES, Field.Index.ANALYZED))
44 doc4.add(Field("field", "one",
45 Field.Store.YES, Field.Index.ANALYZED))
47 writer.addDocument(doc1)
48 writer.addDocument(doc2)
49 writer.addDocument(doc3)
50 writer.addDocument(doc4)
54 self.testData = [('one',2), ('two',1), ('three', 1), ('five', 0)] * 500
55 self.lock = threading.Lock()
61 self.directory.close()
64 def testWithMainThread(self):
65 """ warm up test for runSearch in main thread """
67 self.runSearch(2000, True)
70 def testWithPyLuceneThread(self):
71 """ Run 5 threads with 2000 queries each """
75 threads.append(threading.Thread(target=self.runSearch,
78 for thread in threads:
81 for thread in threads:
86 # and all queries have ran successfully
87 self.assertEqual(10000, self.totalQueries)
90 def runSearch(self, runCount, mainThread=False):
91 """ search for runCount number of times """
93 # problem: if there are any assertion errors in the child
94 # thread, the calling thread is not notified and may still
95 # consider the test case pass. We are using self.totalQueries
96 # to double check that work has actually been done.
99 getVMEnv().attachCurrentThread()
102 searcher = IndexSearcher(self.directory, True)
104 self.query = PhraseQuery()
105 for word, count in self.testData[0:runCount]:
106 query = TermQuery(Term("field", word))
107 topDocs = searcher.search(query, 50)
108 self.assertEqual(topDocs.totalHits, count)
111 self.totalQueries += 1
117 if __name__ == "__main__":
120 if '-loop' in sys.argv:
121 sys.argv.remove('-loop')