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 r"""Return the canonical ``(RHIZA) <UPPERCASE>`` form of a workflow name. 

20 

21 >>> _expected_name("My Workflow") 

22 '(RHIZA) MY WORKFLOW' 

23 

24 Already-canonical input is left alone, which is what makes the hook safe to 

25 run repeatedly: 

26 

27 >>> _expected_name("(RHIZA) MY WORKFLOW") 

28 '(RHIZA) MY WORKFLOW' 

29 

30 A folded or block YAML scalar reaches this function with embedded newlines; 

31 they collapse so the rewritten ``name:`` stays a single line: 

32 

33 >>> _expected_name("build\n and test\n") 

34 '(RHIZA) BUILD AND TEST' 

35 """ 

36 prefix = "(RHIZA) " 

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

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

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

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

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

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

43 

44 

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

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

47 

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

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

50 """ 

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

52 

53 

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

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

56 

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

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

59 """ 

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

61 

62 

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

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

65 count = 0 

66 for line in following: 

67 if not _is_block_continuation(line): 

68 break 

69 count += 1 

70 return count 

71 

72 

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

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

75 

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

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

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

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

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

81 the lines are returned unchanged. 

82 """ 

83 for idx, line in enumerate(lines): 

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

85 tail = idx + 1 

86 if _is_block_scalar_name(line): 

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

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

89 return list(lines) 

90 

91 

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

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

94 

95 ``newline=""`` on the write is load-bearing. The read above is universal-newline, 

96 so ``lines`` always holds ``\n`` regardless of what is on disk; without 

97 ``newline=""`` the write would translate every one of them to ``os.linesep``, and 

98 on Windows a one-line fix would land as a whole-file CRLF diff. Suppressing the 

99 translation makes the hook's output a function of its input alone — identical bytes 

100 on every platform, which is the only sane contract for a hook whose output gets 

101 committed. 

102 """ 

103 with open(filepath, encoding="utf-8") as f_read: 

104 lines = f_read.readlines() 

105 with open(filepath, "w", encoding="utf-8", newline="") as f_write: 

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

107 

108 

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

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

111 

112 Args: 

113 filepath: Path to the workflow file. 

114 

115 Returns: 

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

117 """ 

118 with open(filepath, encoding="utf-8") as f: 

119 try: 

120 content = yaml.safe_load(f) 

121 except yaml.YAMLError as exc: 

122 print(f"Error parsing YAML {filepath}: {exc}", file=sys.stderr) 

123 return False 

124 

125 if not isinstance(content, dict): 

126 # Empty file or not a dict 

127 return True 

128 

129 name = content.get("name") 

130 if not name: 

131 print(f"Error: {filepath} missing 'name' field.", file=sys.stderr) 

132 return False 

133 

134 expected_name = _expected_name(name) 

135 

136 if name == expected_name: 

137 return True 

138 

139 print(f"Updating {filepath}: name '{name}' -> '{expected_name}'", file=sys.stderr) 

140 _rewrite_workflow_name(filepath, expected_name) 

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

142 

143 

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

145 """Execute the script.""" 

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

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

148 for f in files: 

149 if not check_file(f): 

150 failed = True 

151 

152 return 1 if failed else 0 

153 

154 

155def _run() -> None: 

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

157 sys.exit(main()) 

158 

159 

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

161 _run()