Support failure_expanded

Currently, the entire failure from PyTest (PyTest calls it longreprtext) is
set as failure_reason.

However, failure_reason is intended for a short one-line summary, and
gets truncated to ~1KIB, which isn't great for longreprtext.

Buildkite also supports failure_expanded, to included detailed
information and backtraces on errors.

So, derive that from the various shapes that longrepr can take, and
include them in the uploaded JSON payload.

Author: pda

src/buildkite_test_collector/collector/payload.py

@@ -1,7 +1,7 @@
 """Buildkite Test Analytics payload"""
 
 from dataclasses import dataclass, replace, field
-from typing import Dict, Tuple, Optional, Union, Literal, List
+from typing import Dict, Tuple, Optional, Union, Literal, List, Iterable, Mapping
 from datetime import timedelta
 from uuid import UUID
 
@@ -25,6 +25,7 @@ class TestResultPassed:
 class TestResultFailed:
     """Represents a failed test result"""
     failure_reason: Optional[str]
+    failure_expanded: Optional[Iterable[Mapping[str, Iterable[str]]]] = None
 
 
 @dataclass(frozen=True)
@@ -162,9 +163,10 @@ def passed(self) -> 'TestData':
         """Mark this test as passed"""
         return replace(self, result=TestResultPassed())
 
-    def failed(self, failure_reason=None) -> 'TestData':
+    def failed(self, failure_reason=None, failure_expanded=None) -> 'TestData':
         """Mark this test as failed"""
-        return replace(self, result=TestResultFailed(failure_reason=failure_reason))
+        result = TestResultFailed(failure_reason=failure_reason, failure_expanded=failure_expanded)
+        return replace(self, result=result)
 
     def skipped(self) -> 'TestData':
         """Mark this test as skipped"""
@@ -199,6 +201,8 @@ def as_json(self, started_at: Instant) -> JsonDict:
             attrs["result"] = "failed"
             if self.result.failure_reason is not None:
                 attrs["failure_reason"] = self.result.failure_reason
+            if self.result.failure_expanded is not None:
+                attrs["failure_expanded"] = self.result.failure_expanded
 
         if isinstance(self.result, TestResultSkipped):
             attrs["result"] = "skipped"

src/buildkite_test_collector/pytest_plugin/buildkite_plugin.py

@@ -7,6 +7,7 @@
 
 from ..collector.payload import TestData
 from .logger import logger
+from .failure_reasons import failure_reasons
 
 class BuildkitePlugin:
     """Buildkite test collector plugin for Pytest"""
@@ -53,7 +54,11 @@ def pytest_runtest_logreport(self, report):
                 test_data = test_data.passed()
 
             if report.failed:
-                test_data = test_data.failed(report.longreprtext)
+                failure_reason, failure_expanded = failure_reasons(longrepr=report.longrepr)
+                test_data = test_data.failed(
+                    failure_reason=failure_reason,
+                    failure_expanded=failure_expanded
+                )
 
             if report.skipped:
                 test_data = test_data.skipped()
@@ -103,6 +108,7 @@ def pytest_runtest_logfinish(self, nodeid, location):  # pylint: disable=unused-
 
     def finalize_test(self, nodeid):
         """ Attempting to move test data for a nodeid to payload area for upload """
+        logger.debug('Entering finalize_test for %s', nodeid)
         test_data = self.in_flight.get(nodeid)
         if test_data:
             del self.in_flight[nodeid]

src/buildkite_test_collector/pytest_plugin/failure_reasons.py

