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 

8 

9import jmespath 

10 

11import laskea 

12import laskea.api.jira as api 

13 

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""" 

100 

101 

102def generate_template() -> str: 

103 """Return template of a well-formed JSON configuration.""" 

104 return TEMPLATE_EXAMPLE 

105 

106 

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 {} 

113 

114 source_of = {} 

115 

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) 

124 

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) 

133 

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 

142 

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 

151 

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 

160 

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 

169 

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 

178 

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 

187 

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 

196 

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 

205 

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 

214 

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 

227 

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']) 

230 

231 return source_of 

232 

233 

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) 

259 

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) 

270 

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 ) 

276 

277 return configuration, str(cp) 

278 

279 

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) 

318 

319 

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) 

329 

330 

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) 

343 

344 

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) 

373 

374 

375def process(conf: str, options: Mapping[str, bool]) -> None: 

376 """SPOC.""" 

377 configuration, cp = discover_configuration(conf if isinstance(conf, str) else '') 

378 

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}:') 

383 

384 source_of = load_configuration(configuration) 

385 

386 if laskea.DEBUG or verbose: 

387 report_sources_of_effective_configuration(source_of, f'Configuration source after loading from {cp}:') 

388 

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) 

391 

392 create_and_report_effective_configuration( 

393 f'Effective configuration combining {cp}, environment variables, and defaults:' 

394 ) 

395 

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 )