87f22614a3a97d93aaf570ba1c3eeefcbfdb8d5f,src/python/pants/core/goals/test_test.py,,run_test_rule,#,112

Before Change

    include_sources: bool = True,
    valid_targets: bool = True,
) -> Tuple[int, str]:
    console = MockConsole(use_colors=False)
    test_subsystem = create_goal_subsystem(
        TestSubsystem,
        debug=debug,
        use_coverage=use_coverage,
        output=output,
        extra_env_vars=[],
    )
    interactive_runner = InteractiveRunner(rule_runner.scheduler)
    workspace = Workspace(rule_runner.scheduler)
    union_membership = UnionMembership(
        {TestFieldSet: [field_set], CoverageDataCollection: [MockCoverageDataCollection]}
    )

    def mock_find_valid_field_sets(
        _: TargetRootsToFieldSetsRequest,
    ) -> TargetRootsToFieldSets:
        if not valid_targets:
            return TargetRootsToFieldSets({})
        return TargetRootsToFieldSets({tgt: [field_set.create(tgt)] for tgt in targets})

    def mock_debug_request(_: TestFieldSet) -> TestDebugRequest:
        digest = rule_runner.request(
            Digest, [CreateDigest((FileContent(path="program.py", content=b"def test(): pass"),))]
        )
        process = InteractiveProcess(["/usr/bin/python", "program.py"], input_digest=digest)
        return TestDebugRequest(process)

    def mock_coverage_report_generation(
        coverage_data_collection: MockCoverageDataCollection,
    ) -> CoverageReports:
        addresses = ", ".join(
            coverage_data.address.spec for coverage_data in coverage_data_collection
        )
        console_report = ConsoleCoverageReport(f"Ran coverage on {addresses}")
        return CoverageReports(reports=(console_report,))

    result: Test = run_rule_with_mocks(
        run_tests,
        rule_args=[
            console,
            test_subsystem,
            interactive_runner,
            workspace,
            union_membership,
        ],
        mock_gets=[
            MockGet(
                output_type=TargetRootsToFieldSets,
                input_type=TargetRootsToFieldSetsRequest,
                mock=mock_find_valid_field_sets,
            ),
            MockGet(
                output_type=EnrichedTestResult,
                input_type=TestFieldSet,
                mock=lambda fs: fs.test_result,
            ),
            MockGet(
                output_type=TestDebugRequest,
                input_type=TestFieldSet,
                mock=mock_debug_request,
            ),
            MockGet(
                output_type=FieldSetsWithSources,
                input_type=FieldSetsWithSourcesRequest,
                mock=lambda field_sets: FieldSetsWithSources(field_sets if include_sources else ()),
            ),
            // Merge XML results.
            MockGet(
                output_type=Digest,
                input_type=MergeDigests,
                mock=lambda _: EMPTY_DIGEST,
            ),
            MockGet(
                output_type=CoverageReports,
                input_type=CoverageDataCollection,
                mock=mock_coverage_report_generation,
            ),
            MockGet(
                output_type=OpenFiles,
                input_type=OpenFilesRequest,
                mock=lambda _: OpenFiles(()),
            ),
        ],
        union_membership=union_membership,
    )
    assert not console.stdout.getvalue()
    return result.exit_code, console.stderr.getvalue()


def test_empty_target_noops(rule_runner: RuleRunner) -> None:
    exit_code, stderr = run_test_rule(

After Change

        console_report = ConsoleCoverageReport(f"Ran coverage on {addresses}")
        return CoverageReports(reports=(console_report,))

    with mock_console(rule_runner.options_bootstrapper) as (console, stdio_reader):
        result: Test = run_rule_with_mocks(
            run_tests,
            rule_args=[
                console,
                test_subsystem,
                interactive_runner,
                workspace,
                union_membership,
            ],
            mock_gets=[
                MockGet(
                    output_type=TargetRootsToFieldSets,
                    input_type=TargetRootsToFieldSetsRequest,
                    mock=mock_find_valid_field_sets,
                ),
                MockGet(
                    output_type=EnrichedTestResult,
                    input_type=TestFieldSet,
                    mock=lambda fs: fs.test_result,
                ),
                MockGet(
                    output_type=TestDebugRequest,
                    input_type=TestFieldSet,
                    mock=mock_debug_request,
                ),
                MockGet(
                    output_type=FieldSetsWithSources,
                    input_type=FieldSetsWithSourcesRequest,
                    mock=lambda field_sets: FieldSetsWithSources(
                        field_sets if include_sources else ()
                    ),
                ),
                // Merge XML results.
                MockGet(
                    output_type=Digest,
                    input_type=MergeDigests,
                    mock=lambda _: EMPTY_DIGEST,
                ),
                MockGet(
                    output_type=CoverageReports,
                    input_type=CoverageDataCollection,
                    mock=mock_coverage_report_generation,
                ),
                MockGet(
                    output_type=OpenFiles,
                    input_type=OpenFilesRequest,
                    mock=lambda _: OpenFiles(()),
                ),
            ],
            union_membership=union_membership,
        )
        assert not stdio_reader.get_stdout()
        return result.exit_code, stdio_reader.get_stderr()


def test_empty_target_noops(rule_runner: RuleRunner) -> None:
    exit_code, stderr = run_test_rule(
        rule_runner,
        field_set=SuccessfulFieldSet,

Instances