Coverage for liitos / gather.py: 84.33%

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

2 

3import os 

4import pathlib 

5 

6import yaml 

7 

8from liitos import ( 

9 DEFAULT_STRUCTURE_NAME, 

10 KEY_APPROVALS, 

11 KEY_BIND, 

12 KEY_CHANGES, 

13 KEY_LAYOUT, 

14 KEY_META, 

15 KEYS_REQUIRED, 

16 ENCODING, 

17 OptionsType, 

18 PathLike, 

19 log, 

20) 

21 

22Approvals = dict[str, list[str] | list[list[str]]] 

23Assets = dict[str, dict[str, dict[str, str]]] 

24Binder = list[str] 

25Changes = dict[str, list[str] | list[list[str]]] 

26Layout = dict[str, str] 

27Meta = dict[str, str] 

28Structure = dict[str, list[dict[str, str]]] 

29Targets = set[str] 

30Facets = dict[str, Targets] 

31Payload = Approvals | Binder | Changes | Meta 

32Verification = tuple[bool, str] 

33 

34 

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

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

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

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

39 

40 

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

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

43 return set(target for target in structure) 

44 

45 

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

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

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

49 

50 

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

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

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

54 

55 

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

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

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

59 

60 

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

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

63 if name in facet_map[target]: 

64 return True, '' 

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

66 

67 

68def error_context( 

69 payload: Payload, 

70 label: str, 

71 facet: str, 

72 target: str, 

73 path: PathLike, 

74 err: FileNotFoundError | KeyError | ValueError, 

75) -> tuple[Payload, str]: 

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

77 if isinstance(err, FileNotFoundError): 

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

79 if isinstance(err, KeyError): 

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

81 if isinstance(err, ValueError): 

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

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

84 

85 

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

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

88 try: 

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

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

91 except FileNotFoundError as err: 

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

93 

94 

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

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

97 try: 

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

99 except KeyError as err: 

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

101 return load_binder(facet, target, path) 

102 

103 

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

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

106 try: 

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

108 return yaml.safe_load(handle), '' 

109 except FileNotFoundError as err: 

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

111 

112 

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

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

115 try: 

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

117 except KeyError as err: 

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

119 return load_layout(facet, target, path) 

120 

121 

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

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

124 try: 

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

126 return yaml.safe_load(handle), '' 

127 except FileNotFoundError as err: 

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

129 

130 

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

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

133 try: 

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

135 except KeyError as err: 

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

137 return load_meta(facet, target, path) 

138 

139 

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

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

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

143 return error_context( 

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

145 ) # type: ignore 

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

147 try: 

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

149 return yaml.safe_load(handle), '' 

150 except FileNotFoundError as err: 

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

152 

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

154 

155 

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

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

158 try: 

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

160 except KeyError as err: 

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

162 return load_approvals(facet, target, path) 

163 

164 

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

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

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

168 return error_context( 

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

170 ) # type: ignore 

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

172 try: 

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

174 return yaml.safe_load(handle), '' 

175 except FileNotFoundError as err: 

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

177 

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

179 

180 

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

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

183 try: 

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

185 except KeyError as err: 

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

187 return load_changes(facet, target, path) 

188 

189 

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

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

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

193 return True, '' 

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

195 

196 

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

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

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

200 if not predicate: 

201 return predicate, message 

202 for key in KEYS_REQUIRED: 

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

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

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

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

207 return True, '' 

208 

209 

210ASSET_KEY_ACTION = { 

211 KEY_APPROVALS: approvals, 

212 KEY_BIND: binder, 

213 KEY_CHANGES: changes, 

214 KEY_LAYOUT: layout, 

215 KEY_META: meta, 

216} 

217 

218 

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

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

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

222 if not predicate: 

223 return predicate, message 

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

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

226 if not asset: 

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

228 return True, '' 

229 

230 

231def prelude( 

232 doc_root: PathLike, structure_name: str, target_key: str, facet_key: str, command: str 

233) -> tuple[Structure, Assets]: 

234 """DRY.""" 

235 doc_root = pathlib.Path(doc_root) 

236 idem = os.getcwd() 

237 os.chdir(doc_root) 

238 job_description = ( 

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

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

241 ) 

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

243 structure = load_structure(structure_name) 

244 asset_map = assets(structure) 

245 return structure, asset_map 

246 

247 

248def verify( 

249 doc_root: PathLike, 

250 structure_name: str, 

251 target_key: str, 

252 facet_key: str, 

253 options: OptionsType, 

254) -> int: 

255 """Drive the verification.""" 

256 doc_root = pathlib.Path(doc_root) 

257 os.chdir(doc_root) 

258 facet = facet_key 

259 target = target_key 

260 structure_name = structure_name 

261 job_description = ( 

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

263 ) 

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

265 structure = load_structure(structure_name) 

266 target_set = targets(structure) 

267 facet_map = facets(structure) 

268 asset_map = assets(structure) 

269 

270 predicate, message = verify_target(target, target_set) 

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

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

273 return 1 

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

275 

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

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

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

279 return 1 

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

281 

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

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

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

285 return 1 

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

287 

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

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

290 signatures = load_approvals(facet, target, signatures_path) 

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

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

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

294 history = load_changes(facet, target, history_path) 

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

296 

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

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

299 design = load_layout(facet, target, layout_path) 

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

301 

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

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

304 info = load_meta(facet, target, metadata_path) 

305 

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

307 log.info('successful verification') 

308 return 0