Coverage for liitos/gather.py: 86.35%
194 statements
« prev ^ index » next coverage.py v7.6.0, created at 2024-07-16 00:43:25 +00:00
« prev ^ index » next coverage.py v7.6.0, created at 2024-07-16 00:43:25 +00:00
1"""Gather the structure and discover the content."""
3import json
4import os
5import pathlib
6from typing import Dict, List, Set, Tuple, Union
8import yaml
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 log,
20)
22PathLike = Union[str, pathlib.Path]
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]
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
43def targets(structure: Structure) -> Targets:
44 """Extract the targets from the given structure information item."""
45 return set(target for target in structure)
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()}
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
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)}')
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])}'
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})')
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
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)
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
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)
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
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)
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 try:
146 with open(path, 'rt', encoding=ENCODING) as handle:
147 return json.load(handle), ''
148 except FileNotFoundError as err:
149 return error_context({}, 'Approvals', facet, target, path, err) # type: ignore
150 elif str(path).lower().endswith(('yaml', 'yml')): 150 ↛ 157line 150 didn't jump to line 157 because the condition on line 150 was always true
151 try:
152 with open(path, 'rt', encoding=ENCODING) as handle:
153 return yaml.safe_load(handle), ''
154 except FileNotFoundError as err:
155 return error_context({}, 'Approvals', facet, target, path, err) # type: ignore
157 return error_context({}, 'Approvals', facet, target, path, ValueError('json or yaml required')) # type: ignore
160def approvals(facet: str, target: str, asset_struct: Assets) -> Tuple[Approvals, str]:
161 """Yield the approvals for facet of target from link in assets and message (in case of failure)."""
162 try:
163 path = pathlib.Path(asset_struct[target][facet][KEY_APPROVALS])
164 except KeyError as err:
165 return error_context({}, 'Approvals', facet, target, '', err) # type: ignore
166 return load_approvals(facet, target, path)
169def load_changes(facet: str, target: str, path: PathLike) -> Tuple[Approvals, str]:
170 """Yield the changes for facet of target from path and message (in case of failure)."""
171 if str(path).lower().endswith('json'):
172 try:
173 with open(path, 'rt', encoding=ENCODING) as handle:
174 return json.load(handle), ''
175 except FileNotFoundError as err:
176 return error_context({}, 'Changes', facet, target, path, err) # type: ignore
177 elif str(path).lower().endswith(('yaml', 'yml')):
178 try:
179 with open(path, 'rt', encoding=ENCODING) as handle:
180 return yaml.safe_load(handle), ''
181 except FileNotFoundError as err:
182 return error_context({}, 'Changes', facet, target, path, err) # type: ignore
184 return error_context({}, 'Changes', facet, target, path, ValueError('json or yaml required')) # type: ignore
187def changes(facet: str, target: str, asset_struct: Assets) -> Tuple[Changes, str]:
188 """Yield the changes for facet of target from link in assets and message (in case of failure)."""
189 try:
190 path = pathlib.Path(asset_struct[target][facet][KEY_CHANGES])
191 except KeyError as err:
192 return error_context({}, 'Changes', facet, target, '', err) # type: ignore
193 return load_changes(facet, target, path)
196def verify_asset_keys(facet: str, target: str, asset_struct: Assets) -> Verification:
197 """Verify presence of required keys for facet of target yielding predicate and message (in case of failure)."""
198 if all(key in asset_struct[target][facet] for key in KEYS_REQUIRED):
199 return True, ''
200 return False, f'keys in {sorted(KEYS_REQUIRED)} for facet ({facet}) of target ({target}) are missing'
203def verify_asset_links(facet: str, target: str, asset_struct: Assets) -> Verification:
204 """Verify presence of asset links for facet of target yielding predicate and message (in case of failure)."""
205 predicate, message = verify_asset_keys(facet, target, asset_struct)
206 if not predicate:
207 return predicate, message
208 for key in KEYS_REQUIRED:
209 link = pathlib.Path(asset_struct[target][facet][key])
210 log.debug(f' + verifying: {pathlib.Path.cwd() / link}')
211 if not link.is_file() or not link.stat().st_size:
212 return False, f'{key} asset link ({link}) for facet ({facet}) of target ({target}) is invalid'
213 return True, ''
216ASSET_KEY_ACTION = {
217 KEY_APPROVALS: approvals,
218 KEY_BIND: binder,
219 KEY_CHANGES: changes,
220 KEY_LAYOUT: layout,
221 KEY_META: meta,
222}
225def verify_assets(facet: str, target: str, asset_struct: Assets) -> Verification:
226 """Verify assets for facet of target yielding predicate and message (in case of failure)."""
227 predicate, message = verify_asset_links(facet, target, asset_struct)
228 if not predicate:
229 return predicate, message
230 for key, action in ASSET_KEY_ACTION.items(): 230 ↛ 234line 230 didn't jump to line 234 because the loop on line 230 didn't complete
231 asset, message = action(facet, target, asset_struct)
232 if not asset:
233 return False, f'{key} asset for facet ({facet}) of target ({target}) is invalid'
234 return True, ''
237def prelude(
238 doc_root: Union[str, pathlib.Path], structure_name: str, target_key: str, facet_key: str, command: str
239) -> tuple[Structure, Assets]:
240 """DRY."""
241 doc_root = pathlib.Path(doc_root)
242 idem = os.getcwd()
243 os.chdir(doc_root)
244 job_description = (
245 f'facet ({facet_key}) of target ({target_key}) with structure map ({structure_name})'
246 f' in document root ({doc_root}) coming from ({idem})'
247 )
248 log.info(f'executing prelude of command ({command}) for {job_description}')
249 structure = load_structure(structure_name)
250 asset_map = assets(structure)
251 return structure, asset_map
254def verify(
255 doc_root: Union[str, pathlib.Path],
256 structure_name: str,
257 target_key: str,
258 facet_key: str,
259 options: dict[str, Union[bool, str]],
260) -> int:
261 """Drive the verification."""
262 doc_root = pathlib.Path(doc_root)
263 os.chdir(doc_root)
264 facet = facet_key
265 target = target_key
266 structure_name = structure_name
267 job_description = (
268 f'facet ({facet}) for target ({target}) with structure map ({structure_name})' f' in document root ({doc_root})'
269 )
270 log.info(f'starting verification of {job_description}')
271 structure = load_structure(structure_name)
272 target_set = targets(structure)
273 facet_map = facets(structure)
274 asset_map = assets(structure)
276 predicate, message = verify_target(target, target_set)
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'- target ({target}) OK')
282 predicate, message = verify_facet(facet, target, facet_map)
283 if not predicate: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 log.error(f'failed verification with: {message}')
285 return 1
286 log.info(f'- facet ({facet}) of target ({target}) OK')
288 predicate, message = verify_assets(facet, target, asset_map)
289 if not predicate: 289 ↛ 292line 289 didn't jump to line 292 because the condition on line 289 was always true
290 log.error(f'failed verification with: {message}')
291 return 1
292 log.info(f'- assets ({", ".join(sorted(KEYS_REQUIRED))}) for facet ({facet}) of target ({target}) OK')
294 signatures_path = asset_map[target][facet][KEY_APPROVALS]
295 log.info(f'loading signatures from {signatures_path=}')
296 signatures = load_approvals(facet, target, signatures_path)
297 log.info(f'{signatures=}')
298 history_path = asset_map[target][facet][KEY_CHANGES]
299 log.info(f'loading history from {history_path=}')
300 history = load_changes(facet, target, history_path)
301 log.info(f'{history=}')
303 layout_path = asset_map[target][facet][KEY_LAYOUT]
304 log.info(f'loading layout from {layout_path=}')
305 design = load_layout(facet, target, layout_path)
306 log.info(f'{design=}')
308 metadata_path = asset_map[target][facet][KEY_META]
309 log.info(f'loading metadata from {metadata_path=}')
310 info = load_meta(facet, target, metadata_path)
312 log.info(f'{info=}')
313 log.info('successful verification')
314 return 0