Coverage for archdocs/features/kubernetes/manifest_files.py: 100%

1import pathlib 

2import re as py_re 

3import typing 

4 

5from archdocs import settings 

6from archdocs.features.kubernetes import const 

7 

8 

9_API_VERSION_PATTERN: typing.Final = py_re.compile(r"^apiVersion:", flags=settings.TYPICAL_RE_FLAGS) 

10_KIND_PATTERN: typing.Final = py_re.compile(r"^kind:", flags=settings.TYPICAL_RE_FLAGS) 

11 

12 

13def _read_manifest_text(one_file_path: pathlib.Path, /) -> str: 

14 # Mirrors `main._read_source_text`: manifests are hunted through somebody's whole tree and 

15 # a couple of directories above it, so a dangling symlink or a file the process may not open 

16 # costs that one file — not the whole page. 

17 try: 

18 return one_file_path.read_text(errors="ignore") 

19 except OSError: 

20 return "" 

21 

22 

23def _is_manifest_file(one_file_path: pathlib.Path, /) -> bool: 

24 if one_file_path.stem == const.VALUES_FILE_STEM: 

25 return True 

26 file_source: typing.Final = _read_manifest_text(one_file_path) 

27 return bool(_API_VERSION_PATTERN.search(file_source) and _KIND_PATTERN.search(file_source)) 

28 

29 

30def _find_manifest_files(search_dir: pathlib.Path, /) -> list[pathlib.Path]: 

31 all_found_files: typing.Final = ( 

32 one_found_file 

33 for one_file_suffix in const.MANIFEST_FILE_SUFFIXES 

34 for one_found_file in search_dir.rglob(f"*{one_file_suffix}") 

35 if settings.SKIPPED_DIR_NAMES.isdisjoint(one_found_file.relative_to(search_dir).parts) 

36 ) 

37 return sorted( 

38 (one_found_file for one_found_file in all_found_files if _is_manifest_file(one_found_file)), 

39 key=lambda one_found_file: (len(one_found_file.parts), one_found_file), 

40 ) 

41 

42 

43def _iter_search_roots(root_path: pathlib.Path, /) -> typing.Iterator[pathlib.Path]: 

44 # Charts live next to the sources at least as often as inside them, so the search climbs a 

45 # couple of levels up — but never out of the repository the sources belong to. 

46 for one_search_root in (root_path, *list(root_path.parents)[: const.PARENT_SEARCH_DEPTH]): 

47 yield one_search_root 

48 if (one_search_root / const.REPOSITORY_MARKER_NAME).exists(): 

49 return 

50 

51 

52def _find_manifest_dir(search_dir: pathlib.Path, /) -> pathlib.Path | None: 

53 all_manifest_files: typing.Final = _find_manifest_files(search_dir) 

54 if not all_manifest_files: 

55 return None 

56 closest_manifest_dir: typing.Final = all_manifest_files[0].parent 

57 if closest_manifest_dir.name == const.TEMPLATES_DIR_NAME: 

58 return closest_manifest_dir.parent 

59 return closest_manifest_dir 

60 

61 

62def _resolve_manifest_dir(root_path: pathlib.Path, configured_dir: str | pathlib.Path | None, /) -> pathlib.Path | None: 

63 if configured_dir is None: 

64 return next( 

65 ( 

66 one_found_dir 

67 for one_found_dir in map(_find_manifest_dir, _iter_search_roots(root_path)) 

68 if one_found_dir is not None 

69 ), 

70 None, 

71 ) 

72 all_candidate_dirs: typing.Final = ( 

73 (one_search_root / configured_dir).resolve() for one_search_root in _iter_search_roots(root_path) 

74 ) 

75 return next((one_candidate_dir for one_candidate_dir in all_candidate_dirs if one_candidate_dir.is_dir()), None) 

76 

77 

78def read_kubernetes_manifests(root_path: pathlib.Path, configured_dir: str | pathlib.Path | None, /) -> str: 

79 manifest_dir: typing.Final = _resolve_manifest_dir(root_path, configured_dir) 

80 if manifest_dir is None: 

81 return "" 

82 return "".join( 

83 f"{_read_manifest_text(one_manifest_file)}\n" for one_manifest_file in _find_manifest_files(manifest_dir) 

84 )