Coverage for sygaldry / checker.py: 66%

1from __future__ import annotations 

2 

3__author__ = "Rohan B. Dalton" 

4 

5import json 

6import os 

7import re 

8import shutil 

9import subprocess 

10import tempfile 

11from dataclasses import dataclass 

12from pathlib import Path 

13from typing import Any, Optional 

14 

15from .codegen import SourceMapping, generate_check_source 

16from .loader import load_config 

17 

18_MYPY_PATTERN = re.compile( 

19 r"^.+?:(?P<line>\d+):\d+:\s*(?P<severity>error|warning|note):\s*(?P<message>.+?)(?:\s+\[.+\])?$" 

20) 

21_TY_LINE_PATTERN = re.compile(r"^\s*-->\s*.+?:(?P<line>\d+):\d+") 

22_TY_DIAG_PATTERN = re.compile(r"^(?P<severity>error|warning)\[.+?\]:\s*(?P<message>.+)") 

23SUPPORTED_CHECKERS = ("ty", "basedpyright", "pyright", "mypy") 

24 

25 

26@dataclass(frozen=True) 

27class CheckError: 

28 """ 

29 A type error found in a config file. 

30 """ 

31 

32 config_path: str 

33 message: str 

34 severity: str = "error" 

35 

36 

37def check( 

38 path: str | Path | None = None, 

39 *, 

40 type_checker: str | None = None, 

41 config: dict[str, Any] | None = None, 

42) -> list[CheckError]: 

43 """ 

44 Type check a sygaldry config file. 

45 

46 Either *path* or *config* must be provided. When *config* is given 

47 the config dict is used directly (useful from the CLI after loading 

48 and applying overrides). When *path* is given, the file is loaded 

49 via :func:`load_config`. 

50 

51 :param path: Path to a YAML/TOML config file. 

52 :param type_checker: Which checker to run (``ty``, ``basedpyright``, 

53 ``pyright``, or ``mypy``). Auto-detected when *None*. 

54 :param config: Pre-loaded config dict. 

55 :returns: List of type errors found. 

56 """ 

57 if config is None: 

58 if path is None: 

59 raise ValueError("Either path or config must be provided.") 

60 else: 

61 config = load_config(Path(path)) 

62 

63 source, mappings = generate_check_source(config) 

64 

65 checker = type_checker or _detect_type_checker() 

66 return _run_and_parse(checker, source, mappings) 

67 

68 

69def _detect_type_checker() -> str: 

70 """ 

71 Auto-detect an available type checker. 

72 """ 

73 for name in SUPPORTED_CHECKERS: 73 ↛ 77line 73 didn't jump to line 77 because the loop on line 73 didn't complete

74 if shutil.which(name): 74 ↛ 73line 74 didn't jump to line 73 because the condition on line 74 was always true

75 return name 

76 else: 

77 raise RuntimeError( 

78 "No supported type checker found. Install ty, basedpyright, pyright, or mypy." 

79 ) 

80 

81 

82def _run_and_parse( 

83 checker: str, 

84 source: str, 

85 mappings: list[SourceMapping], 

86) -> list[CheckError]: 

87 """ 

88 Write source to a temp file, run the checker, and parse results. 

89 """ 

90 fd, tmp_path = tempfile.mkstemp(suffix=".py", prefix="_sygaldry_check_") 

91 with os.fdopen(fd, "w") as fh: 

92 fh.write(source) 

93 result = _invoke_checker(checker, tmp_path) 

94 os.unlink(tmp_path) 

95 return _parse_output(checker, result, mappings) 

96 

97 

98def _invoke_checker(checker: str, filepath: str) -> subprocess.CompletedProcess: 

99 """ 

100 Run the type checker subprocess. 

101 """ 

102 if checker in {"pyright", "basedpyright"}: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 cmd = [checker, "--outputjson", filepath] 

104 elif checker == "mypy": 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true

105 cmd = ["mypy", "--no-color-output", "--show-column-numbers", filepath] 

106 elif checker == "ty": 106 ↛ 109line 106 didn't jump to line 109 because the condition on line 106 was always true

107 cmd = ["ty", "check", filepath] 

108 else: 

109 raise ValueError(f"Unsupported type checker: {checker}") 

