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 ↛ 76line 73 didn't jump to line 76 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 raise RuntimeError( 

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

78 ) 

79 

80 

81def _run_and_parse( 

82 checker: str, 

83 source: str, 

84 mappings: list[SourceMapping], 

85) -> list[CheckError]: 

86 """ 

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

88 """ 

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

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

91 fh.write(source) 

92 result = _invoke_checker(checker, tmp_path) 

93 os.unlink(tmp_path) 

94 return _parse_output(checker, result, mappings) 

95 

96 

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

98 """ 

99 Run the type checker subprocess. 

100 """ 

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

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

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

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

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

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

107 else: 

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

109 

110 return subprocess.run( 

111 cmd, 

112 capture_output=True, 

113 text=True, 

114 timeout=120, 

115 ) 

116 

117 

118def _parse_output( 

119 checker: str, 

120 result: subprocess.CompletedProcess, 

121 mappings: list[SourceMapping], 

122) -> list[CheckError]: 

123 """ 

124 Parse type checker output into CheckError instances. 

125 """ 

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

127 return _parse_pyright(result, mappings) 

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

129 return _parse_mypy(result, mappings) 

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

131 return _parse_ty(result, mappings) 

132 return list() 

133 

134 

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

136 """ 

137 Map a generated line number to its config path. 

138 """ 

139 best: SourceMapping | None = None 

140 for mapping in mappings: 

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

142 best = mapping 

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

144 

145 

146def _parse_pyright( 

147 result: subprocess.CompletedProcess, 

148 mappings: list[SourceMapping], 

149) -> list[CheckError]: 

150 errors: list[CheckError] = list() 

151 try: 

152 data = json.loads(result.stdout) 

153 except (json.JSONDecodeError, ValueError): 

154 return errors 

155 

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

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

158 # pyright uses 0-based lines 

159 config_path = _line_to_config_path(line + 1, mappings) 

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

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

162 errors.append( 

163 CheckError( 

164 config_path=config_path, 

165 message=message, 

166 severity=severity, 

167 ) 

168 ) 

169 return errors 

170 

171 

172def _parse_mypy( 

173 result: subprocess.CompletedProcess, 

174 mappings: list[SourceMapping], 

175) -> list[CheckError]: 

176 errors: list[CheckError] = list() 

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

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

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

180 severity = match.group("severity") 

181 message = match.group("message") 

182 if severity == "note": 

183 continue 

184 config_path = _line_to_config_path(line_no, mappings) 

185 errors.append( 

186 CheckError( 

187 config_path=config_path, 

188 message=message, 

189 severity=severity, 

190 ) 

191 ) 

192 return errors 

193 

194 

195def _parse_ty( 

196 result: subprocess.CompletedProcess, 

197 mappings: list[SourceMapping], 

198) -> list[CheckError]: 

199 errors: list[CheckError] = list() 

200 output = result.stdout + result.stderr 

201 lines = output.splitlines() 

202 

203 for idx, line in enumerate(lines): 

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

205 severity = diag_match.group("severity") 

206 message = diag_match.group("message") 

207 line_no = _extract_ty_line(lines, idx) 

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

209 errors.append( 

210 CheckError( 

211 config_path=config_path, 

212 message=message, 

213 severity=severity, 

214 ) 

215 ) 

216 

217 return errors 

218 

219 

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

221 """ 

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

223 """ 

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

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

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

227 return None