forked from django-haystack/django-haystack
/
tests.py
175 lines (145 loc) · 6.04 KB
/
tests.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
import shutil
from django.conf import settings
from django.test import TestCase
from haystack import connections, connection_router
from haystack.exceptions import NotHandled
from haystack.query import SearchQuerySet
from haystack.utils.loading import UnifiedIndex
from multipleindex.search_indexes import FooIndex
from multipleindex.models import Foo, Bar
class MultipleIndexTestCase(TestCase):
def setUp(self):
super(MultipleIndexTestCase, self).setUp()
self.ui = connections['default'].get_unified_index()
self.fi = self.ui.get_index(Foo)
self.bi = self.ui.get_index(Bar)
self.solr_backend = connections['default'].get_backend()
self.whoosh_backend = connections['whoosh'].get_backend()
foo_1 = Foo.objects.create(
title='Haystack test',
body='foo 1',
)
foo_2 = Foo.objects.create(
title='Another Haystack test',
body='foo 2',
)
bar_1 = Bar.objects.create(
author='Haystack test',
content='bar 1',
)
bar_2 = Bar.objects.create(
author='Another Haystack test',
content='bar 2',
)
bar_3 = Bar.objects.create(
author='Yet another Haystack test',
content='bar 3',
)
self.fi.reindex(using='default')
self.fi.reindex(using='whoosh')
self.bi.reindex(using='default')
def tearDown(self):
self.fi.clear()
self.bi.clear()
# Because Whoosh doesn't clean up its mess.
shutil.rmtree(settings.HAYSTACK_CONNECTIONS['whoosh']['PATH'])
super(MultipleIndexTestCase, self).setUp()
def test_index_update_object_using(self):
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 2)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
foo_3 = Foo.objects.create(
title='Whee another Haystack test',
body='foo 3',
)
self.fi.update_object(foo_3)
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 3)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
self.fi.update_object(foo_3, using='whoosh')
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 3)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 3)
def test_index_remove_object_using(self):
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 2)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
foo_1 = Foo.objects.get(pk=1)
self.fi.remove_object(foo_1)
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 1)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
self.fi.remove_object(foo_1, using='whoosh')
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 1)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 1)
def test_index_clear_using(self):
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 2)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
self.fi.clear()
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 0)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
self.fi.clear(using='whoosh')
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 0)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 0)
def test_index_update_using(self):
self.fi.clear()
self.fi.clear(using='whoosh')
self.bi.clear()
self.bi.clear(using='whoosh')
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 0)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 0)
self.fi.update()
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 2)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 0)
self.fi.update(using='whoosh')
results = self.solr_backend.search('foo')
self.assertEqual(results['hits'], 2)
results = self.whoosh_backend.search('foo')
self.assertEqual(results['hits'], 2)
def test_searchqueryset_using(self):
# Using the default.
sqs = SearchQuerySet()
self.assertEqual(sqs.count(), 5)
self.assertEqual(sqs.models(Foo).count(), 2)
self.assertEqual(sqs.models(Bar).count(), 3)
self.assertEqual(sqs.using('default').count(), 5)
self.assertEqual(sqs.using('default').models(Foo).count(), 2)
self.assertEqual(sqs.using('default').models(Bar).count(), 3)
self.assertEqual(sqs.using('whoosh').count(), 2)
self.assertEqual(sqs.using('whoosh').models(Foo).count(), 2)
self.assertEqual(sqs.using('whoosh').models(Bar).count(), 0)
def test_searchquery_using(self):
sq = connections['default'].get_query()
# Using the default.
self.assertEqual(sq.get_count(), 5)
# "Swap" to the default.
sq = sq.using('default')
self.assertEqual(sq.get_count(), 5)
# Swap the ``SearchQuery`` used.
sq = sq.using('whoosh')
self.assertEqual(sq.get_count(), 2)
def test_excluded_indexes(self):
wui = connections['whoosh'].get_unified_index()
self.assertEqual(len(wui.collect_indexes()), 1)
self.assertTrue(isinstance(wui.collect_indexes()[0], FooIndex))
# Shouldn't error.
wui.get_index(Foo)
# Should error, since it's not present.
self.assertRaises(NotHandled, wui.get_index, Bar)