110 

111 return subprocess.run( 

112 cmd, 

113 capture_output=True, 

114 text=True, 

115 timeout=120, 

116 ) 

117 

118 

119def _parse_output( 

120 checker: str, 

121 result: subprocess.CompletedProcess, 

122 mappings: list[SourceMapping], 

123) -> list[CheckError]: 

124 """ 

125 Parse type checker output into CheckError instances. 

126 """ 

127 if checker in ("pyright", "basedpyright"): 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true

128 return _parse_pyright(result, mappings) 

129 elif checker == "mypy": 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 return _parse_mypy(result, mappings) 

131 elif checker == "ty": 131 ↛ 133line 131 didn't jump to line 133 because the condition on line 131 was always true

132 return _parse_ty(result, mappings) 

133 return list() 

134 

135 

136def _line_to_config_path(line: int, mappings: list[SourceMapping]) -> str: 

137 """ 

138 Map a generated line number to its config path. 

139 """ 

140 best: SourceMapping | None = None 

141 for mapping in mappings: 

142 if best is None or best.line < mapping.line <= line: 142 ↛ 141line 142 didn't jump to line 141 because the condition on line 142 was always true

143 best = mapping 

144 return best.config_path if best else "<unknown>" 

145 

146 

147def _parse_pyright( 

148 result: subprocess.CompletedProcess, 

149 mappings: list[SourceMapping], 

150) -> list[CheckError]: 

151 errors: list[CheckError] = list() 

152 try: 

153 data = json.loads(result.stdout) 

154 except (json.JSONDecodeError, ValueError): 

155 return errors 

156 

157 for diag in data.get("generalDiagnostics", list()): 

158 line = diag.get("range", {}).get("start", {}).get("line", 0) 

159 # pyright uses 0-based lines 

160 config_path = _line_to_config_path(line + 1, mappings) 

161 severity = diag.get("severity", "error") 

162 message = diag.get("message", "") 

163 errors.append( 

164 CheckError( 

165 config_path=config_path, 

166 message=message, 

167 severity=severity, 

168 ) 

169 ) 

170 return errors 

171 

172 

173def _parse_mypy( 

174 result: subprocess.CompletedProcess, 

175 mappings: list[SourceMapping], 

176) -> list[CheckError]: 

177 errors: list[CheckError] = list() 

178 for raw_line in result.stdout.splitlines(): 

179 if match := _MYPY_PATTERN.match(raw_line): 

180 line_no = int(match.group("line")) 

181 severity = match.group("severity") 

182 message = match.group("message") 

183 if severity == "note": 

184 continue 

185 config_path = _line_to_config_path(line_no, mappings) 

186 errors.append( 

187 CheckError( 

188 config_path=config_path, 

189 message=message, 

190 severity=severity, 

191 ) 

192 ) 

193 return errors 

194 

195 

196def _parse_ty( 

197 result: subprocess.CompletedProcess, 

198 mappings: list[SourceMapping], 

199) -> list[CheckError]: 

200 errors: list[CheckError] = list() 

201 output = result.stdout + result.stderr 

202 lines = output.splitlines() 

203 

204 for idx, line in enumerate(lines): 

205 if diag_match := _TY_DIAG_PATTERN.match(line): 

206 severity = diag_match.group("severity") 

207 message = diag_match.group("message") 

208 line_no = _extract_ty_line(lines, idx) 

209 config_path = _line_to_config_path(line_no, mappings) if line_no else "<unknown>" 

210 errors.append( 

211 CheckError( 

212 config_path=config_path, 

213 message=message, 

214 severity=severity, 

215 ) 

216 ) 

217 

218 return errors 

219 

220 

221def _extract_ty_line(lines: list[str], diag_idx: int) -> int | None: 

222 """ 

223 Look ahead from a ty diagnostic line to find the --> line number. 

224 """ 

225 for idx in range(diag_idx + 1, min(diag_idx + 6, len(lines))): 225 ↛ 228line 225 didn't jump to line 228 because the loop on line 225 didn't complete

226 if match := _TY_LINE_PATTERN.match(lines[idx]): 226 ↛ 225line 226 didn't jump to line 225 because the condition on line 226 was always true

227 return int(match.group("line")) 

228 return None