Pytest sucess rate

conftest.py

import pytest
import re
from collections import defaultdict


# --- Master-side data ---
# This list is created only on the master process to aggregate results.
# We check if a process is a worker by seeing if it has the 'workerinput' attribute.
master_reports: list[pytest.TestReport] = []


def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]):
    print("[conftest] pytest_collection_modifyitems")


def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo) -> pytest.TestReport:
    if call.when == "call":
        marker = item.get_closest_marker("success_rate")
        if marker:
            item.user_properties.append(("success_rate", marker.args, marker.kwargs))


def pytest_runtest_logreport(report: pytest.TestReport):
    master_reports.append(report)


def get_suite_key(nodeid: str, group: bool = True) -> str:
    """Generate a unique key for a test suite by parsing its nodeid."""
    match = re.match(r"(.*)\[(.*)\]$", nodeid)
    if not match:
        return nodeid

    base, param = match.groups()

    if group:
        return base

    if not group:
        math_param_and_iteration = re.match(r"(.*)-(\d+)$", param)

        if math_param_and_iteration:
            param_base, _ = math_param_and_iteration.groups()
            return f"{base}[{param_base}]"
        else:
            return nodeid


def get_success_rate(report: pytest.TestReport):
    """Extract the success rate from a test report's user properties."""
    for prop, args, kwargs in report.user_properties:
        if prop == "success_rate":
            kwargs["success_rate"] = kwargs.get("success_rate", args[0])
            return kwargs
    return None


def group_reports_by_suite(
    reports: list[pytest.TestReport],
) -> dict[str, list[pytest.TestReport]]:
    """Group test reports by their suite key."""
    suites = defaultdict(list)
    for report in reports:
        success_rate_conf = get_success_rate(report)
        suite_key = report.nodeid if not success_rate_conf else get_suite_key(report.nodeid, group=success_rate_conf.get("group", False))
        suites[suite_key].append(report)
    return suites


def color_outcome_print(outcome: str) -> str:
    """Return a colored string based on the test outcome."""
    if outcome == "passed":
        return f"\033[92m{outcome}\033[0m"  # Green
    elif outcome == "failed":
        return f"\033[91m{outcome}\033[0m"  # Red
    elif outcome == "skipped":
        return f"\033[93m{outcome}\033[0m"  # Yellow
    return outcome


def process_suite_with_success_rate(
    suite_key: str, reports: list[pytest.TestReport], required_rate: float
):
    """Process and print the summary for a test suite with a success rate."""
    passed_count = sum(1 for r in reports if r.outcome == "passed")
    total_count = len(reports)
    actual_rate = passed_count / total_count if total_count > 0 else 0
    outcome = "passed" if actual_rate >= required_rate else "failed"

    print(
        f"{suite_key}: {color_outcome_print(outcome)}\t"
        f"Success Rate: {actual_rate:.2f}/{required_rate:.2f} ({passed_count}/{total_count} passed)"
    )
    return outcome == "passed"


def process_single_test(report: pytest.TestReport):
    """Process and print the summary for a single test report."""
    print(f"{report.nodeid}: {color_outcome_print(report.outcome)}")
    return report.outcome == "passed"


def pytest_sessionfinish(session: pytest.Session, exitstatus: int):
    """Summarize test results at the end of the session."""
    print(f"[conftest] pytest_sessionfinish {session=}")

    call_reports = [r for r in master_reports if r.when == "call"]
    suites = group_reports_by_suite(call_reports)

    print("\n--- Summary of tests that ran ---")
    success = []
    for suite_key, reports in suites.items():
        success_rate_conf = get_success_rate(reports[0]) or {}
        required_rate = success_rate_conf.get("success_rate", None)

        assert required_rate is not None or len(reports) == 1, (
            f"Test suite {suite_key} has multiple reports but no success rate defined."
        )
        if required_rate is not None:
            s = process_suite_with_success_rate(suite_key, reports, required_rate)
        else:
            s = process_single_test(reports[0])

        success.append(s)

    overall_success = all(success)
    print(f"{success=} {overall_success=}")
    session.exitstatus = 0 if overall_success else 1

Makefile

test:
	ENV=test python -m pytest tests -s -n 2

some_test.py

@pytest.mark.asyncio
@pytest.mark.success_rate(0.9, group=True)
@pytest.mark.parametrize("user, assistant", [
    ("bonjour", "comment puis je vous aider?"),
    ("hello", "how may I help you?"),
])
async def test_answ(user, assistant):
    assert (await llm(user)) == assistant