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