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
|
# This file is part of the django-environ.
#
# Copyright (c) 2021-2024, Serghei Iakovlev <oss@serghei.pl>
# Copyright (c) 2013-2021, Daniele Faraglia <daniele.faraglia@gmail.com>
#
# For the full copyright and license information, please view
# the LICENSE.txt file that was distributed with this source code.
import pytest
from environ import Env
def test_solr_parsing(solr_url):
url = Env.search_url_config(solr_url)
assert len(url) == 2
assert url['ENGINE'] == 'haystack.backends.solr_backend.SolrEngine'
assert url['URL'] == 'http://127.0.0.1:8983/solr'
def test_solr_multicore_parsing(solr_url):
timeout = 360
index = 'solr_index'
url = '{}/{}?TIMEOUT={}'.format(solr_url, index, timeout)
url = Env.search_url_config(url)
assert url['ENGINE'] == 'haystack.backends.solr_backend.SolrEngine'
assert url['URL'] == 'http://127.0.0.1:8983/solr/solr_index'
assert url['TIMEOUT'] == timeout
assert 'INDEX_NAME' not in url
assert 'PATH' not in url
@pytest.mark.parametrize(
'url,engine,scheme',
[
('elasticsearch://127.0.0.1:9200/index',
'elasticsearch_backend.ElasticsearchSearchEngine',
'http',),
('elasticsearchs://127.0.0.1:9200/index',
'elasticsearch_backend.ElasticsearchSearchEngine',
'https',),
('elasticsearch2://127.0.0.1:9200/index',
'elasticsearch2_backend.Elasticsearch2SearchEngine',
'http',),
('elasticsearch2s://127.0.0.1:9200/index',
'elasticsearch2_backend.Elasticsearch2SearchEngine',
'https',),
('elasticsearch5://127.0.0.1:9200/index',
'elasticsearch5_backend.Elasticsearch5SearchEngine',
'http'),
('elasticsearch5s://127.0.0.1:9200/index',
'elasticsearch5_backend.Elasticsearch5SearchEngine',
'https'),
('elasticsearch7://127.0.0.1:9200/index',
'elasticsearch7_backend.Elasticsearch7SearchEngine',
'http'),
('elasticsearch7s://127.0.0.1:9200/index',
'elasticsearch7_backend.Elasticsearch7SearchEngine',
'https'),
],
ids=[
'elasticsearch',
'elasticsearchs',
'elasticsearch2',
'elasticsearch2s',
'elasticsearch5',
'elasticsearch5s',
'elasticsearch7',
'elasticsearch7s',
]
)
def test_elasticsearch_parsing(url, engine, scheme):
"""Ensure all supported Elasticsearch engines are recognized."""
timeout = 360
url = '{}?TIMEOUT={}'.format(url, timeout)
url = Env.search_url_config(url)
assert url['ENGINE'] == 'haystack.backends.{}'.format(engine)
assert 'INDEX_NAME' in url.keys()
assert url['INDEX_NAME'] == 'index'
assert 'TIMEOUT' in url.keys()
assert url['TIMEOUT'] == timeout
assert 'PATH' not in url
assert url["URL"].startswith(scheme + ":")
def test_custom_search_engine():
"""Override ENGINE determined from schema."""
env_url = 'elasticsearch://127.0.0.1:9200/index'
engine = 'mypackage.backends.whatever'
url = Env.db_url_config(env_url, engine=engine)
assert url['ENGINE'] == engine
@pytest.mark.parametrize('storage', ['file', 'ram'])
def test_whoosh_parsing(whoosh_url, storage):
post_limit = 128 * 1024 * 1024
url = '{}?STORAGE={}&POST_LIMIT={}'.format(whoosh_url, storage, post_limit)
url = Env.search_url_config(url)
assert url['ENGINE'] == 'haystack.backends.whoosh_backend.WhooshEngine'
assert 'PATH' in url.keys()
assert url['PATH'] == '/home/search/whoosh_index'
assert 'STORAGE' in url.keys()
assert url['STORAGE'] == storage
assert 'POST_LIMIT' in url.keys()
assert url['POST_LIMIT'] == post_limit
assert 'INDEX_NAME' not in url
@pytest.mark.parametrize('flags', ['myflags'])
def test_xapian_parsing(xapian_url, flags):
url = '{}?FLAGS={}'.format(xapian_url, flags)
url = Env.search_url_config(url)
assert url['ENGINE'] == 'haystack.backends.xapian_backend.XapianEngine'
assert 'PATH' in url.keys()
assert url['PATH'] == '/home/search/xapian_index'
assert 'FLAGS' in url.keys()
assert url['FLAGS'] == flags
assert 'INDEX_NAME' not in url
def test_simple_parsing(simple_url):
url = Env.search_url_config(simple_url)
assert url['ENGINE'] == 'haystack.backends.simple_backend.SimpleEngine'
assert 'INDEX_NAME' not in url
assert 'PATH' not in url
def test_common_args_parsing(search_url):
excluded_indexes = 'myapp.indexes.A,myapp.indexes.B'
include_spelling = 1
batch_size = 100
params = 'EXCLUDED_INDEXES={}&INCLUDE_SPELLING={}&BATCH_SIZE={}'.format(
excluded_indexes,
include_spelling,
batch_size
)
url = '?'.join([search_url, params])
url = Env.search_url_config(url)
assert 'EXCLUDED_INDEXES' in url.keys()
assert 'myapp.indexes.A' in url['EXCLUDED_INDEXES']
assert 'myapp.indexes.B' in url['EXCLUDED_INDEXES']
assert 'INCLUDE_SPELLING' in url.keys()
assert url['INCLUDE_SPELLING']
assert 'BATCH_SIZE' in url.keys()
assert url['BATCH_SIZE'] == 100
|