@@ -0,0 +1,76 @@
+"""Buildkite Test Engine PyTest failure reason mapping"""
+
+from typing import Iterable, Mapping
+
+# importing these privates isn't ideal, but we're only using them for type checking
+from _pytest._code.code import ExceptionInfo, ExceptionRepr, TerminalRepr
+
+# pylint: disable=too-many-locals
+# pylint: disable=too-many-return-statements
+def failure_reasons(
+    longrepr: None | ExceptionInfo[BaseException] | tuple[str, int, str] | str | TerminalRepr
+) -> tuple[str | None, Iterable[Mapping[str, Iterable[str]]] | None]:
+    """
+    Derives Buildkite's failure_reason & failure_expanded from PyTest's longrepr.
+
+    Args:
+        longrepr: The PyTest longrepr object containing failure information
+
+    Returns:
+        A tuple containing:
+        - A string with the failure reason or None if not available
+        - A list of mappings with additional failure details or None if not available
+    """
+    match longrepr:
+        case None:
+            return None, None
+
+        case str() as s:
+            lines = s.splitlines()
+            failure_reason = lines[0] if lines else s
+            return failure_reason, [{"expanded": lines[1:]}]
+
+        case (path, line, msg)  if \
+                isinstance(path, str) and isinstance(line, int) and isinstance(msg, str):
+            failure_reason = msg
+            return failure_reason, [{"expanded": [], "backtrace": [f"{path}:{line}"]}]
+
+        case ExceptionInfo() as exc_info:
+            failure_reason = exc_info.exconly()
+            expanded = []
+            backtrace = []
+
+            if hasattr(exc_info, "traceback") and exc_info.traceback:
+                for entry in exc_info.traceback:
+                    backtrace.append(f"{entry.path}:{entry.lineno}: {entry.name}")
+                    source = entry.getsource() if hasattr(entry, "getsource") else None
+                    if source:
+                        expanded.extend(str(line) for line in source)
+
+            failure_expanded = {}
+            if len(expanded) > 0:
+                failure_expanded["expanded"] = expanded
+            if len(backtrace) > 0:
+                failure_expanded["backtrace"] = backtrace
+
+            return failure_reason, [failure_expanded]
+
+        case ExceptionRepr() as er if er.reprcrash is not None:
+            failure_reason = er.reprcrash.message # e.g. "ZeroDivisionError: division by zero"
+            failure_expanded = [{"expanded": str(er).splitlines()}]
+            try:
+                failure_expanded[0]["backtrace"] = [
+                    str(getattr(entry, 'reprfileloc', entry))
+                    for entry in er.reprtraceback.reprentries
+                ]
+            except AttributeError:
+                pass
+            return failure_reason, failure_expanded
+
+        case _:
+            lines = str(longrepr).splitlines()
+            if len(lines) == 0:
+                return None, None
+            if len(lines) == 1:
+                return lines[0], None
+            return lines[0], [{"expanded": lines[1:]}]

tests/buildkite_test_collector/collector/test_payload.py

@@ -161,7 +161,8 @@ def test_test_data_as_json_when_failed(failed_test):
     json = failed_test.as_json(Instant.now())
 
     assert json["result"] == "failed"
-    assert json["failure_reason"] == failed_test.result.failure_reason
+    assert json["failure_reason"] == "bogus"
+    assert json["failure_expanded"] == [{'expanded': ['test failed'], 'backtrace': ['test.py:1']}]
 
 
 def test_test_data_as_json_when_skipped(skipped_test):

tests/buildkite_test_collector/conftest.py

@@ -26,7 +26,10 @@ def successful_test(history_finished) -> TestData:
 
 @pytest.fixture
 def failed_test(successful_test) -> TestData:
-    return replace(successful_test, result=TestResultFailed("bogus"))
+    return replace(
+        successful_test,
+        result=TestResultFailed("bogus", [{"expanded": ["test failed"], "backtrace": ["test.py:1"]}])
+    )
 
 
 @pytest.fixture

tests/buildkite_test_collector/pytest_plugin/test_plugin.py

@@ -1,9 +1,11 @@
 import json
 import pytest
 
-from buildkite_test_collector.collector.payload import Payload
+from buildkite_test_collector.collector.payload import Payload, TestData, TestResultFailed, TestResultPassed, TestResultSkipped
 from buildkite_test_collector.pytest_plugin import BuildkitePlugin
 
+from _pytest._code.code import ExceptionInfo
+from _pytest.reports import TestReport
 
 def test_runtest_logstart_with_unstarted_payload(fake_env):
     payload = Payload.init(fake_env)
@@ -16,6 +18,122 @@ def test_runtest_logstart_with_unstarted_payload(fake_env):
     assert plugin.payload.started_at is not None
 
 
