Coverage for liitos/gather.py: 84.47%

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

2 

3import os 

4import pathlib 

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

6 

7import yaml 

8 

9from liitos import ( 

10 DEFAULT_STRUCTURE_NAME, 

11 KEY_APPROVALS, 

12 KEY_BIND, 

13 KEY_CHANGES, 

14 KEY_LAYOUT, 

15 KEY_META, 

16 KEYS_REQUIRED, 

17 ENCODING, 

18 OptionsType, 

19 log, 

20) 

21 

22PathLike = Union[str, pathlib.Path] 

23 

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

25Assets = Dict[str, Dict[str, Dict[str, str]]] 

26Binder = List[str] 

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

28Layout = Dict[str, str] 

29Meta = Dict[str, str] 

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

31Targets = Set[str] 

32Facets = Dict[str, Targets] 

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

34Verification = Tuple[bool, str] 

35 

36 

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

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

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

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

41 

42 

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

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

45 return set(target for target in structure) 

46 

47 

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

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

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

51 

52 

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

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

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

56 

57 

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

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

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

61 

62 

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

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

65 if name in facet_map[target]: 

66 return True, '' 

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

68 

69 

70def error_context( 

71 payload: Payload, 

72 label: str, 

73 facet: str, 

74 target: str, 

75 path: PathLike, 

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

77) -> Tuple[Payload, str]: 

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

79 if isinstance(err, FileNotFoundError): 

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

81 if isinstance(err, KeyError): 

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

83 if isinstance(err, ValueError): 

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

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

86 

87 

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

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

90 try: 

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

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

93 except FileNotFoundError as err: 

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

95 

96 

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

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

99 try: 

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

101 except KeyError as err: 

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

103 return load_binder(facet, target, path) 

104 

105 

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

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

108 try: 

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

110 return yaml.safe_load(handle), '' 

111 except FileNotFoundError as err: 

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

113 

114 

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

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

117 try: 

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

119 except KeyError as err: 

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

121 return load_layout(facet, target, path) 

122 

123 

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

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

126 try: 

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

128 return yaml.safe_load(handle), '' 

129 except FileNotFoundError as err: 

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

131 

132 

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

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

135 try: 

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

137 except KeyError as err: 

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

139 return load_meta(facet, target, path) 

140 

141 

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

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

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

145 return error_context( 

146 {}, 'Approvals', facet, target, path, ValueError('please transform approvals from json to yaml format') 

147 ) # type: ignore 

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

149 try: 

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

151 return yaml.safe_load(handle), '' 

152 except FileNotFoundError as err: 

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

154 

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

156 

157 

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

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

160 try: 

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

162 except KeyError as err: 

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

164 return load_approvals(facet, target, path) 

165 

166 

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

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

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

170 return error_context( 

171 {}, 'Changes', facet, target, path, ValueError('please transform changes from json to yaml format') 

172 ) # type: ignore 

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

174 try: 

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

176 return yaml.safe_load(handle), '' 

177 except FileNotFoundError as err: 

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

179 

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

181 

182 

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

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

185 try: 

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

187 except KeyError as err: 

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

189 return load_changes(facet, target, path) 

190 

191 

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

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

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

195 return True, '' 

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

197 

198 

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

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

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

202 if not predicate: 

203 return predicate, message 

204 for key in KEYS_REQUIRED: 

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

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

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

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

209 return True, '' 

210 

211 

212ASSET_KEY_ACTION = { 

213 KEY_APPROVALS: approvals, 

214 KEY_BIND: binder, 

215 KEY_CHANGES: changes, 

216 KEY_LAYOUT: layout, 

217 KEY_META: meta, 

218} 

219 

220 

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

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

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

224 if not predicate: 

225 return predicate, message 

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

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

228 if not asset: 

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

230 return True, '' 

231 

232 

233def prelude( 

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

235) -> tuple[Structure, Assets]: 

236 """DRY.""" 

237 doc_root = pathlib.Path(doc_root) 

238 idem = os.getcwd() 

239 os.chdir(doc_root) 

240 job_description = ( 

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

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

243 ) 

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

245 structure = load_structure(structure_name) 

246 asset_map = assets(structure) 

247 return structure, asset_map 

248 

249 

250def verify( 

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

252 structure_name: str, 

253 target_key: str, 

254 facet_key: str, 

255 options: OptionsType, 

256) -> int: 

257 """Drive the verification.""" 

258 doc_root = pathlib.Path(doc_root) 

259 os.chdir(doc_root) 

260 facet = facet_key 

261 target = target_key 

262 structure_name = structure_name 

263 job_description = ( 

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

265 ) 

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

267 structure = load_structure(structure_name) 

268 target_set = targets(structure) 

269 facet_map = facets(structure) 

270 asset_map = assets(structure) 

271 

272 predicate, message = verify_target(target, target_set) 

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

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

275 return 1 

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

277 

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

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

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

281 return 1 

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

283 

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

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

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

287 return 1 

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

289 

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

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

292 signatures = load_approvals(facet, target, signatures_path) 

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

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

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

296 history = load_changes(facet, target, history_path) 

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

298 

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

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

301 design = load_layout(facet, target, layout_path) 

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

303 

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

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

306 info = load_meta(facet, target, metadata_path) 

307 

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

309 log.info('successful verification') 

310 return 0