Coverage for liitos/gather.py: 85.09%

1"""Gather the structure and discover the content.""" 

2 

3import json 

4import os 

5import pathlib 

6from typing import Dict, List, Set, Tuple, Union 

7 

8import yaml 

9 

10from liitos import ( 

11 DEFAULT_STRUCTURE_NAME, 

12 KEY_APPROVALS, 

13 KEY_BIND, 

14 KEY_CHANGES, 

15 KEY_LAYOUT, 

16 KEY_META, 

17 KEYS_REQUIRED, 

18 ENCODING, 

19 OptionsType, 

20 log, 

21) 

22 

23PathLike = Union[str, pathlib.Path] 

24 

25Approvals = Dict[str, Union[List[str], List[List[str]]]] 

26Assets = Dict[str, Dict[str, Dict[str, str]]] 

27Binder = List[str] 

28Changes = Dict[str, Union[List[str], List[List[str]]]] 

29Layout = Dict[str, str] 

30Meta = Dict[str, str] 

31Structure = Dict[str, List[Dict[str, str]]] 

32Targets = Set[str] 

33Facets = Dict[str, Targets] 

34Payload = Union[Approvals, Binder, Changes, Meta] 

35Verification = Tuple[bool, str] 

36 

37 

38def load_structure(path: PathLike = DEFAULT_STRUCTURE_NAME) -> Structure: 

39 """Load the structure information and content links from the YAML file per convention.""" 

40 with open(path, 'rt', encoding=ENCODING) as handle: 

41 return yaml.safe_load(handle) # type: ignore 

42 

43 

44def targets(structure: Structure) -> Targets: 

45 """Extract the targets from the given structure information item.""" 

46 return set(target for target in structure) 

47 

48 

49def facets(structure: Structure) -> Facets: 

50 """Extract the facets per target from the given structure information item.""" 

51 return {target: set(facet for facet_data in cnt for facet in facet_data) for target, cnt in structure.items()} 

52 

53 

54def assets(structure: Structure) -> Assets: 

55 """Map the assets to facets of targets.""" 

56 return {t: {f: asset for fd in cnt for f, asset in fd.items()} for t, cnt in structure.items()} # type: ignore 

57 

58 

59def verify_target(name: str, target_set: Targets) -> Verification: 

60 """Verify presence of target yielding predicate and message (in case of failure).""" 

61 return (True, '') if name in target_set else (False, f'target ({name}) not in {sorted(target_set)}') 

62 

63 

64def verify_facet(name: str, target: str, facet_map: Facets) -> Verification: 

65 """Verify presence of facet for target yielding predicate and message (in case of failure).""" 

66 if name in facet_map[target]: 

67 return True, '' 

68 return False, f'facet ({name}) of target ({target}) not in {sorted(facet_map[target])}' 

69 

70 

71def error_context( 

72 payload: Payload, 

73 label: str, 

74 facet: str, 

75 target: str, 

76 path: PathLike, 

77 err: Union[FileNotFoundError, KeyError, ValueError], 

78) -> Tuple[Payload, str]: 

79 """Provide harmonized context for the error situation as per parameters.""" 

80 if isinstance(err, FileNotFoundError): 

81 return payload, f'{label} link not found at ({path}) or invalid for facet ({facet}) of target ({target})' 

82 if isinstance(err, KeyError): 

83 return [], f'{label} not found in assets for facet ({facet}) of target ({target})' 

84 if isinstance(err, ValueError): 

85 return [], f'{label} requires json or yaml format in assets for facet ({facet}) of target ({target})' 

86 raise NotImplementedError(f'error context not implemented for error ({err})') 

87 

88 

89def load_binder(facet: str, target: str, path: PathLike) -> Tuple[Binder, str]: 

90 """Yield the binder for facet of target from path and message (in case of failure).""" 

91 try: 

92 with open(path, 'rt', encoding=ENCODING) as handle: 

93 return [line.strip() for line in handle.readlines() if line.strip()], '' 

94 except FileNotFoundError as err: 

95 return error_context([], 'Binder', facet, target, path, err) # type: ignore 

96 

97 

98def binder(facet: str, target: str, asset_struct: Assets) -> Tuple[Binder, str]: 

99 """Yield the binder for facet of target from link in assets and message (in case of failure).""" 

100 try: 

101 path = pathlib.Path(asset_struct[target][facet][KEY_BIND]) 

102 except KeyError as err: 

103 return error_context([], 'Binder', facet, target, '', err) # type: ignore 

104 return load_binder(facet, target, path) 

105 

106 

107def load_layout(facet: str, target: str, path: PathLike) -> Tuple[Meta, str]: 

108 """Yield the layout for facet of target from path and message (in case of failure).""" 

109 try: 

110 with open(path, 'rt', encoding=ENCODING) as handle: 

111 return yaml.safe_load(handle), '' 

112 except FileNotFoundError as err: 

