lean_ctx_patterns/
pytest.rs

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

lean_ctx_patterns/
pytest.rs
