//! Filters pytest output to show only failures and the summary line.

use crate::core::runner;

use crate::core::utils::{resolved_command, tool_exists, truncate};

use anyhow::Result;

#[derive(Debug, PartialEq)]

enum ParseState {

Header,

TestProgress,

Failures,

Summary,

}

pub fn run(args: &[String], verbose: u8) -> Result<i32> {

let mut cmd = if tool_exists("pytest") {

resolved_command("pytest")

} else {

let mut c = resolved_command("python");

c.arg("-m").arg("pytest");

c

};

let has_tb_flag = args.iter().any(|a| a.starts_with("--tb"));

let has_quiet_flag = args.iter().any(|a| a == "-q" || a == "--quiet");

if !has_tb_flag {

cmd.arg("--tb=short");

}

if !has_quiet_flag {

cmd.arg("-q");

}

for arg in args {

cmd.arg(arg);

}

if verbose > 0 {

eprintln!("Running: pytest --tb=short -q {}", args.join(" "));

}

runner::run_filtered(

cmd,

"pytest",

&args.join(" "),

filter_pytest_output,

runner::RunOptions::stdout_only().tee("pytest"),

)

}

pub(crate) fn filter_pytest_output(output: &str) -> String {

let mut state = ParseState::Header;

let mut test_files: Vec<String> = Vec::new();

let mut failures: Vec<String> = Vec::new();

let mut current_failure: Vec<String> = Vec::new();

let mut summary_line = String::new();

for line in output.lines() {

let trimmed = line.trim();

// State transitions

if trimmed.starts_with("===") && trimmed.contains("test session starts") {

state = ParseState::Header;

continue;

} else if trimmed.starts_with("===") && trimmed.contains("FAILURES") {

state = ParseState::Failures;

continue;

} else if trimmed.starts_with("===") && trimmed.contains("short test summary") {

state = ParseState::Summary;

// Save current failure if any

if !current_failure.is_empty() {

failures.push(current_failure.join("\n"));

current_failure.clear();

}

continue;

} else if trimmed.starts_with("===")

&& (trimmed.contains("passed")

|| trimmed.contains("failed")

|| trimmed.contains("skipped"))

{

summary_line = trimmed.to_string();

continue;

// quiet mode (-q): bare summary without === wrapper, e.g. "5 failed, 1698 passed, 2 skipped in 108.89s"

} else if summary_line.is_empty()

&& !trimmed.starts_with("===")

&& !trimmed.starts_with("FAILED")

&& !trimmed.starts_with("ERROR")

&& (trimmed.contains(" passed")

|| trimmed.contains(" failed")

|| trimmed.contains(" skipped"))

&& trimmed.contains(" in ")

{

summary_line = trimmed.to_string();

continue;

}

// Process based on state

match state {

ParseState::Header => {

if trimmed.starts_with("collected") {

state = ParseState::TestProgress;

}

}

ParseState::TestProgress => {

// Lines like "tests/test_foo.py .... [ 40%]"

if !trimmed.is_empty()

&& !trimmed.starts_with("===")

&& (trimmed.contains(".py") || trimmed.contains("%]"))

{

test_files.push(trimmed.to_string());

}

}

ParseState::Failures => {

// Collect failure details

if trimmed.starts_with("___") {

// New failure section

if !current_failure.is_empty() {

failures.push(current_failure.join("\n"));

current_failure.clear();

}

current_failure.push(trimmed.to_string());

} else if !trimmed.is_empty() && !trimmed.starts_with("===") {

current_failure.push(trimmed.to_string());

}

}

ParseState::Summary => {

// FAILED test lines

if trimmed.starts_with("FAILED") || trimmed.starts_with("ERROR") {

failures.push(trimmed.to_string());

}

}

}

}

// Save last failure if any

if !current_failure.is_empty() {

failures.push(current_failure.join("\n"));

}

// Build compact output

build_pytest_summary(&summary_line, &test_files, &failures)

}

fn build_pytest_summary(summary: &str, _test_files: &[String], failures: &[String]) -> String {

// Parse summary line

let (passed, failed, skipped) = parse_summary_line(summary);

if failed == 0 && passed > 0 {

return format!("Pytest: {} passed", passed);

}

if passed == 0 && failed == 0 && skipped == 0 {

return "Pytest: No tests collected".to_string();

}

let mut result = String::new();

result.push_str(&format!("Pytest: {} passed, {} failed", passed, failed));

if skipped > 0 {

result.push_str(&format!(", {} skipped", skipped));

}

result.push('\n');

result.push_str("═══════════════════════════════════════\n");

if failures.is_empty() {

return result.trim().to_string();

}

// Show failures (limit to key information)

result.push_str("\nFailures:\n");

for (i, failure) in failures.iter().take(5).enumerate() {

// Extract test name and key error info

let lines: Vec<&str> = failure.lines().collect();

// First line is usually test name (after ___)

if let Some(first_line) = lines.first() {

if first_line.starts_with("___") {

// Extract test name between ___

let test_name = first_line.trim_matches('_').trim();

result.push_str(&format!("{}. [FAIL] {}\n", i + 1, test_name));

} else if first_line.starts_with("FAILED") {

// Summary format: "FAILED tests/test_foo.py::test_bar - AssertionError"

let parts: Vec<&str> = first_line.split(" - ").collect();

if let Some(test_path) = parts.first() {

let test_name = test_path.trim_start_matches("FAILED ");

result.push_str(&format!("{}. [FAIL] {}\n", i + 1, test_name));

}

if parts.len() > 1 {

result.push_str(&format!(" {}\n", truncate(parts[1], 100)));

}

continue;

}

}

// Show relevant error lines (assertions, errors, file locations)

let mut relevant_lines = 0;

for line in &lines[1..] {

let line_lower = line.to_lowercase();

let is_relevant = line.trim().starts_with('>')

|| line.trim().starts_with('E')

|| line_lower.contains("assert")

|| line_lower.contains("error")

|| line.contains(".py:");

if is_relevant && relevant_lines < 3 {

result.push_str(&format!(" {}\n", truncate(line, 100)));

relevant_lines += 1;

}

}

if i < failures.len() - 1 {

result.push('\n');

}

}

if failures.len() > 5 {

result.push_str(&format!("\n... +{} more failures\n", failures.len() - 5));

}

result.trim().to_string()

}

