19641045249

Committed 24 Nov 2025 04:11PM UTC coverage: 80.291% (+0.008%) from 80.283%

Build # 19641045249

Build Type

Pull #22915

github

Committed by

web-flow

Commit Message

Merge 4ca3092a3 into 612e19cd8

Pull Request Pull Request #22915: Abstract PosixFS away.

Run Details

78389 of 97631 relevant lines covered (80.29%)

3.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.89

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

# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

from abc import ABCMeta, abstractmethod
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path
from textwrap import dedent
from typing import Any, TypeVar

import pytest

from pants.base.specs import Specs
from pants.core.goals.fix import FixFilesRequest, FixTargetsRequest
from pants.core.goals.fmt import FmtFilesRequest, FmtTargetsRequest
from pants.core.goals.lint import (
    AbstractLintRequest,
    Lint,
    LintFilesRequest,
    LintResult,
    LintSubsystem,
    LintTargetsRequest,
    Partitions,
)
from pants.core.goals.lint_goal import lint
from pants.core.util_rules.distdir import DistDir
from pants.core.util_rules.partitions import PartitionerType, _EmptyMetadata
from pants.engine.addresses import Address
from pants.engine.fs import SpecsPaths, Workspace
from pants.engine.internals.native_engine import EMPTY_SNAPSHOT
from pants.engine.rules import QueryRule
from pants.engine.target import Field, FieldSet, FilteredTargets, MultipleSourcesField, Target
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.option_types import SkipOption
from pants.option.subsystem import Subsystem
from pants.testutil.option_util import create_goal_subsystem
from pants.testutil.rule_runner import RuleRunner, mock_console, run_rule_with_mocks
from pants.util.logging import LogLevel
from pants.util.meta import classproperty

_LintRequestT = TypeVar("_LintRequestT", bound=AbstractLintRequest)


class MockMultipleSourcesField(MultipleSourcesField):
    pass


class MockRequiredField(Field):
    alias = "required"
    required = True


class MockTarget(Target):
    alias = "mock_target"
    core_fields = (MockMultipleSourcesField, MockRequiredField)


@dataclass(frozen=True)
class MockLinterFieldSet(FieldSet):
    required_fields = (MultipleSourcesField,)
    sources: MultipleSourcesField
    required: MockRequiredField


class MockLintRequest(AbstractLintRequest, metaclass=ABCMeta):
    @staticmethod
    @abstractmethod
    def exit_code(_: Iterable[Address]) -> int:
        pass

    @classmethod
    @abstractmethod
    def get_lint_result(cls, elements: Iterable) -> LintResult:
        pass


class MockLintTargetsRequest(MockLintRequest, LintTargetsRequest):
    field_set_type = MockLinterFieldSet

    @classmethod
    def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:
        addresses = [field_set.address for field_set in field_sets]
        return LintResult(cls.exit_code(addresses), "", "", cls.tool_name)