113 return error_context({}, 'Metadata', facet, target, path, err) # type: ignore 

114 

115 

116def layout(facet: str, target: str, asset_struct: Assets) -> Tuple[Meta, str]: 

117 """Yield the layout for facet of target from link in assets and message (in case of failure).""" 

118 try: 

119 path = pathlib.Path(asset_struct[target][facet][KEY_LAYOUT]) 

120 except KeyError as err: 

121 return error_context({}, 'Layout', facet, target, '', err) # type: ignore 

122 return load_layout(facet, target, path) 

123 

124 

125def load_meta(facet: str, target: str, path: PathLike) -> Tuple[Meta, str]: 

126 """Yield the metadata for facet of target from path and message (in case of failure).""" 

127 try: 

128 with open(path, 'rt', encoding=ENCODING) as handle: 

129 return yaml.safe_load(handle), '' 

130 except FileNotFoundError as err: 

131 return error_context({}, 'Metadata', facet, target, path, err) # type: ignore 

132 

133 

134def meta(facet: str, target: str, asset_struct: Assets) -> Tuple[Meta, str]: 

135 """Yield the metadata for facet of target from link in assets and message (in case of failure).""" 

136 try: 

137 path = pathlib.Path(asset_struct[target][facet][KEY_META]) 

138 except KeyError as err: 

139 return error_context({}, 'Metadata', facet, target, '', err) # type: ignore 

140 return load_meta(facet, target, path) 

141 

142 

143def load_approvals(facet: str, target: str, path: PathLike) -> Tuple[Approvals, str]: 

144 """Yield the approvals for facet of target from path and message (in case of failure).""" 

145 if str(path).lower().endswith('json'): 

146 try: 

147 with open(path, 'rt', encoding=ENCODING) as handle: 

148 return json.load(handle), '' 

149 except FileNotFoundError as err: 

150 return error_context({}, 'Approvals', facet, target, path, err) # type: ignore 

151 elif str(path).lower().endswith(('yaml', 'yml')): 151 ↛ 158line 151 didn't jump to line 158 because the condition on line 151 was always true

152 try: 

153 with open(path, 'rt', encoding=ENCODING) as handle: 

154 return yaml.safe_load(handle), '' 

155 except FileNotFoundError as err: 

156 return error_context({}, 'Approvals', facet, target, path, err) # type: ignore 

157 

158 return error_context({}, 'Approvals', facet, target, path, ValueError('json or yaml required')) # type: ignore 

159 

160 

161def approvals(facet: str, target: str, asset_struct: Assets) -> Tuple[Approvals, str]: 

162 """Yield the approvals for facet of target from link in assets and message (in case of failure).""" 

163 try: 

164 path = pathlib.Path(asset_struct[target][facet][KEY_APPROVALS]) 

165 except KeyError as err: 

166 return error_context({}, 'Approvals', facet, target, '', err) # type: ignore 

167 return load_approvals(facet, target, path) 

168 

169 

170def load_changes(facet: str, target: str, path: PathLike) -> Tuple[Approvals, str]: 

171 """Yield the changes for facet of target from path and message (in case of failure).""" 

172 if str(path).lower().endswith('json'): 

173 try: 

174 with open(path, 'rt', encoding=ENCODING) as handle: 

175 return json.load(handle), '' 

176 except FileNotFoundError as err: 

177 return error_context({}, 'Changes', facet, target, path, err) # type: ignore 

178 elif str(path).lower().endswith(('yaml', 'yml')): 

179 try: 

180 with open(path, 'rt', encoding=ENCODING) as handle: 

181 return yaml.safe_load(handle), '' 

182 except FileNotFoundError as err: 

183 return error_context({}, 'Changes', facet, target, path, err) # type: ignore 

184 

185 return error_context({}, 'Changes', facet, target, path, ValueError('json or yaml required')) # type: ignore 

186 

187 

188def changes(facet: str, target: str, asset_struct: Assets) -> Tuple[Changes, str]: 

189 """Yield the changes for facet of target from link in assets and message (in case of failure).""" 

190 try: 

191 path = pathlib.Path(asset_struct[target][facet][KEY_CHANGES]) 

192 except KeyError as err: 

193 return error_context({}, 'Changes', facet, target, '', err) # type: ignore 

194 return load_changes(facet, target, path) 

195 

196 

197def verify_asset_keys(facet: str, target: str, asset_struct: Assets) -> Verification: 

198 """Verify presence of required keys for facet of target yielding predicate and message (in case of failure).""" 

199 if all(key in asset_struct[target][facet] for key in KEYS_REQUIRED): 

200 return True, '' 

201 return False, f'keys in {sorted(KEYS_REQUIRED)} for facet ({facet}) of target ({target}) are missing' 

202 

203 

204def verify_asset_links(facet: str, target: str, asset_struct: Assets) -> Verification: 

