Coverage for tests/conftest.py: 96%
213 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-27 23:45 -0600
1"""Top-level conftest.py"""
3# pylint: disable=missing-function-docstring,redefined-outer-name,R0913
4import typing as t
5from os import environ
6from datetime import datetime, timezone
7import random
8import string
9import pytest
10from elasticsearch8.exceptions import NotFoundError
11from es_client import Builder
12from es_client.helpers.logging import set_logging
13from es_testbed.defaults import NAMEMAPPER
14from es_testbed.helpers.es_api import get_ds_current, get_write_index
16LOGLEVEL = 'DEBUG'
19@pytest.fixture(scope='class')
20def actual_index(entitymgr):
21 def _actual_index(tb, which):
22 if tb.plan.type == 'data_stream':
23 return entitymgr(tb).ds.backing_indices[which]
24 return entitymgr(tb).entity_list[which].name # implied else
26 return _actual_index
29@pytest.fixture(scope='class')
30def actual_rollover(entitymgr):
31 def _actual_rollover(tb):
32 if tb.plan.type == 'data_stream':
33 return entitymgr(tb).last
34 if tb.plan.rollover_alias:
35 if entitymgr(tb).alias.name is not None: 35 ↛ 37line 35 didn't jump to line 37, because the condition on line 35 was never false
36 return entitymgr(tb).alias.name
37 return '' # implied else
39 return _actual_rollover
42@pytest.fixture(scope='class')
43def actual_write_index(actual_rollover):
44 def _actual_write_index(tb):
45 name = actual_rollover(tb)
46 if not name:
47 return name
48 func = get_write_index
49 if tb.plan.type == 'data_stream':
50 func = get_ds_current
51 return func(tb.client, name)
53 return _actual_write_index
56@pytest.fixture(scope='session')
57def client():
58 """Return an Elasticsearch client"""
59 host = environ.get('TEST_ES_SERVER', 'http://127.0.0.1:9200')
60 file = environ.get('ES_CLIENT_FILE', None) # Path to es_client YAML config
61 if file: 61 ↛ 64line 61 didn't jump to line 64, because the condition on line 61 was never false
62 kwargs = {'configfile': file}
63 else:
64 kwargs = {'configdict': {'elasticsearch': {'client': {'hosts': host}}}}
65 set_logging({'loglevel': LOGLEVEL, 'blacklist': ['elastic_transport', 'urllib3']})
66 builder = Builder(**kwargs)
67 builder.connect()
68 return builder.client
71@pytest.fixture(scope='class')
72def cold():
73 """Return the prefix for cold indices"""
74 return 'restored-'
77@pytest.fixture(scope='class')
78def components(namecore):
79 """Return the component names in a list"""
80 components = []
81 components.append(f'{namecore("component")}-000001')
82 components.append(f'{namecore("component")}-000002')
83 return components
86@pytest.fixture(scope='class')
87def entity_count(defaults):
88 def _entity_count(kind):
89 if kind == 'data_stream':
90 return 1
91 return defaults()['entity_count']
93 return _entity_count
96@pytest.fixture(scope='class')
97def defaults() -> t.Dict:
98 def _defaults(sstier: str = 'hot') -> t.Dict:
99 retval = {'entity_count': 3, 'docs': 10, 'match': True, 'searchable': None}
100 if sstier in ['cold', 'frozen']:
101 retval['searchable'] = sstier
102 return retval
104 return _defaults
107@pytest.fixture(scope='class')
108def entitymgr():
109 def _entitymgr(tb):
110 if tb.plan.type == 'data_stream':
111 return tb.data_streammgr
112 return tb.indexmgr # implied else
114 return _entitymgr
117@pytest.fixture(scope='class')
118def first():
119 return 0
122@pytest.fixture(scope='class')
123def frozen():
124 """Return the prefix for frozen indices"""
125 return 'partial-'
128@pytest.fixture(scope='class')
129def get_template(template):
130 def _get_template(client):
131 return client.indices.get_index_template(name=template)['index_templates']
133 return _get_template
136@pytest.fixture(scope='class')
137def idxmain(namecore, ymd):
138 def _idxmain(kind):
139 result = f'{namecore(kind)}'
140 if kind == 'data_stream':
141 return f'.ds-{result}-{ymd}'
142 return result
144 return _idxmain
147@pytest.fixture(scope='class')
148def idxss(first, ssprefix, rollable):
149 def _idxss(tier, which, plan):
150 if which != first:
151 if rollable(plan):
152 return '' # No searchable prefix
153 return ssprefix(tier)
155 return _idxss
158@pytest.fixture(scope='class')
159def idxtail(first, last):
160 def _idxtail(which):
161 if which == first:
162 return '-000001'
163 if which == last: 163 ↛ 165line 163 didn't jump to line 165, because the condition on line 163 was never false
164 return '-000003'
165 return '-000002' # implied else
167 return _idxtail
170@pytest.fixture(scope='class')
171def index_name(first, idxmain, idxss, idxtail):
172 def _index_name(which=first, plan=None, tier: str = 'hot'):
173 prefix = idxss(tier, which, plan)
174 main = idxmain(plan.type)
175 suffix = idxtail(which)
176 return f'{prefix}{main}{suffix}'
178 return _index_name
181@pytest.fixture(scope='class')
182def last():
183 return -1
186@pytest.fixture(scope='class')
187def namecore(prefix, uniq):
188 def _namecore(kind):
189 if kind == 'indices':
190 return f'{prefix}-{NAMEMAPPER["index"]}-{uniq}'
191 return f'{prefix}-{NAMEMAPPER[kind]}-{uniq}'
193 return _namecore
196@pytest.fixture(scope='class')
197def prefix():
198 """Return a random prefix"""
199 return randomstr(length=8, lowercase=True)
202def randomstr(length: int = 16, lowercase: bool = False):
203 """Generate a random string"""
204 letters = string.ascii_uppercase
205 if lowercase: 205 ↛ 207line 205 didn't jump to line 207, because the condition on line 205 was never false
206 letters = string.ascii_lowercase
207 return str(''.join(random.choices(letters + string.digits, k=length)))
210@pytest.fixture(scope='class')
211def repo(client):
212 """Return the elasticsearch repository"""
213 name = environ.get('TEST_ES_REPO', 'found-snapshots') # Going with Cloud default
214 if not repo: 214 ↛ 215line 214 didn't jump to line 215, because the condition on line 214 was never true
215 return False
216 try:
217 client.snapshot.get_repository(name=name)
218 except NotFoundError:
219 return False
220 return name # Return the repo name if it's online
223@pytest.fixture(scope='class')
224def rollable():
225 def _rollable(plan):
226 if plan.type == 'data_stream':
227 return True
228 if plan.rollover_alias:
229 return True
230 return False
232 return _rollable
235@pytest.fixture(scope='class')
236def rollovername(namecore, rollable):
237 def _rollovername(plan):
238 if rollable(plan):
239 if plan.type == 'data_stream':
240 return namecore(plan.type)
241 return namecore('index')
242 return ''
244 return _rollovername
247@pytest.fixture(scope='class')
248def settings(defaults, prefix, repo, uniq):
249 def _settings(
250 plan_type: t.Literal['data_stream', 'index'] = 'data_stream',
251 rollover_alias: bool = False,
252 ilm: t.Union[t.Dict, False] = False,
253 sstier: str = 'hot',
254 ):
255 return {
256 'type': plan_type,
257 'prefix': prefix,
258 'rollover_alias': rollover_alias,
259 'repository': repo,
260 'uniq': uniq,
261 'ilm': ilm,
262 'defaults': defaults(sstier),
263 }
265 return _settings
268@pytest.fixture(scope='class')
269def skip_no_repo(repo) -> None:
270 def _skip_no_repo(skip_it: bool) -> None:
271 if skip_it:
272 if not repo: 272 ↛ 273line 272 didn't jump to line 273, because the condition on line 272 was never true
273 pytest.skip('No snapshot repository', allow_module_level=True)
275 return _skip_no_repo
278@pytest.fixture(scope='class')
279def ssprefix(cold, frozen):
280 def _ssprefix(tier):
281 retval = '' # hot or warm
282 if tier == 'cold':
283 retval = cold
284 if tier == 'frozen':
285 retval = frozen
286 return retval
288 return _ssprefix
291@pytest.fixture(scope='class')
292def template(namecore):
293 """Return the name of the index template"""
294 return f'{namecore("template")}-000001'
297@pytest.fixture(scope='class')
298def uniq():
299 """Return a random uniq value"""
300 return randomstr(length=8, lowercase=True)
303@pytest.fixture(scope='class')
304def write_index_name(last, idxmain, idxss, idxtail, rollable):
305 def _write_index_name(which=last, plan=None, tier: str = 'hot'):
306 if not rollable(plan):
307 return ''
308 prefix = idxss(tier, which, plan)
309 main = idxmain(plan.type)
310 suffix = idxtail(which)
311 return f'{prefix}{main}{suffix}'
313 return _write_index_name
316@pytest.fixture(scope='class')
317def ymd():
318 return datetime.now(timezone.utc).strftime('%Y.%m.%d')