lean_ctx_patterns/
mypy.rs

1// SPDX-License-Identifier: Apache-2.0
2
3macro_rules! static_regex {
4    ($pattern:expr_2021) => {{
5        static RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
6        RE.get_or_init(|| {
7            regex::Regex::new($pattern).expect(concat!("BUG: invalid static regex: ", $pattern))
8        })
9    }};
10}
11
12fn error_re() -> &'static regex::Regex {
13    static_regex!(r"^(.+?):(\d+):\s+(error|warning|note):\s+(.+?)(?:\s+\[(.+)\])?$")
14}
15
16fn summary_re() -> &'static regex::Regex {
17    static_regex!(r"Found (\d+) errors? in (\d+) files?")
18}
19
20pub fn compress(_command: &str, output: &str) -> Option<String> {
21    let trimmed = output.trim();
22    if trimmed.is_empty() {
23        return Some("ok".to_string());
24    }
25
26    if trimmed == "Success: no issues found in source files" || trimmed.contains("no issues found")
27    {
28        return Some("clean".to_string());
29    }
30
31    let mut by_code: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
32    let mut by_severity: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
33    let mut files: std::collections::HashSet<String> = std::collections::HashSet::new();
34    let mut first_errors: Vec<String> = Vec::new();
35
36    for line in trimmed.lines() {
37        if let Some(caps) = error_re().captures(line) {
38            let file = caps[1].to_string();
39            let severity = caps[3].to_string();
40            let msg = caps[4].to_string();
41            let code = caps.get(5).map(|m| m.as_str().to_string());
42
43            files.insert(file.clone());
44            *by_severity.entry(severity).or_insert(0) += 1;
45
46            if let Some(ref c) = code {
47                *by_code.entry(c.clone()).or_insert(0) += 1;
48            }
49
50            if first_errors.len() < 5 {
51                let short_file = file.rsplit('/').next().unwrap_or(&file);
52                let line_num = &caps[2];
53                let code_str = code.as_deref().unwrap_or("?");
54                first_errors.push(format!("  {short_file}:{line_num} [{code_str}] {msg}"));
55            }
56        }
57    }
58
59    if let Some(caps) = summary_re().captures(trimmed) {
60        let errors = &caps[1];
61        let file_count = &caps[2];
62
63        let mut parts = vec![format!("{errors} errors in {file_count} files")];
64
65        if !by_code.is_empty() {
66            let mut codes: Vec<(String, u32)> = by_code.into_iter().collect();
67            codes.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
68            for (code, count) in codes.iter().take(6) {
69                parts.push(format!("  [{code}]: {count}"));
70            }
71            if codes.len() > 6 {
72                parts.push(format!("  ... +{} more codes", codes.len() - 6));
73            }
74        }
75
76        if !first_errors.is_empty() {
77            parts.push("Top errors:".to_string());
78            parts.extend(first_errors);
79        }
80
81        return Some(parts.join("\n"));
82    }
83
84    if !files.is_empty() {
85        let total: u32 = by_severity.values().sum();
86        let err_count = by_severity.get("error").copied().unwrap_or(0);
87        let warn_count = by_severity.get("warning").copied().unwrap_or(0);
88
89        let mut parts = vec![format!(
90            "{total} issues in {} files ({err_count} errors, {warn_count} warnings)",
91            files.len()
92        )];
93
94        if !first_errors.is_empty() {
95            parts.extend(first_errors);
96        }
97
98        return Some(parts.join("\n"));
99    }
100
101    let lines: Vec<&str> = trimmed.lines().collect();
102    if lines.len() <= 8 {
103        Some(trimmed.to_string())
104    } else {
105        Some(format!(
106            "{}\n{}",
107            lines[..8].join("\n"),
108            super::elision_marker(lines.len() - 8)
109        ))
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116
117    #[test]
118    fn mypy_clean_output() {
119        let output = "Success: no issues found in source files";
120        assert_eq!(compress("mypy .", output).unwrap(), "clean");
121    }
122
123    #[test]
124    fn mypy_empty_output() {
125        assert_eq!(compress("mypy .", "").unwrap(), "ok");
126    }
127
128    #[test]
129    fn mypy_errors_with_summary() {
130        let output = r#"src/auth.py:42: error: Argument 1 to "validate" has incompatible type "str"; expected "int"  [arg-type]
131src/auth.py:55: error: Missing return statement  [return]
132src/db.py:10: error: Name "cursor" is not defined  [name-defined]
133Found 3 errors in 2 files (checked 15 source files)"#;
134        let result = compress("mypy .", output).unwrap();
135        assert!(result.contains("3 errors in 2 files"));
136        assert!(result.contains("[arg-type]"));
137    }
138
139    #[test]
140    fn mypy_errors_without_summary() {
141        let output = r#"src/main.py:10: error: Incompatible return value type  [return-value]
142src/main.py:20: warning: Unused "type: ignore" comment  [unused-ignore]"#;
143        let result = compress("mypy src/", output).unwrap();
144        assert!(result.contains("2 issues"));
145        assert!(result.contains("1 errors"));
146        assert!(result.contains("1 warnings"));
147    }
148
149    #[test]
150    fn mypy_with_notes() {
151        let output = "src/api.py:5: error: Missing type annotation  [no-untyped-def]\nsrc/api.py:5: note: Use --disallow-untyped-defs\nFound 1 error in 1 file (checked 3 source files)";
152        let result = compress("mypy .", output).unwrap();
153        assert!(result.contains("1 error"));
154    }
155}
lean_ctx_patterns/mypy.rs

lean_ctx_patterns/
mypy.rs