class SuccessfulRequest(MockLintTargetsRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Successful Linter"

    @classproperty
    def tool_id(cls) -> str:
        return "successfullinter"

    @staticmethod
    def exit_code(_: Iterable[Address]) -> int:
        return 0


class FailingRequest(MockLintTargetsRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Failing Linter"

    @classproperty
    def tool_id(cls) -> str:
        return "failinglinter"

    @staticmethod
    def exit_code(_: Iterable[Address]) -> int:
        return 1


class ConditionallySucceedsRequest(MockLintTargetsRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Conditionally Succeeds Linter"

    @classproperty
    def tool_id(cls) -> str:
        return "conditionallysucceedslinter"

    @staticmethod
    def exit_code(addresses: Iterable[Address]) -> int:
        if any(address.target_name == "bad" for address in addresses):
            return 127
        return 0


class SkippedRequest(MockLintTargetsRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Skipped Linter"

    @classproperty
    def tool_id(cls) -> str:
        return "skippedlinter"

    @staticmethod
    def exit_code(_) -> int:
        return 0


class InvalidField(MultipleSourcesField):
    pass


class InvalidFieldSet(MockLinterFieldSet):
    required_fields = (InvalidField,)


class InvalidRequest(MockLintTargetsRequest):
    field_set_type = InvalidFieldSet

    @classproperty
    def tool_name(cls) -> str:
        return "Invalid Linter"

    @classproperty
    def tool_id(cls) -> str:
        return "invalidlinter"

    @staticmethod
    def exit_code(_: Iterable[Address]) -> int:
        return -1


def _all_lint_requests() -> Iterable[type[MockLintRequest]]:
    classes = [MockLintRequest]
    while classes:
        cls = classes.pop()
        subclasses = cls.__subclasses__()
        classes.extend(subclasses)
        yield from subclasses


def mock_target_partitioner(__implicitly: tuple) -> Partitions[MockLinterFieldSet, Any]:
    request, typ = next(iter(__implicitly[0].items()))
    assert typ == LintTargetsRequest.PartitionRequest
    if type(request) is SkippedRequest.PartitionRequest:
        return Partitions()

    operates_on_paths = {
        getattr(cls, "PartitionRequest"): cls._requires_snapshot for cls in _all_lint_requests()
    }[type(request)]
    if operates_on_paths:
        return Partitions.single_partition(fs.sources.globs for fs in request.field_sets)

    return Partitions.single_partition(request.field_sets)


class MockFilesRequest(MockLintRequest, LintFilesRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Files Linter"

    @classproperty
    def tool_id(cls) -> str:
        return "fileslinter"

    @classmethod
    def get_lint_result(cls, files: Iterable[str]) -> LintResult:
        return LintResult(0, "", "", cls.tool_name)


def mock_file_partitioner(__implicitly: dict) -> Partitions[str, Any]:
    request, typ = next(iter(__implicitly[0].items()))
    assert typ == LintFilesRequest.PartitionRequest
    return Partitions.single_partition(request.files)


def mock_lint_partition(__implicitly: dict) -> LintResult:
    request, typ = next(iter(__implicitly[0].items()))
    assert typ == AbstractLintRequest.Batch
    request_type = {cls.Batch: cls for cls in _all_lint_requests()}[type(request)]
    return request_type.get_lint_result(request.elements)


class MockFmtRequest(MockLintRequest, FmtTargetsRequest):
    field_set_type = MockLinterFieldSet


class SuccessfulFormatter(MockFmtRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Successful Formatter"

    @classproperty
    def tool_id(cls) -> str:
        return "successfulformatter"

    @classmethod
    def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:
        return LintResult(0, "", "", cls.tool_name)


class FailingFormatter(MockFmtRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Failing Formatter"

    @classproperty
    def tool_id(cls) -> str:
        return "failingformatter"

    @classmethod
    def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:
        return LintResult(1, "", "", cls.tool_name)


class BuildFileFormatter(MockLintRequest, FmtFilesRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Bob The BUILDer"

    @classproperty
    def tool_id(cls) -> str:
        return "bob"

    @classmethod
    def get_lint_result(cls, files: Iterable[str]) -> LintResult:
        return LintResult(0, "", "", cls.tool_name)


class MockFixRequest(MockLintRequest, FixTargetsRequest):
    field_set_type = MockLinterFieldSet


class SuccessfulFixer(MockFixRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Successful Fixer"

    @classproperty
    def tool_id(cls) -> str:
        return "successfulfixer"

    @classmethod
    def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:
        return LintResult(0, "", "", cls.tool_name)


class FailingFixer(MockFixRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "Failing Fixer"

    @classproperty
    def tool_id(cls) -> str:
        return "failingfixer"

    @classmethod
    def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:
        return LintResult(1, "", "", cls.tool_name)


class BuildFileFixer(MockLintRequest, FixFilesRequest):
    @classproperty
    def tool_name(cls) -> str:
        return "BUILD Annually"

    @classproperty
    def tool_id(cls) -> str:
        return "buildannually"

    @classmethod
    def get_lint_result(cls, files: Iterable[str]) -> LintResult:
        return LintResult(0, "", "", cls.tool_name)


def mock_fix_partition_as_linter(request) -> LintResult:
    request_type = {cls.Batch: cls for cls in _all_lint_requests()}[type(request)]
    return request_type.get_lint_result(request.elements)


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner()


def make_target(address: Address | None = None) -> Target:
    return MockTarget(
        {MockRequiredField.alias: "present"}, address or Address("", target_name="tests")
    )


def run_lint_rule(
    rule_runner: RuleRunner,
    *,
    lint_request_types: Iterable[type[_LintRequestT]],
    targets: list[Target],
    batch_size: int = 128,
    only: list[str] | None = None,
    skip_formatters: bool = False,
    skip_fixers: bool = False,
) -> tuple[int, str]:
    union_membership = UnionMembership.from_rules(
        {
            *[UnionRule(AbstractLintRequest, t) for t in lint_request_types],
            *[UnionRule(AbstractLintRequest.Batch, rt.Batch) for rt in lint_request_types],
            *[
                UnionRule(LintTargetsRequest.PartitionRequest, rt.PartitionRequest)
                for rt in lint_request_types
                if issubclass(rt, LintTargetsRequest)
            ],
            *[
                UnionRule(LintFilesRequest.PartitionRequest, rt.PartitionRequest)
                for rt in lint_request_types
                if issubclass(rt, LintFilesRequest)
            ],
        }
    )
    lint_subsystem = create_goal_subsystem(
        LintSubsystem,
        batch_size=batch_size,
        only=only or [],
        skip_formatters=skip_formatters,
        skip_fixers=skip_fixers,
    )

    with mock_console(rule_runner.options_bootstrapper) as (console, stdio_reader):
        result: Lint = run_rule_with_mocks(
            lint,
            rule_args=[
                console,
                Workspace(rule_runner.scheduler, _enforce_effects=False),
                Specs.empty(),
                lint_subsystem,
                union_membership,
                DistDir(relpath=Path("dist")),
            ],
            mock_calls={
                "pants.engine.internals.graph.filter_targets": lambda __implicitly: FilteredTargets(
                    tuple(targets)
                ),
                "pants.engine.internals.specs_rules.resolve_specs_paths": lambda _: SpecsPaths(
                    ("f.txt", "BUILD"), ()
                ),
                "pants.core.goals.lint.partition_targets": mock_target_partitioner,
                "pants.core.goals.lint.partition_files": mock_file_partitioner,
                "pants.core.goals.lint.lint_batch": mock_lint_partition,
                "pants.core.goals.lint_goal.run_fixer_or_formatter_as_linter": mock_fix_partition_as_linter,
                "pants.engine.intrinsics.digest_to_snapshot": lambda __implicitly: EMPTY_SNAPSHOT,
            },
            union_membership=union_membership,
            # We don't want temporary warnings to interfere with our expected output.
            show_warnings=False,
        )
        assert not stdio_reader.get_stdout()
        return result.exit_code, stdio_reader.get_stderr()


def test_duplicate_files_in_fixer(rule_runner: RuleRunner) -> None:
    assert MockFixRequest.Batch(
        "tool_name",
        ("a", "a"),
        _EmptyMetadata(),
        rule_runner.make_snapshot({"a": ""}),
    ).files == ("a",)


def test_invalid_target_noops(rule_runner: RuleRunner) -> None:
    exit_code, stderr = run_lint_rule(
        rule_runner, lint_request_types=[InvalidRequest], targets=[make_target()]
    )
    assert exit_code == 0
    assert stderr == ""


def test_summary(rule_runner: RuleRunner) -> None:
    """Test that we render the summary correctly.

    This tests that we:
    * Merge multiple results belonging to the same linter (`--per-file-caching`).
    * Decide correctly between skipped, failed, and succeeded.
    """
    good_address = Address("", target_name="good")
    bad_address = Address("", target_name="bad")

    request_types = [
        ConditionallySucceedsRequest,
        FailingRequest,
        SkippedRequest,
        SuccessfulRequest,
        SuccessfulFormatter,
        FailingFormatter,
        BuildFileFormatter,
        SuccessfulFixer,
        FailingFixer,
        BuildFileFixer,
        MockFilesRequest,
    ]
    targets = [make_target(good_address), make_target(bad_address)]

    exit_code, stderr = run_lint_rule(
        rule_runner,
        lint_request_types=request_types,
        targets=targets,
    )
    assert exit_code == FailingRequest.exit_code([bad_address])
    assert stderr == dedent(
        """\

        ✓ BUILD Annually succeeded.
        ✓ Bob The BUILDer succeeded.
        ✕ Conditionally Succeeds Linter failed.
        ✕ Failing Fixer failed.
        ✕ Failing Formatter failed.
        ✕ Failing Linter failed.
        ✓ Files Linter succeeded.
        ✓ Successful Fixer succeeded.
        ✓ Successful Formatter succeeded.
        ✓ Successful Linter succeeded.

        (One or more formatters failed. Run `pants fmt` to fix.)
        (One or more fixers failed. Run `pants fix` to fix.)
        """
    )

    exit_code, stderr = run_lint_rule(
        rule_runner,
        lint_request_types=request_types,
        targets=targets,
        only=[
            FailingRequest.tool_id,
            MockFilesRequest.tool_id,
            FailingFormatter.tool_id,
            FailingFixer.tool_id,
            BuildFileFormatter.tool_id,
            BuildFileFixer.tool_id,
        ],
    )
    assert stderr == dedent(
        """\

        ✓ BUILD Annually succeeded.
        ✓ Bob The BUILDer succeeded.
        ✕ Failing Fixer failed.
        ✕ Failing Formatter failed.
        ✕ Failing Linter failed.
        ✓ Files Linter succeeded.

        (One or more formatters failed. Run `pants fmt` to fix.)
        (One or more fixers failed. Run `pants fix` to fix.)
        """
    )

    exit_code, stderr = run_lint_rule(
        rule_runner,
        lint_request_types=request_types,
        targets=targets,
        skip_formatters=True,
        skip_fixers=True,
    )
    assert stderr == dedent(
        """\

        ✕ Conditionally Succeeds Linter failed.
        ✕ Failing Linter failed.
        ✓ Files Linter succeeded.
        ✓ Successful Linter succeeded.
        """
    )

    exit_code, stderr = run_lint_rule(
        rule_runner,
        lint_request_types=request_types,
        targets=targets,
        skip_fixers=True,
    )
    assert stderr == dedent(
        """\

        ✓ Bob The BUILDer succeeded.
        ✕ Conditionally Succeeds Linter failed.
        ✕ Failing Formatter failed.
        ✕ Failing Linter failed.
        ✓ Files Linter succeeded.
        ✓ Successful Formatter succeeded.
        ✓ Successful Linter succeeded.

        (One or more formatters failed. Run `pants fmt` to fix.)
        """
    )

    exit_code, stderr = run_lint_rule(
        rule_runner,
        lint_request_types=request_types,
        targets=targets,
        skip_formatters=True,
    )
    assert stderr == dedent(
        """\

        ✓ BUILD Annually succeeded.
        ✕ Conditionally Succeeds Linter failed.
        ✕ Failing Fixer failed.
        ✕ Failing Linter failed.
        ✓ Files Linter succeeded.
        ✓ Successful Fixer succeeded.
        ✓ Successful Linter succeeded.

        (One or more fixers failed. Run `pants fix` to fix.)
        """
    )


def test_default_single_partition_partitioner() -> None:
    class KitchenSubsystem(Subsystem):
        options_scope = "kitchen"
        help = "a cookbook might help"
        name = "The Kitchen"
        skip = SkipOption("lint")

    class LintKitchenRequest(LintTargetsRequest):
        field_set_type = MockLinterFieldSet
        tool_subsystem = KitchenSubsystem  # type: ignore[assignment]
        partitioner_type = PartitionerType.DEFAULT_SINGLE_PARTITION

    rules = [
        *LintKitchenRequest._get_rules(),
        QueryRule(Partitions, [LintKitchenRequest.PartitionRequest]),
    ]
    rule_runner = RuleRunner(rules=rules)
    field_sets = (
        MockLinterFieldSet(
            Address("knife"),
            MultipleSourcesField(["knife"], Address("knife")),
            MockRequiredField("present", Address("")),
        ),
        MockLinterFieldSet(
            Address("bowl"),
            MultipleSourcesField(["bowl"], Address("bowl")),
            MockRequiredField("present", Address("")),
        ),
    )
    partitions = rule_runner.request(Partitions, [LintKitchenRequest.PartitionRequest(field_sets)])
    assert len(partitions) == 1
    assert partitions[0].elements == field_sets

    rule_runner.set_options(["--kitchen-skip"])
    partitions = rule_runner.request(Partitions, [LintKitchenRequest.PartitionRequest(field_sets)])
    assert partitions == Partitions([])


@pytest.mark.parametrize("batch_size", [1, 32, 128, 1024])
def test_batched(rule_runner: RuleRunner, batch_size: int) -> None:
    exit_code, stderr = run_lint_rule(
        rule_runner,
        lint_request_types=[
            ConditionallySucceedsRequest,
            FailingRequest,
            SkippedRequest,
            SuccessfulRequest,
        ],
        targets=[make_target(Address("", target_name=f"good{i}")) for i in range(0, 512)],
        batch_size=batch_size,
    )
    assert exit_code == FailingRequest.exit_code([])
    assert stderr == dedent(
        """\

        ✓ Conditionally Succeeds Linter succeeded.
        ✕ Failing Linter failed.
        ✓ Successful Linter succeeded.
        """
    )


def test_streaming_output_success() -> None:
    result = LintResult(0, "stdout", "stderr", linter_name="linter")
    assert result.level() == LogLevel.INFO
    assert result.message() == dedent(
        """\
        linter succeeded.
        stdout
        stderr

        """
    )


def test_streaming_output_failure() -> None:
    result = LintResult(18, "stdout", "stderr", linter_name="linter")
    assert result.level() == LogLevel.ERROR
    assert result.message() == dedent(
        """\
        linter failed (exit code 18).
        stdout
        stderr

        """
    )


def test_streaming_output_partitions() -> None:
    result = LintResult(
        21, "stdout", "stderr", linter_name="linter", partition_description="ghc9.2"
    )
    assert result.level() == LogLevel.ERROR
    assert result.message() == dedent(
        """\
        linter failed (exit code 21).
        Partition: ghc9.2
        stdout
        stderr

        """
    )

1	# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	from abc import ABCMeta, abstractmethod	2✔
7	from collections.abc import Iterable	2✔
8	from dataclasses import dataclass	2✔
9	from pathlib import Path	2✔
10	from textwrap import dedent	2✔
11	from typing import Any, TypeVar	2✔
12
13	import pytest	2✔
14
15	from pants.base.specs import Specs	2✔
16	from pants.core.goals.fix import FixFilesRequest, FixTargetsRequest	2✔
17	from pants.core.goals.fmt import FmtFilesRequest, FmtTargetsRequest	2✔
18	from pants.core.goals.lint import (	2✔
19	AbstractLintRequest,
20	Lint,
21	LintFilesRequest,
22	LintResult,
23	LintSubsystem,
24	LintTargetsRequest,
25	Partitions,
26	)
27	from pants.core.goals.lint_goal import lint	2✔
28	from pants.core.util_rules.distdir import DistDir	2✔
29	from pants.core.util_rules.partitions import PartitionerType, _EmptyMetadata	2✔
30	from pants.engine.addresses import Address	2✔
31	from pants.engine.fs import SpecsPaths, Workspace	2✔
32	from pants.engine.internals.native_engine import EMPTY_SNAPSHOT	2✔
33	from pants.engine.rules import QueryRule	2✔
34	from pants.engine.target import Field, FieldSet, FilteredTargets, MultipleSourcesField, Target	2✔
35	from pants.engine.unions import UnionMembership, UnionRule	2✔
36	from pants.option.option_types import SkipOption	2✔
37	from pants.option.subsystem import Subsystem	2✔
38	from pants.testutil.option_util import create_goal_subsystem	2✔
39	from pants.testutil.rule_runner import RuleRunner, mock_console, run_rule_with_mocks	2✔
40	from pants.util.logging import LogLevel	2✔
41	from pants.util.meta import classproperty	2✔
42
43	_LintRequestT = TypeVar("_LintRequestT", bound=AbstractLintRequest)	2✔
44
45
46	class MockMultipleSourcesField(MultipleSourcesField):	2✔
47	pass	2✔
48
49
50	class MockRequiredField(Field):	2✔
51	alias = "required"	2✔
52	required = True	2✔
53
54
55	class MockTarget(Target):	2✔
56	alias = "mock_target"	2✔
57	core_fields = (MockMultipleSourcesField, MockRequiredField)	2✔
58
59
60	@dataclass(frozen=True)	2✔
61	class MockLinterFieldSet(FieldSet):	2✔
62	required_fields = (MultipleSourcesField,)	2✔
63	sources: MultipleSourcesField	2✔
64	required: MockRequiredField	2✔
65
66
67	class MockLintRequest(AbstractLintRequest, metaclass=ABCMeta):	2✔
68	@staticmethod	2✔
69	@abstractmethod	2✔
70	def exit_code(_: Iterable[Address]) -> int:	2✔
71	pass	×
72
73	@classmethod	2✔
74	@abstractmethod	2✔
75	def get_lint_result(cls, elements: Iterable) -> LintResult:	2✔
76	pass	×
77
78
79	class MockLintTargetsRequest(MockLintRequest, LintTargetsRequest):	2✔
80	field_set_type = MockLinterFieldSet	2✔
81
82	@classmethod	2✔
83	def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:	2✔
84	addresses = [field_set.address for field_set in field_sets]	2✔
85	return LintResult(cls.exit_code(addresses), "", "", cls.tool_name)	2✔
86
87
88	class SuccessfulRequest(MockLintTargetsRequest):	2✔
89	@classproperty	2✔
90	def tool_name(cls) -> str:	2✔
91	return "Successful Linter"	2✔
92
93	@classproperty	2✔
94	def tool_id(cls) -> str:	2✔
95	return "successfullinter"	2✔
96
97	@staticmethod	2✔
98	def exit_code(_: Iterable[Address]) -> int:	2✔
99	return 0	2✔
100
101
102	class FailingRequest(MockLintTargetsRequest):	2✔
103	@classproperty	2✔
104	def tool_name(cls) -> str:	2✔
105	return "Failing Linter"	2✔
106
107	@classproperty	2✔
108	def tool_id(cls) -> str:	2✔
109	return "failinglinter"	2✔
110
111	@staticmethod	2✔
112	def exit_code(_: Iterable[Address]) -> int:	2✔
113	return 1	2✔
114
115
116	class ConditionallySucceedsRequest(MockLintTargetsRequest):	2✔
117	@classproperty	2✔
118	def tool_name(cls) -> str:	2✔
119	return "Conditionally Succeeds Linter"	2✔
120
121	@classproperty	2✔
122	def tool_id(cls) -> str:	2✔
123	return "conditionallysucceedslinter"	2✔
124
125	@staticmethod	2✔
126	def exit_code(addresses: Iterable[Address]) -> int:	2✔
127	if any(address.target_name == "bad" for address in addresses):	2✔
128	return 127	2✔
129	return 0	2✔
130
131
132	class SkippedRequest(MockLintTargetsRequest):	2✔
133	@classproperty	2✔
134	def tool_name(cls) -> str:	2✔
135	return "Skipped Linter"	×
136
137	@classproperty	2✔
138	def tool_id(cls) -> str:	2✔
139	return "skippedlinter"	2✔
140
141	@staticmethod	2✔
142	def exit_code(_) -> int:	2✔
143	return 0	×
144
145
146	class InvalidField(MultipleSourcesField):	2✔
147	pass	2✔
148
149
150	class InvalidFieldSet(MockLinterFieldSet):	2✔
151	required_fields = (InvalidField,)	2✔
152
153
154	class InvalidRequest(MockLintTargetsRequest):	2✔
155	field_set_type = InvalidFieldSet	2✔
156
157	@classproperty	2✔
158	def tool_name(cls) -> str:	2✔
159	return "Invalid Linter"	×
160
161	@classproperty	2✔
162	def tool_id(cls) -> str:	2✔
163	return "invalidlinter"	2✔
164
165	@staticmethod	2✔
166	def exit_code(_: Iterable[Address]) -> int:	2✔
167	return -1	×
168
169
170	def _all_lint_requests() -> Iterable[type[MockLintRequest]]:	2✔
171	classes = [MockLintRequest]	2✔
172	while classes:	2✔
173	cls = classes.pop()	2✔
174	subclasses = cls.__subclasses__()	2✔
175	classes.extend(subclasses)	2✔
176	yield from subclasses	2✔
177
178
179	def mock_target_partitioner(__implicitly: tuple) -> Partitions[MockLinterFieldSet, Any]:	2✔
180	request, typ = next(iter(__implicitly[0].items()))	2✔
181	assert typ == LintTargetsRequest.PartitionRequest	2✔
182	if type(request) is SkippedRequest.PartitionRequest:	2✔
183	return Partitions()	2✔
184
185	operates_on_paths = {	2✔
186	getattr(cls, "PartitionRequest"): cls._requires_snapshot for cls in _all_lint_requests()
187	}[type(request)]
188	if operates_on_paths:	2✔
189	return Partitions.single_partition(fs.sources.globs for fs in request.field_sets)	2✔
190
191	return Partitions.single_partition(request.field_sets)	2✔
192
193
194	class MockFilesRequest(MockLintRequest, LintFilesRequest):	2✔
195	@classproperty	2✔
196	def tool_name(cls) -> str:	2✔
197	return "Files Linter"	2✔
198
199	@classproperty	2✔
200	def tool_id(cls) -> str:	2✔
201	return "fileslinter"	2✔
202
203	@classmethod	2✔
204	def get_lint_result(cls, files: Iterable[str]) -> LintResult:	2✔
205	return LintResult(0, "", "", cls.tool_name)	2✔
206
207
208	def mock_file_partitioner(__implicitly: dict) -> Partitions[str, Any]:	2✔
209	request, typ = next(iter(__implicitly[0].items()))	2✔
210	assert typ == LintFilesRequest.PartitionRequest	2✔
211	return Partitions.single_partition(request.files)	2✔
212
213
214	def mock_lint_partition(__implicitly: dict) -> LintResult:	2✔
215	request, typ = next(iter(__implicitly[0].items()))	2✔
216	assert typ == AbstractLintRequest.Batch	2✔
217	request_type = {cls.Batch: cls for cls in _all_lint_requests()}[type(request)]	2✔
218	return request_type.get_lint_result(request.elements)	2✔
219
220
221	class MockFmtRequest(MockLintRequest, FmtTargetsRequest):	2✔
222	field_set_type = MockLinterFieldSet	2✔
223
224
225	class SuccessfulFormatter(MockFmtRequest):	2✔
226	@classproperty	2✔
227	def tool_name(cls) -> str:	2✔
228	return "Successful Formatter"	2✔
229
230	@classproperty	2✔
231	def tool_id(cls) -> str:	2✔
232	return "successfulformatter"	2✔
233
234	@classmethod	2✔
235	def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:	2✔
236	return LintResult(0, "", "", cls.tool_name)	2✔
237
238
239	class FailingFormatter(MockFmtRequest):	2✔
240	@classproperty	2✔
241	def tool_name(cls) -> str:	2✔
242	return "Failing Formatter"	2✔
243
244	@classproperty	2✔
245	def tool_id(cls) -> str:	2✔
246	return "failingformatter"	2✔
247
248	@classmethod	2✔
249	def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:	2✔
250	return LintResult(1, "", "", cls.tool_name)	2✔
251
252
253	class BuildFileFormatter(MockLintRequest, FmtFilesRequest):	2✔
254	@classproperty	2✔
255	def tool_name(cls) -> str:	2✔
256	return "Bob The BUILDer"	2✔
257
258	@classproperty	2✔
259	def tool_id(cls) -> str:	2✔
260	return "bob"	2✔
261
262	@classmethod	2✔
263	def get_lint_result(cls, files: Iterable[str]) -> LintResult:	2✔
264	return LintResult(0, "", "", cls.tool_name)	2✔
265
266
267	class MockFixRequest(MockLintRequest, FixTargetsRequest):	2✔
268	field_set_type = MockLinterFieldSet	2✔
269
270
271	class SuccessfulFixer(MockFixRequest):	2✔
272	@classproperty	2✔
273	def tool_name(cls) -> str:	2✔
274	return "Successful Fixer"	2✔
275
276	@classproperty	2✔
277	def tool_id(cls) -> str:	2✔
278	return "successfulfixer"	2✔
279
280	@classmethod	2✔
281	def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:	2✔
282	return LintResult(0, "", "", cls.tool_name)	2✔
283
284
285	class FailingFixer(MockFixRequest):	2✔
286	@classproperty	2✔
287	def tool_name(cls) -> str:	2✔
288	return "Failing Fixer"	2✔
289
290	@classproperty	2✔
291	def tool_id(cls) -> str:	2✔
292	return "failingfixer"	2✔
293
294	@classmethod	2✔
295	def get_lint_result(cls, field_sets: Iterable[MockLinterFieldSet]) -> LintResult:	2✔
296	return LintResult(1, "", "", cls.tool_name)	2✔
297
298
299	class BuildFileFixer(MockLintRequest, FixFilesRequest):	2✔
300	@classproperty	2✔
301	def tool_name(cls) -> str:	2✔
302	return "BUILD Annually"	2✔
303
304	@classproperty	2✔
305	def tool_id(cls) -> str:	2✔
306	return "buildannually"	2✔
307
308	@classmethod	2✔
309	def get_lint_result(cls, files: Iterable[str]) -> LintResult:	2✔
310	return LintResult(0, "", "", cls.tool_name)	2✔
311
312
313	def mock_fix_partition_as_linter(request) -> LintResult:	2✔
314	request_type = {cls.Batch: cls for cls in _all_lint_requests()}[type(request)]	2✔
315	return request_type.get_lint_result(request.elements)	2✔
316
317
318	@pytest.fixture	2✔
319	def rule_runner() -> RuleRunner:	2✔
320	return RuleRunner()	2✔
321
322
323	def make_target(address: Address \| None = None) -> Target:	2✔
324	return MockTarget(	2✔
325	{MockRequiredField.alias: "present"}, address or Address("", target_name="tests")
326	)
327
328
329	def run_lint_rule(	2✔
330	rule_runner: RuleRunner,
331	*,
332	lint_request_types: Iterable[type[_LintRequestT]],
333	targets: list[Target],
334	batch_size: int = 128,
335	only: list[str] \| None = None,
336	skip_formatters: bool = False,
337	skip_fixers: bool = False,
338	) -> tuple[int, str]:
339	union_membership = UnionMembership.from_rules(	2✔
340	{
341	*[UnionRule(AbstractLintRequest, t) for t in lint_request_types],
342	*[UnionRule(AbstractLintRequest.Batch, rt.Batch) for rt in lint_request_types],
343	*[
344	UnionRule(LintTargetsRequest.PartitionRequest, rt.PartitionRequest)
345	for rt in lint_request_types
346	if issubclass(rt, LintTargetsRequest)
347	],
348	*[
349	UnionRule(LintFilesRequest.PartitionRequest, rt.PartitionRequest)
350	for rt in lint_request_types
351	if issubclass(rt, LintFilesRequest)
352	],
353	}
354	)
355	lint_subsystem = create_goal_subsystem(	2✔
356	LintSubsystem,
357	batch_size=batch_size,
358	only=only or [],
359	skip_formatters=skip_formatters,
360	skip_fixers=skip_fixers,
361	)
362
363	with mock_console(rule_runner.options_bootstrapper) as (console, stdio_reader):	2✔
364	result: Lint = run_rule_with_mocks(	2✔
365	lint,
366	rule_args=[
367	console,
368	Workspace(rule_runner.scheduler, _enforce_effects=False),
369	Specs.empty(),
370	lint_subsystem,
371	union_membership,
372	DistDir(relpath=Path("dist")),
373	],
374	mock_calls={
375	"pants.engine.internals.graph.filter_targets": lambda __implicitly: FilteredTargets(
376	tuple(targets)
377	),
378	"pants.engine.internals.specs_rules.resolve_specs_paths": lambda _: SpecsPaths(
379	("f.txt", "BUILD"), ()
380	),
381	"pants.core.goals.lint.partition_targets": mock_target_partitioner,
382	"pants.core.goals.lint.partition_files": mock_file_partitioner,
383	"pants.core.goals.lint.lint_batch": mock_lint_partition,
384	"pants.core.goals.lint_goal.run_fixer_or_formatter_as_linter": mock_fix_partition_as_linter,
385	"pants.engine.intrinsics.digest_to_snapshot": lambda __implicitly: EMPTY_SNAPSHOT,
386	},
387	union_membership=union_membership,
388	# We don't want temporary warnings to interfere with our expected output.
389	show_warnings=False,
390	)
391	assert not stdio_reader.get_stdout()	2✔
392	return result.exit_code, stdio_reader.get_stderr()	2✔
393
394
395	def test_duplicate_files_in_fixer(rule_runner: RuleRunner) -> None:	2✔
396	assert MockFixRequest.Batch(	2✔
397	"tool_name",
398	("a", "a"),
399	_EmptyMetadata(),
400	rule_runner.make_snapshot({"a": ""}),
401	).files == ("a",)
402
403
404	def test_invalid_target_noops(rule_runner: RuleRunner) -> None:	2✔
405	exit_code, stderr = run_lint_rule(	2✔
406	rule_runner, lint_request_types=[InvalidRequest], targets=[make_target()]
407	)
408	assert exit_code == 0	2✔
409	assert stderr == ""	2✔
410
411
412	def test_summary(rule_runner: RuleRunner) -> None:	2✔
413	"""Test that we render the summary correctly.
414
415	This tests that we:
416	* Merge multiple results belonging to the same linter (`--per-file-caching`).
417	* Decide correctly between skipped, failed, and succeeded.
418	"""
419	good_address = Address("", target_name="good")	2✔
420	bad_address = Address("", target_name="bad")	2✔
421
422	request_types = [	2✔
423	ConditionallySucceedsRequest,
424	FailingRequest,
425	SkippedRequest,
426	SuccessfulRequest,
427	SuccessfulFormatter,
428	FailingFormatter,
429	BuildFileFormatter,
430	SuccessfulFixer,
431	FailingFixer,
432	BuildFileFixer,
433	MockFilesRequest,
434	]
435	targets = [make_target(good_address), make_target(bad_address)]	2✔
436
437	exit_code, stderr = run_lint_rule(	2✔
438	rule_runner,
439	lint_request_types=request_types,
440	targets=targets,
441	)
442	assert exit_code == FailingRequest.exit_code([bad_address])	2✔
443	assert stderr == dedent(	1✔
444	"""\
445
446	✓ BUILD Annually succeeded.
447	✓ Bob The BUILDer succeeded.
448	✕ Conditionally Succeeds Linter failed.
449	✕ Failing Fixer failed.
450	✕ Failing Formatter failed.
451	✕ Failing Linter failed.
452	✓ Files Linter succeeded.
453	✓ Successful Fixer succeeded.
454	✓ Successful Formatter succeeded.
455	✓ Successful Linter succeeded.
456
457	(One or more formatters failed. Run `pants fmt` to fix.)
458	(One or more fixers failed. Run `pants fix` to fix.)
459	"""
460	)
461
462	exit_code, stderr = run_lint_rule(	1✔
463	rule_runner,
464	lint_request_types=request_types,
465	targets=targets,
466	only=[
467	FailingRequest.tool_id,
468	MockFilesRequest.tool_id,
469	FailingFormatter.tool_id,
470	FailingFixer.tool_id,
471	BuildFileFormatter.tool_id,
472	BuildFileFixer.tool_id,
473	],
474	)
475	assert stderr == dedent(	1✔
476	"""\
477
478	✓ BUILD Annually succeeded.
479	✓ Bob The BUILDer succeeded.
480	✕ Failing Fixer failed.
481	✕ Failing Formatter failed.
482	✕ Failing Linter failed.
483	✓ Files Linter succeeded.
484
485	(One or more formatters failed. Run `pants fmt` to fix.)
486	(One or more fixers failed. Run `pants fix` to fix.)
487	"""
488	)
489
490	exit_code, stderr = run_lint_rule(	1✔
491	rule_runner,
492	lint_request_types=request_types,
493	targets=targets,
494	skip_formatters=True,
495	skip_fixers=True,
496	)
497	assert stderr == dedent(	1✔
498	"""\
499
500	✕ Conditionally Succeeds Linter failed.
501	✕ Failing Linter failed.
502	✓ Files Linter succeeded.
503	✓ Successful Linter succeeded.
504	"""
505	)
506
507	exit_code, stderr = run_lint_rule(	1✔
508	rule_runner,
509	lint_request_types=request_types,
510	targets=targets,
511	skip_fixers=True,
512	)
513	assert stderr == dedent(	1✔
514	"""\
515
516	✓ Bob The BUILDer succeeded.
517	✕ Conditionally Succeeds Linter failed.
518	✕ Failing Formatter failed.
519	✕ Failing Linter failed.
520	✓ Files Linter succeeded.
521	✓ Successful Formatter succeeded.
522	✓ Successful Linter succeeded.
523
524	(One or more formatters failed. Run `pants fmt` to fix.)
525	"""
526	)
527
528	exit_code, stderr = run_lint_rule(	1✔
529	rule_runner,
530	lint_request_types=request_types,
531	targets=targets,
532	skip_formatters=True,
533	)
534	assert stderr == dedent(	1✔
535	"""\
536
537	✓ BUILD Annually succeeded.
538	✕ Conditionally Succeeds Linter failed.
539	✕ Failing Fixer failed.
540	✕ Failing Linter failed.
541	✓ Files Linter succeeded.
542	✓ Successful Fixer succeeded.
543	✓ Successful Linter succeeded.
544
545	(One or more fixers failed. Run `pants fix` to fix.)
546	"""
547	)
548
549
550	def test_default_single_partition_partitioner() -> None:	2✔
551	class KitchenSubsystem(Subsystem):	2✔
552	options_scope = "kitchen"	2✔
553	help = "a cookbook might help"	2✔
554	name = "The Kitchen"	2✔
555	skip = SkipOption("lint")	2✔
556
557	class LintKitchenRequest(LintTargetsRequest):	2✔
558	field_set_type = MockLinterFieldSet	2✔
559	tool_subsystem = KitchenSubsystem # type: ignore[assignment]	2✔
560	partitioner_type = PartitionerType.DEFAULT_SINGLE_PARTITION	2✔
561
562	rules = [	2✔
563	*LintKitchenRequest._get_rules(),
564	QueryRule(Partitions, [LintKitchenRequest.PartitionRequest]),
565	]
566	rule_runner = RuleRunner(rules=rules)	2✔
567	field_sets = (	2✔
568	MockLinterFieldSet(
569	Address("knife"),
570	MultipleSourcesField(["knife"], Address("knife")),
571	MockRequiredField("present", Address("")),
572	),
573	MockLinterFieldSet(
574	Address("bowl"),
575	MultipleSourcesField(["bowl"], Address("bowl")),
576	MockRequiredField("present", Address("")),
577	),
578	)
579	partitions = rule_runner.request(Partitions, [LintKitchenRequest.PartitionRequest(field_sets)])	2✔
580	assert len(partitions) == 1	2✔
581	assert partitions[0].elements == field_sets	2✔
582
583	rule_runner.set_options(["--kitchen-skip"])	2✔
584	partitions = rule_runner.request(Partitions, [LintKitchenRequest.PartitionRequest(field_sets)])	2✔
585	assert partitions == Partitions([])	2✔
586
587
588	@pytest.mark.parametrize("batch_size", [1, 32, 128, 1024])	2✔
589	def test_batched(rule_runner: RuleRunner, batch_size: int) -> None:	2✔
590	exit_code, stderr = run_lint_rule(	2✔
591	rule_runner,
592	lint_request_types=[
593	ConditionallySucceedsRequest,
594	FailingRequest,
595	SkippedRequest,
596	SuccessfulRequest,
597	],
598	targets=[make_target(Address("", target_name=f"good{i}")) for i in range(0, 512)],
599	batch_size=batch_size,
600	)
601	assert exit_code == FailingRequest.exit_code([])	2✔
602	assert stderr == dedent(	2✔
603	"""\
604
605	✓ Conditionally Succeeds Linter succeeded.
606	✕ Failing Linter failed.
607	✓ Successful Linter succeeded.
608	"""
609	)
610
611
612	def test_streaming_output_success() -> None:	2✔
613	result = LintResult(0, "stdout", "stderr", linter_name="linter")	2✔
614	assert result.level() == LogLevel.INFO	2✔
615	assert result.message() == dedent(	2✔
616	"""\
617	linter succeeded.
618	stdout
619	stderr
620
621	"""
622	)
623
624
625	def test_streaming_output_failure() -> None:	2✔
626	result = LintResult(18, "stdout", "stderr", linter_name="linter")	2✔
627	assert result.level() == LogLevel.ERROR	2✔
628	assert result.message() == dedent(	2✔
629	"""\
630	linter failed (exit code 18).
631	stdout
632	stderr
633
634	"""
635	)
636
637
638	def test_streaming_output_partitions() -> None:	2✔
639	result = LintResult(	2✔
640	21, "stdout", "stderr", linter_name="linter", partition_description="ghc9.2"
641	)
642	assert result.level() == LogLevel.ERROR	2✔
643	assert result.message() == dedent(	2✔
644	"""\
645	linter failed (exit code 21).
646	Partition: ghc9.2
647	stdout
648	stderr
649
650	"""
651	)

pantsbuild / pants / 19641045249

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous