lean_ctx/core/patterns/
pytest.rs

1/// Dedicated compression pattern for verbose pytest output (`pytest -v`, `pytest --tb=short`).
2///
3/// Handles:
4/// - Per-test PASSED/FAILED lines with full module paths → consolidated summary
5/// - Fixture setup/teardown lines → stripped
6/// - Collection lines (`collecting...`, `collected N items`) → stripped
7/// - Short tracebacks for failures → kept but trimmed
8pub fn compress(command: &str, output: &str) -> Option<String> {
9    // Only activate for pytest commands or output that looks like verbose pytest
10    let is_pytest_cmd = command.contains("pytest") || command.contains("py.test");
11    let has_verbose_markers =
12        (output.contains("::") && output.contains(" PASSED")) || output.contains(" FAILED");
13    let has_session = output.contains("test session starts");
14
15    if !is_pytest_cmd && !has_verbose_markers && !has_session {
16        return None;
17    }
18
19    let mut passed: Vec<String> = Vec::new();
20    let mut failed: Vec<String> = Vec::new();
21    let mut skipped = 0u32;
22    let mut errors = 0u32;
23    let mut xfailed = 0u32;
24    let mut xpassed = 0u32;
25    let mut warnings = 0u32;
26    let mut duration = String::new();
27    let mut failure_details: Vec<String> = Vec::new();
28    let mut in_failure_block = false;
29    let mut current_failure: Vec<String> = Vec::new();
30
31    for line in output.lines() {
32        let trimmed = line.trim();
33
34        // Skip empty lines
35        if trimmed.is_empty() {
36            if in_failure_block && !current_failure.is_empty() {
37                current_failure.push(String::new());
38            }
39            continue;
40        }
41
42        // Skip fixture setup/teardown lines
43        if trimmed.starts_with("SETUP")
44            || trimmed.starts_with("TEARDOWN")
45            || trimmed.contains("--- fixtures ---")
46            || trimmed.starts_with("---------- fixtures")
47        {
48            continue;
49        }
50
51        // Skip collection lines
52        if trimmed.starts_with("collecting ")
53            || trimmed.starts_with("collected ")
54            || trimmed.starts_with("<Module ")
55            || trimmed.starts_with("<Class ")
56            || trimmed.starts_with("<Function ")
57            || trimmed.starts_with("platform ")
58            || trimmed.starts_with("rootdir:")
59            || trimmed.starts_with("configfile:")
60            || trimmed.starts_with("plugins:")
61            || trimmed.starts_with("cachedir:")
62        {
63            continue;
64        }
65
66        // Skip session header
67        if trimmed.contains("test session starts")
68            || (trimmed.starts_with('=')
69                && trimmed.ends_with('=')
70                && trimmed.len() > 3
71                && !trimmed.contains("passed")
72                && !trimmed.contains("failed")
73                && !trimmed.contains("error"))
74        {
75            continue;
76        }
77
78        // Detect verbose per-test result lines: `path/test_file.py::test_name PASSED [ 75%]`
79        // The status may be followed by whitespace and a percentage indicator.
80        if trimmed.contains("::") {
81            match extract_status(trimmed) {
82                Some("PASSED") => {
83                    let name = extract_test_name(trimmed);
84                    passed.push(name);
85                    in_failure_block = false;
86                    continue;
87                }
88                Some("FAILED") => {
89                    let name = extract_test_name(trimmed);
90                    failed.push(name);
91                    in_failure_block = false;
92                    continue;
93                }
94                Some("SKIPPED") => {
95                    skipped += 1;
96                    in_failure_block = false;
97                    continue;
98                }
99                Some("XFAIL") => {
100                    xfailed += 1;
101                    in_failure_block = false;
102                    continue;
103                }
104                Some("XPASS") => {
105                    xpassed += 1;
106                    in_failure_block = false;
107                    continue;
108                }
109                Some("ERROR") => {
110                    errors += 1;
111                    in_failure_block = false;
112                    continue;
113                }
114                _ => {}
115            }
116        }
117
118        // Detect failure section header: `___ test_name ___` or `FAILED test_name`
119        if (trimmed.starts_with("___") && trimmed.ends_with("___"))
120            || trimmed.starts_with("FAILED ")
121        {
122            // Save previous failure block
123            if !current_failure.is_empty() {
124                let detail = current_failure.join("\n");
125                if !detail.trim().is_empty() {
126                    failure_details.push(detail);
127                }
128                current_failure.clear();
129            }
130            in_failure_block = true;
131            continue;
132        }
133
134        // Capture failure traceback lines (keep short, max 5 lines per failure)
135        if in_failure_block {
136            if current_failure.len() < 5 {
137                current_failure.push(trimmed.to_string());
138            }
139            continue;
140        }
141
142        // Parse summary line: `=== 42 passed, 1 failed in 3.21s ===`
143        if (trimmed.starts_with('=') || trimmed.starts_with('-'))
144            && (trimmed.contains("passed")
145                || trimmed.contains("failed")
146                || trimmed.contains("error"))
147        {
148            if let Some(d) = extract_duration(trimmed) {
149                duration = d;
150            }
151            // Also extract counters from summary as fallback
152            if let Some(n) = extract_counter(trimmed, " passed")
153                && passed.is_empty()
154                && n > 0
155            {
156                // Use counter from summary if we didn't see individual lines
157                for _ in 0..n {
158                    passed.push(String::new());
159                }
160            }
161            if let Some(n) = extract_counter(trimmed, " failed")
162                && failed.is_empty()
163                && n > 0
164            {
165                for _ in 0..n {
166                    failed.push(String::new());
167                }
168            }
169            if let Some(n) = extract_counter(trimmed, " skipped")
170                && skipped == 0
171            {
172                skipped = n;
173            }
174            if let Some(n) = extract_counter(trimmed, " xfailed")
175                && xfailed == 0
176            {
177                xfailed = n;
178            }
179            if let Some(n) = extract_counter(trimmed, " xpassed")
180                && xpassed == 0
181            {
182                xpassed = n;
183            }
184            if let Some(n) = extract_counter(trimmed, " warning") {
185                warnings = n;
186            }
187            if let Some(n) = extract_counter(trimmed, " error")
188                && errors == 0
189            {
190                errors = n;
191            }
192        }
193    }
194
195    // Save last failure block
196    if !current_failure.is_empty() {
197        let detail = current_failure.join("\n");
198        if !detail.trim().is_empty() {
199            failure_details.push(detail);
200        }
201    }
202
203    let passed_count = passed.len() as u32;
204    let failed_count = failed.len() as u32;
205
206    if passed_count == 0 && failed_count == 0 && errors == 0 {
207        return None;
208    }
209
210    // Build compressed output
211    let mut result = String::from("pytest: ");
212
213    if failed_count == 0 && errors == 0 {
214        result.push_str(&format!("✓ {passed_count} passed"));
215    } else {
216        result.push_str(&format!("{passed_count} passed, {failed_count} failed"));
217    }
218
219    if skipped > 0 {
220        result.push_str(&format!(", {skipped} skipped"));
221    }
222    if xfailed > 0 {
223        result.push_str(&format!(", {xfailed} xfailed"));
224    }
225    if xpassed > 0 {
226        result.push_str(&format!(", {xpassed} xpassed"));
227    }
228    if errors > 0 {
229        result.push_str(&format!(", {errors} errors"));
230    }
231    if warnings > 0 {
232        result.push_str(&format!(", {warnings} warnings"));
233    }
234
235    if !duration.is_empty() {
236        result.push_str(&format!(" in {duration}"));
237    }
238
239    // Show passed test names when count is small (preserves identifiers for debugging)
240    let named_passed: Vec<&String> = passed.iter().filter(|s| !s.is_empty()).collect();
241    if !named_passed.is_empty() && named_passed.len() <= 10 {
242        let names: Vec<&str> = named_passed.iter().map(|s| s.as_str()).collect();
243        result.push_str(&format!("\n  ran: {}", names.join(", ")));
244    }
245
246    // Show failed test names (up to 5)
247    let named_failures: Vec<&String> = failed.iter().filter(|s| !s.is_empty()).collect();
248    if !named_failures.is_empty() {
249        for f in named_failures.iter().take(5) {
250            result.push_str(&format!("\n  FAIL: {f}"));
251        }
252        if named_failures.len() > 5 {
253            result.push_str(&format!("\n  ...+{} more", named_failures.len() - 5));
254        }
255    }
256
257    // Show failure details (up to 3 blocks, trimmed)
258    if !failure_details.is_empty() {
259        for detail in failure_details.iter().take(3) {
260            let short: String = detail.lines().take(3).collect::<Vec<_>>().join("\n");
261            result.push_str(&format!("\n  > {short}"));
262        }
263    }
264
265    Some(result)
266}
267
268/// Extracts the test status from a verbose pytest line.
269/// Handles lines like: `tests/test_auth.py::test_name PASSED                  [ 75%]`
270/// Returns the status keyword if found.
271fn extract_status(line: &str) -> Option<&'static str> {
272    const STATUSES: &[&str] = &["PASSED", "FAILED", "SKIPPED", "XFAIL", "XPASS", "ERROR"];
273    // Strip trailing percentage indicator and whitespace
274    let stripped = if let Some(bracket_pos) = line.rfind('[') {
275        if line[bracket_pos..].contains('%') {
276            line[..bracket_pos].trim()
277        } else {
278            line.trim()
279        }
280    } else {
281        line.trim()
282    };
283
284    STATUSES.iter().find(|&&s| stripped.ends_with(s)).copied()
285}
286
287/// Extracts the short test name from a verbose pytest line.
288/// Input: `tests/test_auth.py::TestLogin::test_expired_token PASSED                  [ 75%]`
289/// Output: `test_auth.py::test_expired_token`
290fn extract_test_name(line: &str) -> String {
291    let trimmed = line.trim();
292
293    // Strip trailing percentage indicator `[ 75%]`
294    let without_pct = if let Some(bracket_pos) = trimmed.rfind('[') {
295        if trimmed[bracket_pos..].contains('%') {
296            trimmed[..bracket_pos].trim()
297        } else {
298            trimmed
299        }
300    } else {
301        trimmed
302    };
303
304    // Remove the status suffix (PASSED, FAILED, etc.)
305    let name_part = without_pct
306        .rsplit_once(' ')
307        .map_or(without_pct, |(name, _status)| name.trim());
308
309    // Shorten: keep filename::test_name, drop intermediate path
310    if let Some(last_slash) = name_part.rfind('/') {
311        name_part[last_slash + 1..].to_string()
312    } else {
313        name_part.to_string()
314    }
315}
316
317fn extract_duration(line: &str) -> Option<String> {
318    // Look for "in X.XXs" pattern
319    if let Some(pos) = line.find(" in ") {
320        let after = &line[pos + 4..];
321        let dur: String = after
322            .chars()
323            .take_while(|c| c.is_ascii_digit() || *c == '.' || *c == 's' || *c == 'm')
324            .collect();
325        let dur = dur.trim_end_matches('=').trim().to_string();
326        if !dur.is_empty() {
327            return Some(dur);
328        }
329    }
330    None
331}
332
333fn extract_counter(line: &str, keyword: &str) -> Option<u32> {
334    let pos = line.find(keyword)?;
335    let before = &line[..pos];
336    let num_str = before.split_whitespace().last()?;
337    let clean: String = num_str.chars().filter(char::is_ascii_digit).collect();
338    clean.parse::<u32>().ok()
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344
345    #[test]
346    fn verbose_all_passed() {
347        let output = "\
348============================= test session starts ==============================
349platform linux -- Python 3.11.5, pytest-7.4.3, pluggy-1.3.0
350rootdir: /home/user/project
351configfile: pyproject.toml
352plugins: cov-4.1.0
353collecting ... collected 3 items
354
355tests/test_math.py::test_add PASSED                                      [ 33%]
356tests/test_math.py::test_subtract PASSED                                 [ 66%]
357tests/test_math.py::test_multiply PASSED                                 [100%]
358
359============================== 3 passed in 0.42s ===============================";
360
361        let result = compress("pytest -v", output).expect("should compress");
362        assert!(result.contains("✓ 3 passed"));
363        assert!(result.contains("0.42s"));
364        assert!(!result.contains("rootdir"));
365        assert!(!result.contains("collecting"));
366        assert!(!result.contains("platform"));
367    }
368
369    #[test]
370    fn verbose_mixed_results() {
371        let output = "\
372============================= test session starts ==============================
373platform linux -- Python 3.11.5, pytest-7.4.3
374collected 4 items
375
376tests/test_auth.py::test_login PASSED                                    [ 25%]
377tests/test_auth.py::test_logout PASSED                                   [ 50%]
378tests/test_auth.py::test_expired_token FAILED                            [ 75%]
379tests/test_auth.py::test_refresh SKIPPED                                 [100%]
380
381=========================== short test summary info ============================
382FAILED tests/test_auth.py::test_expired_token
383============================== 1 failed, 2 passed, 1 skipped in 1.23s ===============================";
384
385        let result = compress("pytest -v", output).expect("should compress");
386        assert!(result.contains("2 passed"));
387        assert!(result.contains("1 failed"));
388        assert!(result.contains("1 skipped"));
389        assert!(result.contains("FAIL:"));
390        assert!(result.contains("test_expired_token"));
391    }
392
393    #[test]
394    fn strips_fixture_lines() {
395        let output = "\
396============================= test session starts ==============================
397collected 2 items
398
399SETUP    S session_fixture
400tests/test_db.py::test_insert PASSED                                     [ 50%]
401TEARDOWN S session_fixture
402tests/test_db.py::test_query PASSED                                      [100%]
403
404============================== 2 passed in 0.31s ===============================";
405
406        let result = compress("pytest -v --setup-show", output).expect("should compress");
407        assert!(result.contains("✓ 2 passed"));
408        assert!(!result.contains("SETUP"));
409        assert!(!result.contains("TEARDOWN"));
410    }
411
412    #[test]
413    fn strips_collection_lines() {
414        let output = "\
415============================= test session starts ==============================
416platform linux -- Python 3.11.5
417collecting ... collected 5 items
418<Module tests/test_api.py>
419  <Class TestUsers>
420    <Function test_list>
421    <Function test_create>
422
423tests/test_api.py::TestUsers::test_list PASSED                           [ 20%]
424tests/test_api.py::TestUsers::test_create PASSED                         [ 40%]
425tests/test_api.py::TestUsers::test_delete PASSED                         [ 60%]
426tests/test_api.py::TestUsers::test_update PASSED                         [ 80%]
427tests/test_api.py::TestUsers::test_get PASSED                            [100%]
428
429============================== 5 passed in 2.10s ===============================";
430
431        let result = compress("pytest -v --collect-only", output).expect("should compress");
432        assert!(result.contains("✓ 5 passed"));
433        assert!(!result.contains("<Module"));
434        assert!(!result.contains("<Class"));
435        assert!(!result.contains("<Function"));
436        assert!(!result.contains("collecting"));
437    }
438
439    #[test]
440    fn non_pytest_returns_none() {
441        let output = "Hello world\nThis is not pytest output\n";
442        assert!(compress("echo hello", output).is_none());
443    }
444
445    #[test]
446    fn failure_with_traceback() {
447        let output = "\
448============================= test session starts ==============================
449collected 2 items
450
451tests/test_calc.py::test_divide PASSED                                   [ 50%]
452tests/test_calc.py::test_divide_zero FAILED                              [100%]
453
454=================================== FAILURES ===================================
455___________________________ test_divide_zero ___________________________________
456
457    def test_divide_zero():
458>       assert divide(1, 0) == 0
459E       ZeroDivisionError: division by zero
460
461src/calc.py:10: ZeroDivisionError
462=========================== short test summary info ============================
463FAILED tests/test_calc.py::test_divide_zero
464============================== 1 failed, 1 passed in 0.15s ===============================";
465
466        let result = compress("pytest -v --tb=short", output).expect("should compress");
467        assert!(result.contains("1 passed"));
468        assert!(result.contains("1 failed"));
469        assert!(result.contains("FAIL:"));
470        assert!(result.contains("test_divide_zero"));
471    }
472}
lean_ctx/core/patterns/pytest.rs

lean_ctx/core/patterns/
pytest.rs
