# encoding: utf-8 from __future__ import absolute_import, division, print_function, unicode_literals import os import unittest from datetime import timedelta from decimal import Decimal from django.conf import settings from django.test import TestCase from django.test.utils import override_settings from django.utils.datetime_safe import date, datetime from whoosh.fields import BOOLEAN, DATETIME, KEYWORD, NUMERIC, TEXT from whoosh.qparser import QueryParser from haystack import connections, indexes, reset_search_queries from haystack.exceptions import SearchBackendError, SkipDocument from haystack.inputs import AutoQuery from haystack.models import SearchResult from haystack.query import SQ, SearchQuerySet from haystack.utils.loading import UnifiedIndex from ..core.models import AFourthMockModel, AnotherMockModel, MockModel from ..mocks import MockSearchResult from .testcases import WhooshTestCase class WhooshMockSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField(document=True, use_template=True) name = indexes.CharField(model_attr='author') pub_date = indexes.DateTimeField(model_attr='pub_date') def get_model(self): return MockModel class WhooshMockSearchIndexWithSkipDocument(WhooshMockSearchIndex): def prepare_text(self, obj): if obj.author == 'daniel3': raise SkipDocument return obj.author class WhooshAnotherMockSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField(document=True) name = indexes.CharField(model_attr='author') pub_date = indexes.DateTimeField(model_attr='pub_date') def get_model(self): return AnotherMockModel def prepare_text(self, obj): return obj.author class AllTypesWhooshMockSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField(document=True, use_template=True) name = indexes.CharField(model_attr='author', indexed=False) pub_date = indexes.DateTimeField(model_attr='pub_date') sites = indexes.MultiValueField() seen_count = indexes.IntegerField(indexed=False) is_active = indexes.BooleanField(default=True) def get_model(self): return MockModel class WhooshMaintainTypeMockSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField(document=True) month = indexes.CharField(indexed=False) pub_date = indexes.DateTimeField(model_attr='pub_date') def get_model(self): return MockModel def prepare_text(self, obj): return "Indexed!\n%s" % obj.pk def prepare_month(self, obj): return "%02d" % obj.pub_date.month class WhooshBoostMockSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField( document=True, use_template=True, template_name='search/indexes/core/mockmodel_template.txt' ) author = indexes.CharField(model_attr='author', weight=2.0) editor = indexes.CharField(model_attr='editor') pub_date = indexes.DateTimeField(model_attr='pub_date') def get_model(self): return AFourthMockModel def prepare(self, obj): data = super(WhooshBoostMockSearchIndex, self).prepare(obj) if obj.pk % 2 == 0: data['boost'] = 2.0 return data class WhooshAutocompleteMockModelSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField(model_attr='foo', document=True) name = indexes.CharField(model_attr='author') pub_date = indexes.DateTimeField(model_attr='pub_date') text_auto = indexes.EdgeNgramField(model_attr='foo') name_auto = indexes.EdgeNgramField(model_attr='author') def get_model(self): return MockModel class WhooshSearchBackendTestCase(WhooshTestCase): fixtures = ['bulk_data.json'] def setUp(self): super(WhooshSearchBackendTestCase, self).setUp() self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wmmi = WhooshMockSearchIndex() self.wmmidni = WhooshMockSearchIndexWithSkipDocument() self.wmtmmi = WhooshMaintainTypeMockSearchIndex() self.ui.build(indexes=[self.wmmi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.sample_objs = MockModel.objects.all() def tearDown(self): connections['whoosh']._index = self.old_ui super(WhooshSearchBackendTestCase, self).tearDown() def whoosh_search(self, query): self.raw_whoosh = self.raw_whoosh.refresh() searcher = self.raw_whoosh.searcher() return searcher.search(self.parser.parse(query), limit=1000) def test_non_silent(self): bad_sb = connections['whoosh'].backend('bad', PATH='/tmp/bad_whoosh', SILENTLY_FAIL=False) bad_sb.use_file_storage = False bad_sb.storage = 'omg.wtf.bbq' try: bad_sb.update(self.wmmi, self.sample_objs) self.fail() except: pass try: bad_sb.remove('core.mockmodel.1') self.fail() except: pass try: bad_sb.clear() self.fail() except: pass try: bad_sb.search('foo') self.fail() except: pass def test_update(self): self.sb.update(self.wmmi, self.sample_objs) # Check what Whoosh thinks is there. self.assertEqual(len(self.whoosh_search(u'*')), 23) self.assertEqual([doc.fields()['id'] for doc in self.whoosh_search(u'*')], [u'core.mockmodel.%s' % i for i in range(1, 24)]) def test_update_with_SkipDocument_raised(self): self.sb.update(self.wmmidni, self.sample_objs) # Check what Whoosh thinks is there. res = self.whoosh_search(u'*') self.assertEqual(len(res), 14) ids = [1, 2, 5, 6, 7, 8, 9, 11, 12, 14, 15, 18, 20, 21] self.assertListEqual( [doc.fields()['id'] for doc in res], [u'core.mockmodel.%s' % i for i in ids] ) def test_remove(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(self.sb.index.doc_count(), 23) self.sb.remove(self.sample_objs[0]) self.assertEqual(self.sb.index.doc_count(), 22) def test_clear(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(self.sb.index.doc_count(), 23) self.sb.clear() self.assertEqual(self.sb.index.doc_count(), 0) self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(self.sb.index.doc_count(), 23) self.sb.clear([AnotherMockModel]) self.assertEqual(self.sb.index.doc_count(), 23) self.sb.clear([MockModel]) self.assertEqual(self.sb.index.doc_count(), 0) self.sb.index.refresh() self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(self.sb.index.doc_count(), 23) self.sb.clear([AnotherMockModel, MockModel]) self.assertEqual(self.raw_whoosh.doc_count(), 0) def test_search(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(len(self.whoosh_search(u'*')), 23) # No query string should always yield zero results. self.assertEqual(self.sb.search(u''), {'hits': 0, 'results': []}) # A one letter query string gets nabbed by a stopwords filter. Should # always yield zero results. self.assertEqual(self.sb.search(u'a'), {'hits': 0, 'results': []}) # Possible AttributeError? # self.assertEqual(self.sb.search(u'a b'), {'hits': 0, 'results': [], 'spelling_suggestion': '', 'facets': {}}) self.assertEqual(self.sb.search(u'*')['hits'], 23) self.assertEqual([result.pk for result in self.sb.search(u'*')['results']], [u'%s' % i for i in range(1, 24)]) self.assertEqual(self.sb.search(u'Indexe')['hits'], 23) self.assertEqual(self.sb.search(u'Indexe')['spelling_suggestion'], u'indexed') self.assertEqual(self.sb.search(u'', facets=['name']), {'hits': 0, 'results': []}) results = self.sb.search(u'Index*', facets=['name']) results = self.sb.search(u'index*', facets=['name']) self.assertEqual(results['hits'], 23) self.assertEqual(results['facets'], {}) self.assertEqual(self.sb.search(u'', date_facets={'pub_date': {'start_date': date(2008, 2, 26), 'end_date': date(2008, 2, 26), 'gap': '/MONTH'}}), {'hits': 0, 'results': []}) results = self.sb.search(u'Index*', date_facets={'pub_date': {'start_date': date(2008, 2, 26), 'end_date': date(2008, 2, 26), 'gap': '/MONTH'}}) results = self.sb.search(u'index*', date_facets={'pub_date': {'start_date': date(2008, 2, 26), 'end_date': date(2008, 2, 26), 'gap': '/MONTH'}}) self.assertEqual(results['hits'], 23) self.assertEqual(results['facets'], {}) self.assertEqual(self.sb.search(u'', query_facets={'name': '[* TO e]'}), {'hits': 0, 'results': []}) results = self.sb.search(u'Index*', query_facets={'name': '[* TO e]'}) results = self.sb.search(u'index*', query_facets={'name': '[* TO e]'}) self.assertEqual(results['hits'], 23) self.assertEqual(results['facets'], {}) # self.assertEqual(self.sb.search('', narrow_queries=set(['name:daniel1'])), {'hits': 0, 'results': []}) # results = self.sb.search('Index*', narrow_queries=set(['name:daniel1'])) # self.assertEqual(results['hits'], 1) # Ensure that swapping the ``result_class`` works. self.assertTrue(isinstance(self.sb.search(u'Index*', result_class=MockSearchResult)['results'][0], MockSearchResult)) # Check the use of ``limit_to_registered_models``. self.assertEqual(self.sb.search(u'', limit_to_registered_models=False), {'hits': 0, 'results': []}) self.assertEqual(self.sb.search(u'*', limit_to_registered_models=False)['hits'], 23) self.assertEqual([result.pk for result in self.sb.search(u'*', limit_to_registered_models=False)['results']], [u'%s' % i for i in range(1, 24)]) # Stow. old_limit_to_registered_models = getattr(settings, 'HAYSTACK_LIMIT_TO_REGISTERED_MODELS', True) settings.HAYSTACK_LIMIT_TO_REGISTERED_MODELS = False self.assertEqual(self.sb.search(u''), {'hits': 0, 'results': []}) self.assertEqual(self.sb.search(u'*')['hits'], 23) self.assertEqual([result.pk for result in self.sb.search(u'*')['results']], [u'%s' % i for i in range(1, 24)]) # Restore. settings.HAYSTACK_LIMIT_TO_REGISTERED_MODELS = old_limit_to_registered_models def test_highlight(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(len(self.whoosh_search(u'*')), 23) self.assertEqual(self.sb.search(u'', highlight=True), {'hits': 0, 'results': []}) self.assertEqual(self.sb.search(u'index*', highlight=True)['hits'], 23) query = self.sb.search('Index*', highlight=True)['results'] result = [result.highlighted['text'][0] for result in query] self.assertEqual(result, ['Indexed!\n%d' % i for i in range(1, 24)]) def test_search_all_models(self): wamsi = WhooshAnotherMockSearchIndex() self.ui.build(indexes=[self.wmmi, wamsi]) self.sb.update(self.wmmi, self.sample_objs) self.sb.update(wamsi, AnotherMockModel.objects.all()) self.assertEqual(len(self.whoosh_search(u'*')), 25) self.ui.build(indexes=[self.wmmi]) def test_more_like_this(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(len(self.whoosh_search(u'*')), 23) # Now supported by Whoosh (as of 1.8.4). See the ``LiveWhooshMoreLikeThisTestCase``. self.assertEqual(self.sb.more_like_this(self.sample_objs[0])['hits'], 22) # Make sure that swapping the ``result_class`` doesn't blow up. try: self.sb.more_like_this(self.sample_objs[0], result_class=MockSearchResult) except: self.fail() def test_delete_index(self): self.sb.update(self.wmmi, self.sample_objs) self.assertTrue(self.sb.index.doc_count() > 0) self.sb.delete_index() self.assertEqual(self.sb.index.doc_count(), 0) def test_order_by(self): self.sb.update(self.wmmi, self.sample_objs) results = self.sb.search(u'*', sort_by=['pub_date']) self.assertEqual([result.pk for result in results['results']], [u'1', u'3', u'2', u'4', u'5', u'6', u'7', u'8', u'9', u'10', u'11', u'12', u'13', u'14', u'15', u'16', u'17', u'18', u'19', u'20', u'21', u'22', u'23']) results = self.sb.search(u'*', sort_by=['-pub_date']) self.assertEqual([result.pk for result in results['results']], [u'23', u'22', u'21', u'20', u'19', u'18', u'17', u'16', u'15', u'14', u'13', u'12', u'11', u'10', u'9', u'8', u'7', u'6', u'5', u'4', u'2', u'3', u'1']) results = self.sb.search(u'*', sort_by=['id']) self.assertEqual([result.pk for result in results['results']], [u'1', u'10', u'11', u'12', u'13', u'14', u'15', u'16', u'17', u'18', u'19', u'2', u'20', u'21', u'22', u'23', u'3', u'4', u'5', u'6', u'7', u'8', u'9']) results = self.sb.search(u'*', sort_by=['-id']) self.assertEqual([result.pk for result in results['results']], [u'9', u'8', u'7', u'6', u'5', u'4', u'3', u'23', u'22', u'21', u'20', u'2', u'19', u'18', u'17', u'16', u'15', u'14', u'13', u'12', u'11', u'10', u'1']) results = self.sb.search(u'*', sort_by=['-pub_date', '-id']) self.assertEqual([result.pk for result in results['results']], [u'23', u'22', u'21', u'20', u'19', u'18', u'17', u'16', u'15', u'14', u'13', u'12', u'11', u'10', u'9', u'8', u'7', u'6', u'5', u'4', u'2', u'3', u'1']) self.assertRaises(SearchBackendError, self.sb.search, u'*', sort_by=['-pub_date', 'id']) def test__from_python(self): self.assertEqual(self.sb._from_python('abc'), u'abc') self.assertEqual(self.sb._from_python(1), 1) self.assertEqual(self.sb._from_python(2653), 2653) self.assertEqual(self.sb._from_python(25.5), 25.5) self.assertEqual(self.sb._from_python([1, 2, 3]), u'1,2,3') self.assertTrue("a': 1" in self.sb._from_python({'a': 1, 'c': 3, 'b': 2})) self.assertEqual(self.sb._from_python(datetime(2009, 5, 9, 16, 14)), datetime(2009, 5, 9, 16, 14)) self.assertEqual(self.sb._from_python(datetime(2009, 5, 9, 0, 0)), datetime(2009, 5, 9, 0, 0)) self.assertEqual(self.sb._from_python(datetime(1899, 5, 18, 0, 0)), datetime(1899, 5, 18, 0, 0)) self.assertEqual(self.sb._from_python(datetime(2009, 5, 18, 1, 16, 30, 250)), datetime(2009, 5, 18, 1, 16, 30, 250)) def test__to_python(self): self.assertEqual(self.sb._to_python('abc'), 'abc') self.assertEqual(self.sb._to_python('1'), 1) self.assertEqual(self.sb._to_python('2653'), 2653) self.assertEqual(self.sb._to_python('25.5'), 25.5) self.assertEqual(self.sb._to_python('[1, 2, 3]'), [1, 2, 3]) self.assertEqual(self.sb._to_python('{"a": 1, "b": 2, "c": 3}'), {'a': 1, 'c': 3, 'b': 2}) self.assertEqual(self.sb._to_python('2009-05-09T16:14:00'), datetime(2009, 5, 9, 16, 14)) self.assertEqual(self.sb._to_python('2009-05-09T00:00:00'), datetime(2009, 5, 9, 0, 0)) self.assertEqual(self.sb._to_python(None), None) def test_range_queries(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(len(self.whoosh_search(u'[d TO]')), 23) self.assertEqual(len(self.whoosh_search(u'name:[d TO]')), 23) self.assertEqual(len(self.whoosh_search(u'Ind* AND name:[d to]')), 23) self.assertEqual(len(self.whoosh_search(u'Ind* AND name:[to c]')), 0) def test_date_queries(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(len(self.whoosh_search(u"pub_date:20090717003000")), 1) self.assertEqual(len(self.whoosh_search(u"pub_date:20090717000000")), 0) self.assertEqual(len(self.whoosh_search(u'Ind* AND pub_date:[to 20090717003000]')), 3) def test_escaped_characters_queries(self): self.sb.update(self.wmmi, self.sample_objs) self.assertEqual(len(self.whoosh_search(u"Indexed\!")), 23) self.assertEqual(len(self.whoosh_search(u"http\:\/\/www\.example\.com")), 0) def test_build_schema(self): ui = UnifiedIndex() ui.build(indexes=[AllTypesWhooshMockSearchIndex()]) (content_field_name, schema) = self.sb.build_schema(ui.all_searchfields()) self.assertEqual(content_field_name, 'text') schema_names = set(schema.names()) required_schema = {'django_ct', 'django_id', 'id', 'is_active', 'name', 'pub_date', 'seen_count', 'sites', 'text'} self.assertTrue(required_schema.issubset(schema_names)) self.assertIsInstance(schema._fields['text'], TEXT) self.assertIsInstance(schema._fields['pub_date'], DATETIME) self.assertIsInstance(schema._fields['seen_count'], NUMERIC) self.assertIsInstance(schema._fields['sites'], KEYWORD) self.assertIsInstance(schema._fields['is_active'], BOOLEAN) def test_verify_type(self): old_ui = connections['whoosh'].get_unified_index() ui = UnifiedIndex() wmtmmi = WhooshMaintainTypeMockSearchIndex() ui.build(indexes=[wmtmmi]) connections['whoosh']._index = ui sb = connections['whoosh'].get_backend() sb.setup() sb.update(wmtmmi, self.sample_objs) self.assertEqual(sb.search(u'*')['hits'], 23) self.assertEqual([result.month for result in sb.search(u'*')['results']], [u'06', u'07', u'06', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07', u'07']) connections['whoosh']._index = old_ui @unittest.skipIf(settings.HAYSTACK_CONNECTIONS['whoosh'].get('STORAGE') != 'file', 'testing writability requires Whoosh to use STORAGE=file') def test_writable(self): if not os.path.exists(settings.HAYSTACK_CONNECTIONS['whoosh']['PATH']): os.makedirs(settings.HAYSTACK_CONNECTIONS['whoosh']['PATH']) os.chmod(settings.HAYSTACK_CONNECTIONS['whoosh']['PATH'], 0o400) try: self.sb.setup() self.fail() except IOError: # Yay. We failed pass os.chmod(settings.HAYSTACK_CONNECTIONS['whoosh']['PATH'], 0o755) def test_slicing(self): self.sb.update(self.wmmi, self.sample_objs) page_1 = self.sb.search(u'*', start_offset=0, end_offset=20) page_2 = self.sb.search(u'*', start_offset=20, end_offset=30) self.assertEqual(len(page_1['results']), 20) self.assertEqual([result.pk for result in page_1['results']], [u'%s' % i for i in range(1, 21)]) self.assertEqual(len(page_2['results']), 3) self.assertEqual([result.pk for result in page_2['results']], [u'21', u'22', u'23']) # This used to throw an error. page_0 = self.sb.search(u'*', start_offset=0, end_offset=0) self.assertEqual(len(page_0['results']), 1) @unittest.expectedFailure def test_scoring(self): self.sb.update(self.wmmi, self.sample_objs) page_1 = self.sb.search(u'index', start_offset=0, end_offset=20) page_2 = self.sb.search(u'index', start_offset=20, end_offset=30) self.assertEqual(len(page_1['results']), 20) self.assertEqual(["%0.2f" % result.score for result in page_1['results']], ['0.51', '0.51', '0.51', '0.51', '0.51', '0.51', '0.51', '0.51', '0.51', '0.40', '0.40', '0.40', '0.40', '0.40', '0.40', '0.40', '0.40', '0.40', '0.40', '0.40']) self.assertEqual(len(page_2['results']), 3) self.assertEqual(["%0.2f" % result.score for result in page_2['results']], ['0.40', '0.40', '0.40']) class WhooshBoostBackendTestCase(WhooshTestCase): def setUp(self): super(WhooshBoostBackendTestCase, self).setUp() self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wmmi = WhooshBoostMockSearchIndex() self.ui.build(indexes=[self.wmmi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.sample_objs = [] for i in range(1, 5): mock = AFourthMockModel() mock.id = i if i % 2: mock.author = 'daniel' mock.editor = 'david' else: mock.author = 'david' mock.editor = 'daniel' mock.pub_date = date(2009, 2, 25) - timedelta(days=i) self.sample_objs.append(mock) def tearDown(self): connections['whoosh']._index = self.ui super(WhooshBoostBackendTestCase, self).tearDown() @unittest.expectedFailure def test_boost(self): self.sb.update(self.wmmi, self.sample_objs) self.raw_whoosh = self.raw_whoosh.refresh() searcher = self.raw_whoosh.searcher() self.assertEqual(len(searcher.search(self.parser.parse(u'*'), limit=1000)), 2) results = SearchQuerySet('whoosh').filter(SQ(author='daniel') | SQ(editor='daniel')) self.assertEqual([result.id for result in results], [ 'core.afourthmockmodel.1', 'core.afourthmockmodel.3', ]) self.assertEqual(results[0].boost, 1.1) class LiveWhooshSearchQueryTestCase(WhooshTestCase): def setUp(self): super(LiveWhooshSearchQueryTestCase, self).setUp() # Stow. self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wmmi = WhooshMockSearchIndex() self.wmtmmi = WhooshMaintainTypeMockSearchIndex() self.ui.build(indexes=[self.wmmi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.sample_objs = [] for i in range(1, 4): mock = MockModel() mock.id = i mock.author = 'daniel%s' % i mock.pub_date = date(2009, 2, 25) - timedelta(days=i) self.sample_objs.append(mock) self.sq = connections['whoosh'].get_query() def tearDown(self): connections['whoosh']._index = self.old_ui super(LiveWhooshSearchQueryTestCase, self).tearDown() def test_get_spelling(self): self.sb.update(self.wmmi, self.sample_objs) self.sq.add_filter(SQ(content='Indexe')) self.assertEqual(self.sq.get_spelling_suggestion(), u'indexed') def test_log_query(self): from django.conf import settings reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) # Stow. with self.settings(DEBUG=False): len(self.sq.get_results()) self.assertEqual(len(connections['whoosh'].queries), 0) with self.settings(DEBUG=True): # Redefine it to clear out the cached results. self.sq = connections['whoosh'].get_query() self.sq.add_filter(SQ(name='bar')) len(self.sq.get_results()) self.assertEqual(len(connections['whoosh'].queries), 1) self.assertEqual(connections['whoosh'].queries[0]['query_string'], 'name:(bar)') # And again, for good measure. self.sq = connections['whoosh'].get_query() self.sq.add_filter(SQ(name='baz')) self.sq.add_filter(SQ(text='foo')) len(self.sq.get_results()) self.assertEqual(len(connections['whoosh'].queries), 2) self.assertEqual(connections['whoosh'].queries[0]['query_string'], 'name:(bar)') self.assertEqual(connections['whoosh'].queries[1]['query_string'], u'(name:(baz) AND text:(foo))') @override_settings(DEBUG=True) class LiveWhooshSearchQuerySetTestCase(WhooshTestCase): def setUp(self): super(LiveWhooshSearchQuerySetTestCase, self).setUp() # Stow. self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wmmi = WhooshMockSearchIndex() self.ui.build(indexes=[self.wmmi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.sample_objs = [] for i in range(1, 4): mock = MockModel() mock.id = i mock.author = 'daniel%s' % i mock.pub_date = date(2009, 2, 25) - timedelta(days=i) self.sample_objs.append(mock) self.sq = connections['whoosh'].get_query() self.sqs = SearchQuerySet('whoosh') def tearDown(self): connections['whoosh']._index = self.old_ui super(LiveWhooshSearchQuerySetTestCase, self).tearDown() def test_various_searchquerysets(self): self.sb.update(self.wmmi, self.sample_objs) sqs = self.sqs.filter(content='Index') self.assertEqual(sqs.query.build_query(), u'(Index)') self.assertEqual(len(sqs), 3) sqs = self.sqs.auto_query('Indexed!') self.assertEqual(sqs.query.build_query(), u"('Indexed!')") self.assertEqual(len(sqs), 3) sqs = self.sqs.auto_query('Indexed!').filter(pub_date__lte=date(2009, 8, 31)) self.assertEqual(sqs.query.build_query(), u"(('Indexed!') AND pub_date:([to 20090831000000]))") self.assertEqual(len(sqs), 3) sqs = self.sqs.auto_query('Indexed!').filter(pub_date__lte=date(2009, 2, 23)) self.assertEqual(sqs.query.build_query(), u"(('Indexed!') AND pub_date:([to 20090223000000]))") self.assertEqual(len(sqs), 2) sqs = self.sqs.auto_query('Indexed!').filter(pub_date__lte=date(2009, 2, 25)).filter(django_id__in=[1, 2]).exclude(name='daniel1') self.assertEqual(sqs.query.build_query(), u'((\'Indexed!\') AND pub_date:([to 20090225000000]) AND django_id:(1 OR 2) AND NOT (name:(daniel1)))') self.assertEqual(len(sqs), 1) sqs = self.sqs.auto_query('re-inker') self.assertEqual(sqs.query.build_query(), u"('re-inker')") self.assertEqual(len(sqs), 0) sqs = self.sqs.auto_query('0.7 wire') self.assertEqual(sqs.query.build_query(), u"('0.7' wire)") self.assertEqual(len(sqs), 0) sqs = self.sqs.auto_query("daler-rowney pearlescent 'bell bronze'") self.assertEqual(sqs.query.build_query(), u"('daler-rowney' pearlescent 'bell bronze')") self.assertEqual(len(sqs), 0) sqs = self.sqs.models(MockModel) self.assertEqual(sqs.query.build_query(), u'*') self.assertEqual(len(sqs), 3) def test_all_regression(self): sqs = SearchQuerySet('whoosh') self.assertEqual([result.pk for result in sqs], []) self.sb.update(self.wmmi, self.sample_objs) self.assertTrue(self.sb.index.doc_count() > 0) sqs = SearchQuerySet('whoosh') self.assertEqual(len(sqs), 3) self.assertEqual(sorted([result.pk for result in sqs]), [u'1', u'2', u'3']) try: sqs = repr(SearchQuerySet('whoosh')) except: self.fail() def test_regression_space_query(self): self.sb.update(self.wmmi, self.sample_objs) self.assertTrue(self.sb.index.doc_count() > 0) sqs = SearchQuerySet('whoosh').auto_query(" ") self.assertEqual(len(sqs), 3) sqs = SearchQuerySet('whoosh').filter(content=" ") self.assertEqual(len(sqs), 0) def test_iter(self): self.sb.update(self.wmmi, self.sample_objs) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) sqs = self.sqs.auto_query('Indexed!') results = [int(result.pk) for result in iter(sqs)] self.assertEqual(sorted(results), [1, 2, 3]) self.assertEqual(len(connections['whoosh'].queries), 1) def test_slice(self): self.sb.update(self.wmmi, self.sample_objs) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) results = self.sqs.auto_query('Indexed!') self.assertEqual(sorted([int(result.pk) for result in results[1:3]]), [1, 2]) self.assertEqual(len(connections['whoosh'].queries), 1) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) results = self.sqs.auto_query('Indexed!') self.assertEqual(int(results[0].pk), 1) self.assertEqual(len(connections['whoosh'].queries), 1) def test_values_slicing(self): self.sb.update(self.wmmi, self.sample_objs) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) # TODO: this would be a good candidate for refactoring into a TestCase subclass shared across backends # The values will come back as strings because Hasytack doesn't assume PKs are integers. # We'll prepare this set once since we're going to query the same results in multiple ways: expected_pks = ['3', '2', '1'] results = self.sqs.all().order_by('pub_date').values('pk') self.assertListEqual([i['pk'] for i in results[1:11]], expected_pks) results = self.sqs.all().order_by('pub_date').values_list('pk') self.assertListEqual([i[0] for i in results[1:11]], expected_pks) results = self.sqs.all().order_by('pub_date').values_list('pk', flat=True) self.assertListEqual(results[1:11], expected_pks) self.assertEqual(len(connections['whoosh'].queries), 3) def test_manual_iter(self): self.sb.update(self.wmmi, self.sample_objs) results = self.sqs.auto_query('Indexed!') reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) results = [int(result.pk) for result in results._manual_iter()] self.assertEqual(sorted(results), [1, 2, 3]) self.assertEqual(len(connections['whoosh'].queries), 1) def test_fill_cache(self): self.sb.update(self.wmmi, self.sample_objs) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) results = self.sqs.auto_query('Indexed!') self.assertEqual(len(results._result_cache), 0) self.assertEqual(len(connections['whoosh'].queries), 0) results._fill_cache(0, 10) self.assertEqual(len([result for result in results._result_cache if result is not None]), 3) self.assertEqual(len(connections['whoosh'].queries), 1) results._fill_cache(10, 20) self.assertEqual(len([result for result in results._result_cache if result is not None]), 3) self.assertEqual(len(connections['whoosh'].queries), 2) def test_cache_is_full(self): self.sb.update(self.wmmi, self.sample_objs) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) self.assertEqual(self.sqs._cache_is_full(), False) results = self.sqs.auto_query('Indexed!') result_list = [i for i in iter(results)] self.assertEqual(results._cache_is_full(), True) self.assertEqual(len(connections['whoosh'].queries), 1) def test_count(self): more_samples = [] for i in range(1, 50): mock = MockModel() mock.id = i mock.author = 'daniel%s' % i mock.pub_date = date(2009, 2, 25) - timedelta(days=i) more_samples.append(mock) self.sb.update(self.wmmi, more_samples) reset_search_queries() self.assertEqual(len(connections['whoosh'].queries), 0) results = self.sqs.all() self.assertEqual(len(results), 49) self.assertEqual(results._cache_is_full(), False) self.assertEqual(len(connections['whoosh'].queries), 1) def test_query_generation(self): sqs = self.sqs.filter(SQ(content=AutoQuery("hello world")) | SQ(title=AutoQuery("hello world"))) self.assertEqual(sqs.query.build_query(), u"((hello world) OR title:(hello world))") def test_result_class(self): self.sb.update(self.wmmi, self.sample_objs) # Assert that we're defaulting to ``SearchResult``. sqs = self.sqs.all() self.assertTrue(isinstance(sqs[0], SearchResult)) # Custom class. sqs = self.sqs.result_class(MockSearchResult).all() self.assertTrue(isinstance(sqs[0], MockSearchResult)) # Reset to default. sqs = self.sqs.result_class(None).all() self.assertTrue(isinstance(sqs[0], SearchResult)) class LiveWhooshMultiSearchQuerySetTestCase(WhooshTestCase): fixtures = ['bulk_data.json'] def setUp(self): super(LiveWhooshMultiSearchQuerySetTestCase, self).setUp() # Stow. self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wmmi = WhooshMockSearchIndex() self.wamsi = WhooshAnotherMockSearchIndex() self.ui.build(indexes=[self.wmmi, self.wamsi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.wmmi.update(using='whoosh') self.wamsi.update(using='whoosh') self.sqs = SearchQuerySet('whoosh') def tearDown(self): connections['whoosh']._index = self.old_ui super(LiveWhooshMultiSearchQuerySetTestCase, self).tearDown() def test_searchquerysets_with_models(self): sqs = self.sqs.all() self.assertEqual(sqs.query.build_query(), u'*') self.assertEqual(len(sqs), 25) sqs = self.sqs.models(MockModel) self.assertEqual(sqs.query.build_query(), u'*') self.assertEqual(len(sqs), 23) sqs = self.sqs.models(AnotherMockModel) self.assertEqual(sqs.query.build_query(), u'*') self.assertEqual(len(sqs), 2) class LiveWhooshMoreLikeThisTestCase(WhooshTestCase): fixtures = ['bulk_data.json'] def setUp(self): super(LiveWhooshMoreLikeThisTestCase, self).setUp() # Stow. self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wmmi = WhooshMockSearchIndex() self.wamsi = WhooshAnotherMockSearchIndex() self.ui.build(indexes=[self.wmmi, self.wamsi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.wmmi.update() self.wamsi.update() self.sqs = SearchQuerySet('whoosh') def tearDown(self): connections['whoosh']._index = self.old_ui super(LiveWhooshMoreLikeThisTestCase, self).tearDown() # We expect failure here because, despite not changing the code, Whoosh # 2.5.1 returns incorrect counts/results. Huzzah. @unittest.expectedFailure def test_more_like_this(self): mlt = self.sqs.more_like_this(MockModel.objects.get(pk=22)) self.assertEqual(mlt.count(), 22) self.assertEqual(sorted([result.pk for result in mlt]), sorted([u'9', u'8', u'7', u'6', u'5', u'4', u'3', u'2', u'1', u'21', u'20', u'19', u'18', u'17', u'16', u'15', u'14', u'13', u'12', u'11', u'10', u'23'])) self.assertEqual(len([result.pk for result in mlt]), 22) alt_mlt = self.sqs.filter(name='daniel3').more_like_this(MockModel.objects.get(pk=13)) self.assertEqual(alt_mlt.count(), 8) self.assertEqual(sorted([result.pk for result in alt_mlt]), sorted([u'4', u'3', u'22', u'19', u'17', u'16', u'10', u'23'])) self.assertEqual(len([result.pk for result in alt_mlt]), 8) alt_mlt_with_models = self.sqs.models(MockModel).more_like_this(MockModel.objects.get(pk=11)) self.assertEqual(alt_mlt_with_models.count(), 22) self.assertEqual(sorted([result.pk for result in alt_mlt_with_models]), sorted([u'9', u'8', u'7', u'6', u'5', u'4', u'3', u'2', u'1', u'22', u'21', u'20', u'19', u'18', u'17', u'16', u'15', u'14', u'13', u'12', u'10', u'23'])) self.assertEqual(len([result.pk for result in alt_mlt_with_models]), 22) if hasattr(MockModel.objects, 'defer'): # Make sure MLT works with deferred bits. mi = MockModel.objects.defer('foo').get(pk=22) deferred = self.sqs.models(MockModel).more_like_this(mi) self.assertEqual(deferred.count(), 22) self.assertEqual(sorted([result.pk for result in deferred]), sorted([u'9', u'8', u'7', u'6', u'5', u'4', u'3', u'2', u'1', u'21', u'20', u'19', u'18', u'17', u'16', u'15', u'14', u'13', u'12', u'11', u'10', u'23'])) self.assertEqual(len([result.pk for result in deferred]), 22) # Ensure that swapping the ``result_class`` works. self.assertTrue(isinstance(self.sqs.result_class(MockSearchResult).more_like_this(MockModel.objects.get(pk=21))[0], MockSearchResult)) @override_settings(DEBUG=True) class LiveWhooshAutocompleteTestCase(WhooshTestCase): fixtures = ['bulk_data.json'] def setUp(self): super(LiveWhooshAutocompleteTestCase, self).setUp() # Stow. self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wacsi = WhooshAutocompleteMockModelSearchIndex() self.ui.build(indexes=[self.wacsi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui # Stow. import haystack self.sb.setup() self.sqs = SearchQuerySet('whoosh') # Wipe it clean. self.sqs.query.backend.clear() self.wacsi.update(using='whoosh') def tearDown(self): connections['whoosh']._index = self.old_ui super(LiveWhooshAutocompleteTestCase, self).tearDown() def test_autocomplete(self): autocomplete = self.sqs.autocomplete(text_auto='mod') self.assertEqual(autocomplete.count(), 5) self.assertEqual([result.pk for result in autocomplete], [u'1', u'12', u'6', u'7', u'14']) self.assertTrue('mod' in autocomplete[0].text.lower()) self.assertTrue('mod' in autocomplete[1].text.lower()) self.assertTrue('mod' in autocomplete[2].text.lower()) self.assertTrue('mod' in autocomplete[3].text.lower()) self.assertTrue('mod' in autocomplete[4].text.lower()) self.assertEqual(len([result.pk for result in autocomplete]), 5) def test_edgengram_regression(self): autocomplete = self.sqs.autocomplete(text_auto='ngm') self.assertEqual(autocomplete.count(), 0) def test_extra_whitespace(self): autocomplete = self.sqs.autocomplete(text_auto='mod ') self.assertEqual(autocomplete.count(), 5) class WhooshRoundTripSearchIndex(indexes.SearchIndex, indexes.Indexable): text = indexes.CharField(document=True, default='') name = indexes.CharField() is_active = indexes.BooleanField() post_count = indexes.IntegerField() average_rating = indexes.FloatField() price = indexes.DecimalField() pub_date = indexes.DateField() created = indexes.DateTimeField() tags = indexes.MultiValueField() sites = indexes.MultiValueField() # For a regression involving lists with nothing in them. empty_list = indexes.MultiValueField() def get_model(self): return MockModel def prepare(self, obj): prepped = super(WhooshRoundTripSearchIndex, self).prepare(obj) prepped.update({ 'text': 'This is some example text.', 'name': 'Mister Pants', 'is_active': True, 'post_count': 25, 'average_rating': 3.6, 'price': Decimal('24.99'), 'pub_date': date(2009, 11, 21), 'created': datetime(2009, 11, 21, 21, 31, 00), 'tags': ['staff', 'outdoor', 'activist', 'scientist'], 'sites': [3, 5, 1], 'empty_list': [], }) return prepped @override_settings(DEBUG=True) class LiveWhooshRoundTripTestCase(WhooshTestCase): def setUp(self): super(LiveWhooshRoundTripTestCase, self).setUp() # Stow. self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wrtsi = WhooshRoundTripSearchIndex() self.ui.build(indexes=[self.wrtsi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sb.delete_index() self.sqs = SearchQuerySet('whoosh') # Wipe it clean. self.sqs.query.backend.clear() # Fake indexing. mock = MockModel() mock.id = 1 self.sb.update(self.wrtsi, [mock]) def tearDown(self): super(LiveWhooshRoundTripTestCase, self).tearDown() def test_round_trip(self): results = self.sqs.filter(id='core.mockmodel.1') # Sanity check. self.assertEqual(results.count(), 1) # Check the individual fields. result = results[0] self.assertEqual(result.id, 'core.mockmodel.1') self.assertEqual(result.text, 'This is some example text.') self.assertEqual(result.name, 'Mister Pants') self.assertEqual(result.is_active, True) self.assertEqual(result.post_count, 25) self.assertEqual(result.average_rating, 3.6) self.assertEqual(result.price, u'24.99') self.assertEqual(result.pub_date, datetime(2009, 11, 21, 0, 0)) self.assertEqual(result.created, datetime(2009, 11, 21, 21, 31, 00)) self.assertEqual(result.tags, ['staff', 'outdoor', 'activist', 'scientist']) self.assertEqual(result.sites, [u'3', u'5', u'1']) self.assertEqual(result.empty_list, []) # Check boolean filtering... results = self.sqs.filter(id='core.mockmodel.1', is_active=True) self.assertEqual(results.count(), 1) @override_settings(DEBUG=True) class LiveWhooshRamStorageTestCase(TestCase): def setUp(self): super(LiveWhooshRamStorageTestCase, self).setUp() # Stow. self.old_whoosh_storage = settings.HAYSTACK_CONNECTIONS['whoosh'].get('STORAGE', 'file') settings.HAYSTACK_CONNECTIONS['whoosh']['STORAGE'] = 'ram' self.old_ui = connections['whoosh'].get_unified_index() self.ui = UnifiedIndex() self.wrtsi = WhooshRoundTripSearchIndex() self.ui.build(indexes=[self.wrtsi]) self.sb = connections['whoosh'].get_backend() connections['whoosh']._index = self.ui # Stow. import haystack self.sb.setup() self.raw_whoosh = self.sb.index self.parser = QueryParser(self.sb.content_field_name, schema=self.sb.schema) self.sqs = SearchQuerySet('whoosh') # Wipe it clean. self.sqs.query.backend.clear() # Fake indexing. mock = MockModel() mock.id = 1 self.sb.update(self.wrtsi, [mock]) def tearDown(self): self.sqs.query.backend.clear() settings.HAYSTACK_CONNECTIONS['whoosh']['STORAGE'] = self.old_whoosh_storage connections['whoosh']._index = self.old_ui super(LiveWhooshRamStorageTestCase, self).tearDown() def test_ram_storage(self): results = self.sqs.filter(id='core.mockmodel.1') # Sanity check. self.assertEqual(results.count(), 1) # Check the individual fields. result = results[0] self.assertEqual(result.id, 'core.mockmodel.1') self.assertEqual(result.text, 'This is some example text.') self.assertEqual(result.name, 'Mister Pants') self.assertEqual(result.is_active, True) self.assertEqual(result.post_count, 25) self.assertEqual(result.average_rating, 3.6) self.assertEqual(result.pub_date, datetime(2009, 11, 21, 0, 0)) self.assertEqual(result.created, datetime(2009, 11, 21, 21, 31, 00)) self.assertEqual(result.tags, ['staff', 'outdoor', 'activist', 'scientist']) self.assertEqual(result.sites, [u'3', u'5', u'1']) self.assertEqual(result.empty_list, [])