fn parse_summary_line(summary: &str) -> (usize, usize, usize) {

let mut passed = 0;

let mut failed = 0;

let mut skipped = 0;

// Parse lines like "=== 4 passed, 1 failed in 0.50s ==="

let parts: Vec<&str> = summary.split(',').collect();

for part in parts {

let words: Vec<&str> = part.split_whitespace().collect();

for (i, word) in words.iter().enumerate() {

if i > 0 {

if word.contains("passed") {

if let Ok(n) = words[i - 1].parse::<usize>() {

passed = n;

}

} else if word.contains("failed") {

if let Ok(n) = words[i - 1].parse::<usize>() {

failed = n;

}

} else if word.contains("skipped") {

if let Ok(n) = words[i - 1].parse::<usize>() {

skipped = n;

}

}

}

}

}

(passed, failed, skipped)

}

#[cfg(test)]

mod tests {

use super::*;

#[test]

fn test_filter_pytest_all_pass() {

let output = r#"=== test session starts ===

platform darwin -- Python 3.11.0

collected 5 items

tests/test_foo.py ..... [100%]

=== 5 passed in 0.50s ==="#;

let result = filter_pytest_output(output);

assert!(result.contains("Pytest"));

assert!(result.contains("5 passed"));

}

#[test]

fn test_filter_pytest_with_failures() {

let output = r#"=== test session starts ===

collected 5 items

tests/test_foo.py ..F.. [100%]

=== FAILURES ===

___ test_something ___

def test_something():

> assert False

E assert False

tests/test_foo.py:10: AssertionError

=== short test summary info ===

FAILED tests/test_foo.py::test_something - assert False

=== 4 passed, 1 failed in 0.50s ==="#;

let result = filter_pytest_output(output);

assert!(result.contains("4 passed, 1 failed"));

assert!(result.contains("test_something"));

assert!(result.contains("assert False"));

}

#[test]

fn test_filter_pytest_multiple_failures() {

let output = r#"=== test session starts ===

collected 3 items

tests/test_foo.py FFF [100%]

=== FAILURES ===

___ test_one ___

E AssertionError: expected 5

___ test_two ___

E ValueError: invalid value

=== short test summary info ===

FAILED tests/test_foo.py::test_one - AssertionError: expected 5

FAILED tests/test_foo.py::test_two - ValueError: invalid value

FAILED tests/test_foo.py::test_three - KeyError

=== 3 failed in 0.20s ==="#;

let result = filter_pytest_output(output);

assert!(result.contains("3 failed"));

assert!(result.contains("test_one"));

assert!(result.contains("test_two"));

assert!(result.contains("expected 5"));

}

#[test]

fn test_filter_pytest_no_tests() {

let output = r#"=== test session starts ===

collected 0 items

=== no tests ran in 0.00s ==="#;

let result = filter_pytest_output(output);

assert!(result.contains("No tests collected"));

}

#[test]

fn test_parse_summary_line() {

assert_eq!(parse_summary_line("=== 5 passed in 0.50s ==="), (5, 0, 0));

assert_eq!(

parse_summary_line("=== 4 passed, 1 failed in 0.50s ==="),

(4, 1, 0)

);

assert_eq!(

parse_summary_line("=== 3 passed, 1 failed, 2 skipped in 1.0s ==="),

(3, 1, 2)

);

}

#[test]

fn test_filter_pytest_quiet_mode_failures() {

// In -q mode, the final summary line has NO === wrapper

// This was causing "No tests collected" to be reported incorrectly

let output = r#"=== test session starts ===

platform linux -- Python 3.12.11, pytest-8.1.0

collected 1705 items

.......F.......

=== FAILURES ===

___ test_something ___

E AssertionError: expected True

=== short test summary info ===

FAILED tests/test_foo.py::test_something - AssertionError

5 failed, 1698 passed, 2 skipped in 108.89s"#;

let result = filter_pytest_output(output);

assert!(

!result.contains("No tests collected"),

"Should not report 'No tests collected' when tests ran. Got: {}",

result

);

assert!(

result.contains("1698") || result.contains("5 failed"),

"Should show actual test counts. Got: {}",

result

);

}

#[test]

fn test_filter_pytest_only_skipped() {

// If only skipped tests, should NOT say "No tests collected"

let output = r#"=== test session starts ===

collected 3 items

=== 3 skipped in 0.10s ==="#;

let result = filter_pytest_output(output);

assert!(

!result.contains("No tests collected"),

"Should not say 'No tests collected' when tests were skipped. Got: {}",

result

);

}

}