Coverage for laskea/config.py: 76.71%
255 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-10 22:19:18 +00:00
1"""Configuration API for laskea."""
2import copy
3import json
4import os
5import pathlib
6import sys
7from typing import Mapping, no_type_check
9import jmespath
11import laskea
12from laskea import log
13import laskea.api.jira as api
15TEMPLATE_EXAMPLE = """\
16{
17 "table": {
18 "column": {
19 "fields": [
20 "Key",
21 "Summary",
22 "Custom Field Name",
23 ["Custom Field Other", "Display Name"]
24 ],
25 "field_map": {
26 "key": [
27 "key",
28 "key"
29 ],
30 "summary": [
31 "summary",
32 "fields.summary"
33 ],
34 "custom field name": [
35 "customfield_11501",
36 "fields.customfield_11501"
37 ],
38 "custom field other": [
39 "customfield_13901",
40 "fields.customfield_13901[].value"
41 ]
42 },
43 "filter_map": {
44 "key": {},
45 "summary": {},
46 "custom field name": {
47 "order": ["keep", "drop", "replace"],
48 "keep": [
49 ["startswith", "ABC-"],
50 ["contains", "Z"],
51 ["icontains", "m"],
52 ["equals", "DEF-42"],
53 ["endswith", "-123"]
54 ],
55 "drop": [
56 ["matches", "[A-Z]+-\\d+"]
57 ],
58 "replace": [
59 ["DEF-", "definition-"]
60 ]
61 },
62 "custom field other": {}
63 },
64 "lf_only": true,
65 "join_string": " <br>"
66 },
67 "caption": "$NL$$NL$Table: Search '$QUERY_TEXT$' resulted in $ISSUE_COUNT$ issue$SINGULAR$$PLURAL$s$"
68 },
69 "remote": {
70 "is_cloud": false,
71 "user": "",
72 "token": "",
73 "base_url": "https://remote-jira-instance.example.com/"
74 },
75 "local": {
76 "markers": "[[[fill ]]] [[[end]]]",
77 "quiet": false,
78 "verbose": false,
79 "strict": false,
80 "checksums": false
81 },
82 "excel": {
83 "mbom": "mbom.xlsm"
84 },
85 "tabulator": {
86 "overview": {
87 "base_url": "https://example.com/metrics/",
88 "path": "$year$/kpi-table-$year$.json",
89 "years": [2022],
90 "matrix": [
91 ["section", "Section", False, "L"],
92 ["name", "Name", False, "L"],
93 ["unit", "Unit", False, "C"],
94 ["all", "ALL", True, "R"],
95 ["pr1", "PR1", True, "R"],
96 ["pr2", "PR2", True, "R"],
97 ["pr3", "PR3", True, "R"],
98 ["description", "Description", False, "L"]
99 ]
100 },
101 "metrics": {
102 "base_url": "https://example.com/metrics/",
103 "paths": {
104 "review_effectivity": "$year$/review_effectivity/kpi-review_effectivity-per_product-report-$year$.json",
105 "sprint_effectivity": "$year$/sprint_effectivity/kpi-sprint_effectivity-per_product-report-$year$.json",
106 "task_traceability": "$year$/task_traceability/kpi-task_traceability-per_product-report-$year$.json",
107 },
108 "years": [2021, 2022],
109 "matrix": [
110 ["month", "Month", False, "L"],
111 ["all", "ALL", True, "R"],
112 ["pr1", "PR1", True, "R"],
113 ["pr2", "PR2", True, "R"],
114 ["pr3", "PR3", True, "R"],
115 ["trend_all", "±ALL", True, "R"],
116 ["trend_pr1", "±PR1", True, "R"],
117 ["trend_pr2", "±PR2", True, "R"],
118 ["trend_pr3", "±PR3", True, "R"]
119 ]
120 }
121 }
122}
123"""
126def generate_template() -> str:
127 """Return template of a well-formed JSON configuration."""
128 return TEMPLATE_EXAMPLE
131@no_type_check
132def load_configuration(configuration: dict[str, object]) -> dict[str, str]:
133 """LaterAlligator."""
134 if not configuration:
135 log.warning('Warning: Requested load from empty configuration')
136 return {}
138 source_of = {}
140 column_fields = jmespath.search('table.column.fields', configuration)
141 if column_fields:
142 source_of['column_fields'] = 'config'
143 api.BASE_COL_FIELDS = copy.deepcopy(column_fields)
144 column_fields = os.getenv(f'{laskea.APP_ENV}_COL_FIELDS', '')
145 if column_fields: 145 ↛ 146line 145 didn't jump to line 146, because the condition on line 145 was never true
146 source_of['column_fields'] = 'env'
147 api.BASE_COL_FIELDS = json.loads(column_fields)
149 field_map = jmespath.search('table.column.field_map', configuration)
150 if field_map:
151 source_of['field_map'] = 'config'
152 api.BASE_COL_MAPS = copy.deepcopy(field_map)
153 field_map = os.getenv(f'{laskea.APP_ENV}_COL_MAPS', '')
154 if field_map: 154 ↛ 155line 154 didn't jump to line 155, because the condition on line 154 was never true
155 source_of['field_map'] = 'env'
156 api.BASE_COL_MAPS = json.loads(field_map)
158 filter_map = jmespath.search('table.column.filter_map', configuration)
159 if filter_map:
160 source_of['filter_map'] = 'config'
161 api.BASE_COL_FILTERS = copy.deepcopy(filter_map)
162 filter_map = os.getenv(f'{laskea.APP_ENV}_COL_FILTERS', '')
163 if filter_map: 163 ↛ 164line 163 didn't jump to line 164, because the condition on line 163 was never true
164 source_of['filter_map'] = 'env'
165 api.BASE_COL_FILTERS = json.loads(filter_map)
167 lf_only = jmespath.search('table.column.lf_only', configuration)
168 if lf_only:
169 source_of['lf_only'] = 'config'
170 api.BASE_LF_ONLY = lf_only
171 lf_only = os.getenv(f'{laskea.APP_ENV}_LF_ONLY', '')
172 if lf_only: 172 ↛ 173line 172 didn't jump to line 173, because the condition on line 172 was never true
173 source_of['lf_only'] = 'env'
174 api.BASE_LF_ONLY = lf_only
176 join_string = jmespath.search('table.column.join_string', configuration)
177 if join_string:
178 source_of['join_string'] = 'config'
179 api.BASE_JOIN_STRING = join_string
180 join_string = os.getenv(f'{laskea.APP_ENV}_JOIN_STRING', '')
181 if join_string: 181 ↛ 182line 181 didn't jump to line 182, because the condition on line 181 was never true
182 source_of['join_string'] = 'env'
183 api.BASE_JOIN_STRING = join_string
185 caption = jmespath.search('table.caption', configuration)
186 if caption:
187 source_of['caption'] = 'config'
188 api.BASE_CAPTION = caption
189 caption = os.getenv(f'{laskea.APP_ENV}_CAPTION', '')
190 if caption: 190 ↛ 191line 190 didn't jump to line 191, because the condition on line 190 was never true
191 source_of['caption'] = 'env'
192 api.BASE_CAPTION = caption
194 remote_user = jmespath.search('remote.user', configuration)
195 if remote_user: 195 ↛ 196line 195 didn't jump to line 196, because the condition on line 195 was never true
196 source_of['remote_user'] = 'config'
197 api.BASE_USER = remote_user
198 remote_user = os.getenv(f'{laskea.APP_ENV}_USER', '')
199 if remote_user: 199 ↛ 203line 199 didn't jump to line 203, because the condition on line 199 was never false
200 source_of['remote_user'] = 'env'
201 api.BASE_USER = remote_user
203 remote_token = jmespath.search('remote.token', configuration)
204 if remote_token:
205 source_of['remote_token'] = 'config' # nosec
206 api.BASE_TOKEN = remote_token
207 remote_token = os.getenv(f'{laskea.APP_ENV}_TOKEN', '')
208 if remote_token: 208 ↛ 212line 208 didn't jump to line 212, because the condition on line 208 was never false
209 source_of['remote_token'] = 'env' # nosec
210 api.BASE_TOKEN = remote_token
212 remote_base_url = jmespath.search('remote.base_url', configuration)
213 if remote_base_url:
214 source_of['remote_base_url'] = 'config'
215 api.BASE_URL = remote_base_url
216 remote_base_url = os.getenv(f'{laskea.APP_ENV}_BASE_URL', '')
217 if remote_base_url: 217 ↛ 221line 217 didn't jump to line 221, because the condition on line 217 was never false
218 source_of['remote_base_url'] = 'env'
219 api.BASE_URL = remote_base_url
221 local_markers = jmespath.search('local.markers', configuration)
222 if local_markers:
223 source_of['local_markers'] = 'config'
224 laskea.BASE_MARKERS = local_markers
225 local_markers = os.getenv(f'{laskea.APP_ENV}_MARKERS', '')
226 if local_markers: 226 ↛ 227line 226 didn't jump to line 227, because the condition on line 226 was never true
227 source_of['local_markers'] = 'env'
228 laskea.BASE_MARKERS = local_markers
230 verbose = bool(jmespath.search('local.verbose', configuration))
231 if verbose: 231 ↛ 232line 231 didn't jump to line 232, because the condition on line 231 was never true
232 source_of['verbose'] = 'config'
233 laskea.DEBUG = verbose
234 verbose = bool(os.getenv(f'{laskea.APP_ENV}_DEBUG', ''))
235 if verbose: 235 ↛ 236line 235 didn't jump to line 236, because the condition on line 235 was never true
236 source_of['verbose'] = 'env'
237 laskea.DEBUG = verbose
239 is_cloud = bool(jmespath.search('remote.is_cloud', configuration))
240 if is_cloud: 240 ↛ 241line 240 didn't jump to line 241, because the condition on line 240 was never true
241 source_of['is_cloud'] = 'config'
242 laskea.IS_CLOUD = is_cloud
243 is_cloud = bool(os.getenv(f'{laskea.APP_ENV}_IS_CLOUD', ''))
244 if is_cloud: 244 ↛ 248line 244 didn't jump to line 248, because the condition on line 244 was never false
245 source_of['is_cloud'] = 'env'
246 laskea.IS_CLOUD = is_cloud
248 strict = bool(jmespath.search('local.strict', configuration))
249 if strict: 249 ↛ 250line 249 didn't jump to line 250, because the condition on line 249 was never true
250 source_of['strict'] = 'config'
251 laskea.STRICT = strict
252 strict = bool(os.getenv(f'{laskea.APP_ENV}_STRICT', ''))
253 if strict: 253 ↛ 254line 253 didn't jump to line 254, because the condition on line 253 was never true
254 source_of['strict'] = 'env'
255 laskea.STRICT = strict
257 checksums = bool(jmespath.search('local.checksums', configuration))
258 if checksums: 258 ↛ 259line 258 didn't jump to line 259, because the condition on line 258 was never true
259 source_of['checksums'] = 'config'
260 laskea.CHECKSUMS = checksums
261 checksums = bool(os.getenv(f'{laskea.APP_ENV}_CHECKSUMS', ''))
262 if checksums: 262 ↛ 263line 262 didn't jump to line 263, because the condition on line 262 was never true
263 source_of['checksums'] = 'env'
264 laskea.CHECKSUMS = checksums
266 quiet = bool(jmespath.search('local.quiet', configuration))
267 if quiet: 267 ↛ 268line 267 didn't jump to line 268, because the condition on line 267 was never true
268 source_of['quiet'] = 'config'
269 laskea.QUIET = quiet
270 if source_of['verbose'] == 'config':
271 laskea.DEBUG = quiet
272 quiet = bool(os.getenv(f'{laskea.APP_ENV}_QUIET', ''))
273 if quiet: 273 ↛ 274line 273 didn't jump to line 274, because the condition on line 273 was never true
274 source_of['quiet'] = 'env'
275 laskea.QUIET = quiet
276 source_of['verbose'] = 'env'
277 laskea.DEBUG = quiet
279 if 'tabulator' in configuration: 279 ↛ 280line 279 didn't jump to line 280, because the condition on line 279 was never true
280 laskea.TABULATOR = copy.deepcopy(configuration['tabulator'])
282 return source_of
285@no_type_check
286def discover_configuration(conf: str) -> tuple[dict[str, object], str]:
287 """Try to retrieve the configuration following the "(explicit, local, parents, home)
288 first wun wins" strategy."""
289 configuration = None
290 if conf:
291 cp = pathlib.Path(conf)
292 if not cp.is_file() or not cp.stat().st_size:
293 print('Given configuration path is no file or empty', file=sys.stderr)
294 sys.exit(2)
295 if not laskea.QUIET: 295 ↛ 297line 295 didn't jump to line 297, because the condition on line 295 was never false
296 print(f'Reading configuration file {cp} as requested...', file=sys.stderr)
297 with cp.open(encoding=laskea.ENCODING) as handle:
298 configuration = json.load(handle)
299 else:
300 cn = laskea.DEFAULT_CONFIG_NAME
301 cwd = pathlib.Path.cwd().resolve()
302 for pp in (cwd, *cwd.parents): 302 ↛ 311line 302 didn't jump to line 311, because the loop on line 302 didn't complete
303 cp = pp / cn
304 if cp.is_file() and cp.stat().st_size:
305 if not laskea.QUIET: 305 ↛ 307line 305 didn't jump to line 307, because the condition on line 305 was never false
306 print(f'Reading from discovered configuration path {cp}', file=sys.stderr)
307 with cp.open() as handle:
308 configuration = json.load(handle)
309 return configuration, str(cp)
311 cp = pathlib.Path.home() / laskea.DEFAULT_CONFIG_NAME
312 if cp.is_file() and cp.stat().st_size:
313 if not laskea.QUIET:
314 print(
315 f'Reading configuration file {cp} from home directory at {pathlib.Path.home()} ...',
316 file=sys.stderr,
317 )
318 with cp.open() as handle:
319 configuration = json.load(handle)
320 return configuration, str(cp)
322 if not laskea.QUIET:
323 print(
324 f'User home configuration path to {cp} is no file or empty - ignoring configuration data',
325 file=sys.stderr,
326 )
328 return configuration, str(cp)
331@no_type_check
332def report_context(command: str, transaction_mode: str, vector: list[str]) -> None:
333 """DRY."""
334 if laskea.QUIET:
335 return
336 print(f'Command: ({command})', file=sys.stderr)
337 print(f'- Transaction mode: ({transaction_mode})', file=sys.stderr)
338 print('Environment(variable values):', file=sys.stderr)
339 app_env_user = f'{laskea.APP_ENV}_USER'
340 app_env_token = f'{laskea.APP_ENV}_TOKEN'
341 app_env_base_url = f'{laskea.APP_ENV}_BASE_URL'
342 app_env_col_fields = f'{laskea.APP_ENV}_COL_FIELDS'
343 app_env_col_maps = f'{laskea.APP_ENV}_COL_MAPS'
344 app_env_markers = f'{laskea.APP_ENV}_MARKERS'
345 app_env_lf_only = f'{laskea.APP_ENV}_LF_ONLY'
346 app_env_caption = f'{laskea.APP_ENV}_CAPTION'
347 app_env_join_string = f'{laskea.APP_ENV}_JOIN_STRING'
348 empty = ''
349 print(f'- {laskea.APP_ENV}_USER: ({os.getenv(app_env_user, empty)})', file=sys.stderr)
350 print(
351 f'- {laskea.APP_ENV}_TOKEN: ({laskea.MASK_DISPLAY if len(os.getenv(app_env_token, empty)) else empty})',
352 file=sys.stderr,
353 )
354 print(f'- {laskea.APP_ENV}_BASE_URL: ({os.getenv(app_env_base_url, empty)})', file=sys.stderr)
355 print(f'- {laskea.APP_ENV}_COL_FIELDS: ({os.getenv(app_env_col_fields, empty)})', file=sys.stderr)
356 print(f'- {laskea.APP_ENV}_COL_MAPS: ({os.getenv(app_env_col_maps, empty)})', file=sys.stderr)
357 print(f'- {laskea.APP_ENV}_MARKERS: ({os.getenv(app_env_markers, empty)})', file=sys.stderr)
358 print(f'- {laskea.APP_ENV}_LF_ONLY: ({os.getenv(app_env_lf_only, empty)})', file=sys.stderr)
359 print(f'- {laskea.APP_ENV}_CAPTION: ({os.getenv(app_env_caption, empty)})', file=sys.stderr)
360 print(f'- {laskea.APP_ENV}_JOIN_STRING: ({os.getenv(app_env_join_string, empty)})', file=sys.stderr)
361 print('Effective(variable values):', file=sys.stderr)
362 print(f'- RemoteUser: ({api.BASE_USER})', file=sys.stderr)
363 print(f'- RemoteToken: ({"*" * len(api.BASE_PASS)})', file=sys.stderr)
364 print(f'- RemoteBaseURL: ({api.BASE_URL})', file=sys.stderr)
365 print(f'- ColumnFields(table): ({api.BASE_COL_FIELDS})', file=sys.stderr)
366 print(f'- ColumnMaps(remote->table): ({api.BASE_COL_MAPS})', file=sys.stderr)
367 print(f'- ColumnFilters(remote->table): ({api.BASE_COL_FILTERS})', file=sys.stderr)
368 print(f'- Markers(pattern): ({laskea.BASE_MARKERS})', file=sys.stderr)
369 print(f'- lf_only: ({laskea.BASE_LF_ONLY})', file=sys.stderr)
370 print(f'- caption: ({laskea.BASE_CAPTION})', file=sys.stderr)
371 print(f'- join_string: ({laskea.BASE_JOIN_STRING})', file=sys.stderr)
372 print(f'- CallVector: ({vector})', file=sys.stderr)
375@no_type_check
376def report_sources_of_effective_configuration(source_of: dict[str, str], header: str) -> None:
377 """DRY."""
378 if laskea.QUIET:
379 return
380 print(header, file=sys.stderr)
381 print('# --- BEGIN ---', file=sys.stderr)
382 print(json.dumps(source_of, indent=2), file=sys.stderr)
383 print('# --- E N D ---', file=sys.stderr)
386@no_type_check
387def safe_report_configuration(configuration: dict[str, object], header: str) -> None:
388 """DRY."""
389 if laskea.QUIET:
390 return
391 print(header, file=sys.stderr)
392 print('# --- BEGIN ---', file=sys.stderr)
393 fake_configuration = copy.deepcopy(configuration)
394 if jmespath.search('remote.token', fake_configuration):
395 fake_configuration['remote']['token'] = laskea.MASK_DISPLAY # noqa
396 print(json.dumps(fake_configuration, indent=2), file=sys.stderr)
397 print('# --- E N D ---', file=sys.stderr)
400@no_type_check
401def create_and_report_effective_configuration(header: str) -> None:
402 """DRY."""
403 if laskea.QUIET:
404 return
405 effective = {
406 'table': {
407 'column': {
408 'fields': copy.deepcopy(api.BASE_COL_FIELDS),
409 'field_map': copy.deepcopy(api.BASE_COL_MAPS),
410 'filter_map': copy.deepcopy(api.BASE_COL_FILTERS),
411 'lf_only': api.BASE_LF_ONLY,
412 'join_string': api.BASE_JOIN_STRING,
413 },
414 'caption': api.BASE_CAPTION,
415 },
416 'remote': {
417 'is_cloud': api.BASE_IS_CLOUD,
418 'user': api.BASE_USER,
419 'token': '',
420 'base_url': api.BASE_URL,
421 },
422 'local': {
423 'markers': laskea.BASE_MARKERS,
424 'quiet': laskea.QUIET,
425 'verbose': laskea.DEBUG,
426 'strict': laskea.STRICT,
427 'checksums': laskea.CHECKSUMS,
428 },
429 }
430 safe_report_configuration(effective, header)
433def process(conf: str, options: Mapping[str, bool]) -> None:
434 """SPOC."""
435 configuration, cp = discover_configuration(conf if isinstance(conf, str) else '')
437 verbose = bool(options.get('verbose', ''))
438 if configuration is not None: 438 ↛ 454line 438 didn't jump to line 454, because the condition on line 438 was never false
439 if laskea.DEBUG or verbose:
440 safe_report_configuration(configuration, f'Loaded configuration from {cp}:')
442 source_of = load_configuration(configuration)
444 if laskea.DEBUG or verbose:
445 report_sources_of_effective_configuration(source_of, f'Configuration source after loading from {cp}:')
447 if not laskea.QUIET: 447 ↛ 450line 447 didn't jump to line 450, because the condition on line 447 was never false
448 print('Configuration interface combined file, environment, and commandline values!', file=sys.stderr)
450 create_and_report_effective_configuration(
451 f'Effective configuration combining {cp}, environment variables, and defaults:'
452 )
454 if laskea.DEBUG or verbose:
455 log.info(f'Upstream JIRA instance is addressed per {"cloud" if api.BASE_IS_CLOUD else "server"} rules')