Coverage for src/rhiza_hooks/check_workflow_names.py: 100%

1#!/usr/bin/env python3 

2"""Script to ensure GitHub Actions workflows have the (RHIZA) prefix. 

3 

4This hook checks that all rhiza workflow files have their 'name' field 

5properly formatted with the (RHIZA) prefix in uppercase. If not, it 

6automatically updates the file. 

7 

8Migrated from: https://github.com/Jebel-Quant/rhiza/.rhiza/scripts/check_workflow_names.py 

9""" 

10 

11from __future__ import annotations 

12 

13import sys 

14 

15import yaml 

16 

17 

18def _expected_name(name: str) -> str: 

19 """Return the canonical ``(RHIZA) <UPPERCASE>`` form of a workflow name.""" 

20 prefix = "(RHIZA) " 

21 # Remove prefix if present to verify the rest of the string 

22 clean_name = name[len(prefix) :] if name.startswith(prefix) else name 

23 # Collapse any internal/trailing whitespace (e.g. from a folded/block YAML 

24 # scalar, where PyYAML yields newlines) so the rewrite is a single line. 

25 clean_name = " ".join(clean_name.split()) 

26 return f"{prefix}{clean_name.upper()}" 

27 

28 

29def _is_block_scalar_name(line: str) -> bool: 

30 """True if a top-level ``name:`` line opens a block scalar (``|`` / ``>``). 

31 

32 Detection keys on the leading indicator character (``[:1]``) so chomping 

33 variants like ``>-`` / ``|-`` are recognised too. 

34 """ 

35 return line[len("name:") :].strip()[:1] in ("|", ">") 

36 

37 

38def _is_block_continuation(line: str) -> bool: 

39 """True if ``line`` continues a block scalar's value. 

40 

41 YAML indentation is spaces, so a scalar's continuation lines are blank or 

42 space-indented; the first flush-left line ends the scalar. 

43 """ 

44 return line.strip() == "" or line.startswith(" ") 

45 

46 

47def _count_block_continuations(following: list[str]) -> int: 

48 """Count the leading block-scalar continuation lines in ``following``.""" 

49 count = 0 

50 for line in following: 

51 if not _is_block_continuation(line): 

52 break 

53 count += 1 

54 return count 

55 

56 

57def _replace_name_lines(lines: list[str], expected_name: str) -> list[str]: 

58 """Return ``lines`` with the first top-level ``name:`` set to ``expected_name``. 

59 

60 Only the top-level workflow ``name`` is rewritten: it is the sole ``name:`` 

61 key at column 0, so job- and step-level ``name:`` keys (always indented) 

62 never match, and only the first match is replaced. If the value is a block 

63 scalar (``name: >`` / ``name: |``), its continuation lines are dropped so no 

64 orphan scalar text is left behind. When no top-level ``name:`` line exists, 

65 the lines are returned unchanged. 

66 """ 

67 for idx, line in enumerate(lines): 

68 if line.startswith("name:"): 

69 tail = idx + 1 

70 if _is_block_scalar_name(line): 

71 tail += _count_block_continuations(lines[idx + 1 :]) 

72 return [*lines[:idx], f'name: "{expected_name}"\n', *lines[tail:]] 

73 return list(lines) 

74 

75 

76def _rewrite_workflow_name(filepath: str, expected_name: str) -> None: 

77 """Rewrite the top-level ``name:`` of a workflow file, preserving comments.""" 

78 with open(filepath) as f_read: 

79 lines = f_read.readlines() 

80 with open(filepath, "w") as f_write: 

81 f_write.writelines(_replace_name_lines(lines, expected_name)) 

82 

83 

84def check_file(filepath: str) -> bool: 

85 """Check if the workflow file has the correct name prefix and update if needed. 

86 

87 Args: 

88 filepath: Path to the workflow file. 

89 

90 Returns: 

91 bool: True if file is correct, False if it was updated or has errors. 

92 """ 

93 with open(filepath) as f: 

94 try: 

95 content = yaml.safe_load(f) 

96 except yaml.YAMLError as exc: 

97 print(f"Error parsing YAML {filepath}: {exc}") 

98 return False 

99 

100 if not isinstance(content, dict): 

101 # Empty file or not a dict 

102 return True 

103 

104 name = content.get("name") 

105 if not name: 

106 print(f"Error: {filepath} missing 'name' field.") 

107 return False 

108 

109 expected_name = _expected_name(name) 

110 

111 if name == expected_name: 

112 return True 

113 

114 print(f"Updating {filepath}: name '{name}' -> '{expected_name}'") 

115 _rewrite_workflow_name(filepath, expected_name) 

116 return False # Fail so pre-commit knows files were modified 

117 

118 

119def main(argv: list[str] | None = None) -> int: 

120 """Execute the script.""" 

121 files = argv if argv is not None else sys.argv[1:] 

122 failed = False # pragma: no mutate # equivalent: only ever read via `if failed` 

123 for f in files: 

124 if not check_file(f): 

125 failed = True 

126 

127 if failed: 

128 sys.exit(1) 

129 return 0 

130 

131 

132def _run() -> None: 

133 """Entry point: delegate to :func:`main` and exit with its return code.""" 

134 sys.exit(main()) 

135 

136 

137if __name__ == "__main__": # pragma: no cover # pragma: no mutate 

138 _run()