+def test_pytest_runtest_logreport_simple_pass(fake_env):
+    payload = Payload.init(fake_env)
+    plugin = BuildkitePlugin(payload)
+
+    location = ("", None, "")
+    report = TestReport(nodeid="", location=location, keywords={}, outcome="passed", longrepr=None, when="call")
+
+    plugin.pytest_runtest_logstart(report.nodeid, location)
+    plugin.pytest_runtest_logreport(report)
+
+    test_data = plugin.in_flight.get(report.nodeid)
+    assert test_data is not None
+
+    assert isinstance(test_data.result, TestResultPassed)
+
+
+def test_pytest_runtest_logreport_fail_oneline(fake_env):
+    payload = Payload.init(fake_env)
+    plugin = BuildkitePlugin(payload)
+
+    location = ("", None, "")
+    longrepr = "the reason the test failed"
+    report = TestReport(nodeid="", location=location, keywords={}, outcome="failed", longrepr=longrepr, when="call")
+
+    plugin.pytest_runtest_logstart(report.nodeid, location)
+    plugin.pytest_runtest_logreport(report)
+    test_data = plugin.in_flight.get(report.nodeid)
+    plugin.pytest_runtest_logfinish(report.nodeid, location)
+
+    assert isinstance(test_data, TestData)
+    assert isinstance(test_data.result, TestResultFailed)
+    assert test_data.result.failure_reason == "the reason the test failed"
+
+
+def test_pytest_runtest_logreport_fail_multiline(fake_env):
+    payload = Payload.init(fake_env)
+    plugin = BuildkitePlugin(payload)
+
+    location = ("", None, "")
+    longrepr = "the reason the test failed\n.. is quite complicated\nso here is more detail"
+    report = TestReport(nodeid="", location=location, keywords={}, outcome="failed", longrepr=longrepr, when="call")
+
+    plugin.pytest_runtest_logstart(report.nodeid, location)
+    plugin.pytest_runtest_logreport(report)
+    test_data = plugin.in_flight.get(report.nodeid)
+    plugin.pytest_runtest_logfinish(report.nodeid, location)
+
+    assert isinstance(test_data, TestData)
+    assert isinstance(test_data.result, TestResultFailed)
+    assert test_data.result.failure_reason == "the reason the test failed"
+    assert test_data.result.failure_expanded == [{"expanded": [".. is quite complicated", "so here is more detail"]}]
+
+
+def test_pytest_runtest_logreport_fail_exception(fake_env):
+    payload = Payload.init(fake_env)
+    plugin = BuildkitePlugin(payload)
+
+    location = ("", None, "")
+    try:
+        raise Exception("a fake exception for testing")
+    except Exception as e:
+        longrepr = ExceptionInfo.from_exception(e)
+    report = TestReport(nodeid="", location=location, keywords={}, outcome="failed", longrepr=longrepr, when="call")
+
+    plugin.pytest_runtest_logstart(report.nodeid, location)
+    plugin.pytest_runtest_logreport(report)
+    test_data = plugin.in_flight.get(report.nodeid)
+    plugin.pytest_runtest_logfinish(report.nodeid, location)
+
+    assert isinstance(test_data, TestData)
+    assert isinstance(test_data.result, TestResultFailed)
+    assert test_data.result.failure_reason == "Exception: a fake exception for testing"
+
+    assert isinstance(test_data.result.failure_expanded, list)
+    fe = test_data.result.failure_expanded[0]
+    assert list(fe.keys()) == ["expanded", "backtrace"]
+    assert isinstance(fe["expanded"], list)
+    assert len(fe["expanded"]) > 0
+    assert isinstance(fe["backtrace"], list)
+    assert len(fe["backtrace"]) > 0
+
+
+def test_pytest_runtest_logreport_simple_fail(fake_env):
+    payload = Payload.init(fake_env)
+    plugin = BuildkitePlugin(payload)
+
+    location = ("", None, "")
+    report = TestReport(nodeid="", location=location, keywords={}, outcome="failed", longrepr=None, when="call")
+
+    plugin.pytest_runtest_logstart(report.nodeid, location)
+    plugin.pytest_runtest_logreport(report)
+
+    test_data = plugin.in_flight.get(report.nodeid)
+    assert test_data is not None
+
+    assert isinstance(test_data.result, TestResultFailed)
+
+
+def test_pytest_runtest_logreport_simple_skip(fake_env):
+    payload = Payload.init(fake_env)
+    plugin = BuildkitePlugin(payload)
+
+    location = ("path/to/test.py", 100, "")
+    longrepr = ("path/to/test.py", 123, "skippy")
+    report = TestReport(nodeid="", location=location, keywords={}, outcome="skipped", longrepr=longrepr, when="call")
+
+    plugin.pytest_runtest_logstart(report.nodeid, location)
+    plugin.pytest_runtest_logreport(report)
+
+    test_data = plugin.in_flight.get(report.nodeid)
+    assert isinstance(test_data, TestData)
+
+    assert isinstance(test_data.result, TestResultSkipped)
+    # TODO: track skip reason as failure_reason via longrepr
+
+
 def test_save_json_payload_without_merge(fake_env, tmp_path, successful_test):
     payload = Payload.init(fake_env)
     payload = Payload.started(payload)