205 """Verify presence of asset links for facet of target yielding predicate and message (in case of failure).""" 

206 predicate, message = verify_asset_keys(facet, target, asset_struct) 

207 if not predicate: 

208 return predicate, message 

209 for key in KEYS_REQUIRED: 

210 link = pathlib.Path(asset_struct[target][facet][key]) 

211 log.debug(f' + verifying: {pathlib.Path.cwd() / link}') 

212 if not link.is_file() or not link.stat().st_size: 

213 return False, f'{key} asset link ({link}) for facet ({facet}) of target ({target}) is invalid' 

214 return True, '' 

215 

216 

217ASSET_KEY_ACTION = { 

218 KEY_APPROVALS: approvals, 

219 KEY_BIND: binder, 

220 KEY_CHANGES: changes, 

221 KEY_LAYOUT: layout, 

222 KEY_META: meta, 

223} 

224 

225 

226def verify_assets(facet: str, target: str, asset_struct: Assets) -> Verification: 

227 """Verify assets for facet of target yielding predicate and message (in case of failure).""" 

228 predicate, message = verify_asset_links(facet, target, asset_struct) 

229 if not predicate: 

230 return predicate, message 

231 for key, action in ASSET_KEY_ACTION.items(): 231 ↛ 235line 231 didn't jump to line 235 because the loop on line 231 didn't complete

232 asset, message = action(facet, target, asset_struct) 

233 if not asset: 

234 return False, f'{key} asset for facet ({facet}) of target ({target}) is invalid' 

235 return True, '' 

236 

237 

238def prelude( 

239 doc_root: Union[str, pathlib.Path], structure_name: str, target_key: str, facet_key: str, command: str 

240) -> tuple[Structure, Assets]: 

241 """DRY.""" 

242 doc_root = pathlib.Path(doc_root) 

243 idem = os.getcwd() 

244 os.chdir(doc_root) 

245 job_description = ( 

246 f'facet ({facet_key}) of target ({target_key}) with structure map ({structure_name})' 

247 f' in document root ({doc_root}) coming from ({idem})' 

248 ) 

249 log.info(f'executing prelude of command ({command}) for {job_description}') 

250 structure = load_structure(structure_name) 

251 asset_map = assets(structure) 

252 return structure, asset_map 

253 

254 

255def verify( 

256 doc_root: Union[str, pathlib.Path], 

257 structure_name: str, 

258 target_key: str, 

259 facet_key: str, 

260 options: OptionsType, 

261) -> int: 

262 """Drive the verification.""" 

263 doc_root = pathlib.Path(doc_root) 

264 os.chdir(doc_root) 

265 facet = facet_key 

266 target = target_key 

267 structure_name = structure_name 

268 job_description = ( 

269 f'facet ({facet}) for target ({target}) with structure map ({structure_name})' f' in document root ({doc_root})' 

270 ) 

271 log.info(f'starting verification of {job_description}') 

272 structure = load_structure(structure_name) 

273 target_set = targets(structure) 

274 facet_map = facets(structure) 

275 asset_map = assets(structure) 

276 

277 predicate, message = verify_target(target, target_set) 

278 if not predicate: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true

279 log.error(f'failed verification with: {message}') 

280 return 1 

281 log.info(f'- target ({target}) OK') 

282 

283 predicate, message = verify_facet(facet, target, facet_map) 

284 if not predicate: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true

285 log.error(f'failed verification with: {message}') 

286 return 1 

287 log.info(f'- facet ({facet}) of target ({target}) OK') 

288 

289 predicate, message = verify_assets(facet, target, asset_map) 

290 if not predicate: 290 ↛ 293line 290 didn't jump to line 293 because the condition on line 290 was always true

291 log.error(f'failed verification with: {message}') 

292 return 1 

293 log.info(f'- assets ({", ".join(sorted(KEYS_REQUIRED))}) for facet ({facet}) of target ({target}) OK') 

294 

295 signatures_path = asset_map[target][facet][KEY_APPROVALS] 

296 log.info(f'loading signatures from {signatures_path=}') 

297 signatures = load_approvals(facet, target, signatures_path) 

298 log.info(f'{signatures=}') 

299 history_path = asset_map[target][facet][KEY_CHANGES] 

300 log.info(f'loading history from {history_path=}') 

301 history = load_changes(facet, target, history_path) 

302 log.info(f'{history=}') 

303 

304 layout_path = asset_map[target][facet][KEY_LAYOUT] 

305 log.info(f'loading layout from {layout_path=}') 

306 design = load_layout(facet, target, layout_path) 

307 log.info(f'{design=}') 

308 

309 metadata_path = asset_map[target][facet][KEY_META] 

310 log.info(f'loading metadata from {metadata_path=}') 

311 info = load_meta(facet, target, metadata_path) 

312 

313 log.info(f'{info=}') 

314 log.info('successful verification') 

315 return 0