Add the ability to write junit test results file for pytest runs.

src/python/pants/backend/python/rules/pytest_runner.py

@@ -29,7 +29,13 @@
 )
 from pants.backend.python.subsystems.pytest import PyTest
 from pants.backend.python.subsystems.subprocess_environment import SubprocessEncodingEnvironment
-from pants.core.goals.test import TestConfiguration, TestDebugRequest, TestOptions, TestResult
+from pants.core.goals.test import (
+    TestConfiguration,
+    TestDebugRequest,
+    TestOptions,
+    TestResult,
+    XmlTestResultsData,
+)
 from pants.core.target_types import FilesSources, ResourcesSources
 from pants.core.util_rules.determine_source_files import SourceFiles, SpecifiedSourceFilesRequest
 from pants.engine.addresses import Addresses
@@ -59,6 +65,7 @@ class TestTargetSetup:
     args: Tuple[str, ...]
     input_files_digest: Digest
     timeout_seconds: Optional[int]
+    xml_results: bool = False
 
     # Prevent this class from being detected by pytest as a test class.
     __test__ = False
@@ -215,6 +222,7 @@ async def setup_pytest_for_target(
         args=(*pytest.options.args, *coverage_args, *specified_source_file_names),
         input_files_digest=merged_input_files,
         timeout_seconds=config.timeout.calculate_from_global_options(pytest),
+        xml_results=test_options.values.xml_results,
     )
 
 
@@ -228,15 +236,23 @@ async def run_python_test(
     test_options: TestOptions,
 ) -> TestResult:
     """Runs pytest for one target."""
-    env = {"PYTEST_ADDOPTS": f"--color={'yes' if global_options.options.colors else 'no'}"}
+    add_opts = [f"--color={'yes' if global_options.options.colors else 'no'}"]
+    test_results_file = f"{config.address.path_safe_spec}.xml"
+    if test_setup.xml_results:
+        add_opts.append(f"--junitxml={test_results_file}")
+    env = {"PYTEST_ADDOPTS": " ".join(add_opts)}
     run_coverage = test_options.values.run_coverage
+    output_dirs = [".coverage"] if run_coverage else []
+    if test_setup.xml_results:
+        output_dirs.append(test_results_file)
+
     request = test_setup.test_runner_pex.create_execute_request(
         python_setup=python_setup,
         subprocess_encoding_environment=subprocess_encoding_environment,
         pex_path=f"./{test_setup.test_runner_pex.output_filename}",
         pex_args=test_setup.args,
         input_files=test_setup.input_files_digest,
-        output_directories=(".coverage",) if run_coverage else None,
+        output_directories=tuple(output_dirs) if output_dirs else None,
         description=f"Run Pytest for {config.address.reference()}",
         timeout_seconds=(
             test_setup.timeout_seconds if test_setup.timeout_seconds is not None else 9999
@@ -247,7 +263,15 @@ async def run_python_test(
     coverage_data = (
         PytestCoverageData(config.address, result.output_directory_digest) if run_coverage else None
     )
-    return TestResult.from_fallible_process_result(result, coverage_data=coverage_data)
+    # TODO: How to distinguish coverage data from XML results data.
+    test_results = (
+        XmlTestResultsData(config.address, result.output_directory_digest)
+        if test_setup.xml_results
+        else None
+    )
+    return TestResult.from_fallible_process_result(
+        result, coverage_data=coverage_data, test_results=test_results
+    )
 
 
 @named_rule(desc="Run pytest in an interactive process")

src/python/pants/core/goals/test.py

@@ -47,19 +47,24 @@ class TestResult:
     stdout: str
     stderr: str
     coverage_data: Optional["CoverageData"] = None
+    test_results: Optional["XmlTestResultsData"] = None
 
     # Prevent this class from being detected by pytest as a test class.
     __test__ = False
 
     @staticmethod
     def from_fallible_process_result(
-        process_result: FallibleProcessResult, *, coverage_data: Optional["CoverageData"] = None,
+        process_result: FallibleProcessResult,
+        *,
+        coverage_data: Optional["CoverageData"] = None,
+        test_results: Optional["XmlTestResultsData"] = None,
     ) -> "TestResult":
         return TestResult(
             status=Status.SUCCESS if process_result.exit_code == 0 else Status.FAILURE,
             stdout=process_result.stdout.decode(),
             stderr=process_result.stderr.decode(),
             coverage_data=coverage_data,
+            test_results=test_results,
         )
 
 
@@ -93,6 +98,12 @@ class AddressAndTestResult:
     test_result: TestResult
 
 
+@dataclass(frozen=True)
+class XmlTestResultsData:
+    address: Address
+    digest: Digest
+
+
 class CoverageData(ABC):
     """Base class for inputs to a coverage report.
 
@@ -185,6 +196,12 @@ def register_options(cls, register) -> None:
             help="If a coverage report file is generated, open it on the local system if the "
             "system supports this.",
         )
+        register(
+            "--xml-results",
+            type=bool,
+            default=False,
+            help="Enable generating XML test results file",
+        )
 
 
 class Test(Goal):