21552830208

Committed 31 Jan 2026 11:40PM UTC coverage: 80.277% (-0.05%) from 80.324%

Build # 21552830208

Build Type

Pull #23062

github

Committed by

web-flow

Commit Message

Merge 808a9786c into 2c4dcf9cf

Pull Request Pull Request #23062: Remove support for Get

Run Details

18 of 25 new or added lines in 4 files covered. (72.0%)

17119 existing lines in 541 files now uncovered.

78278 of 97510 relevant lines covered (80.28%)

3.36 hits per line

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

85.23

/src/python/pants/engine/internals/dep_rules.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Protocol

from pants.engine.addresses import Address
from pants.engine.internals.target_adaptor import TargetAdaptor
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionMembership, union

logger = logging.getLogger(__name__)


class DependencyRulesError(Exception):
    pass


class DependencyRuleActionDeniedError(DependencyRulesError):
    @classmethod
    def create(cls, description_of_origin: str) -> DependencyRuleActionDeniedError:
        return cls(f"Dependency rule violation for {description_of_origin}")


class DependencyRuleAction(Enum):
    ALLOW = "allow"
    DENY = "deny"
    WARN = "warn"

    def execute(
        self, *, description_of_origin: str, return_exc: bool = False
    ) -> DependencyRuleActionDeniedError | None:
        if self is DependencyRuleAction.ALLOW:
            return None
        err = DependencyRuleActionDeniedError.create(description_of_origin)
        if self is DependencyRuleAction.DENY:
            if return_exc:
                return err
            else:
                raise err
        if self is DependencyRuleAction.WARN:
            logger.warning(str(err))
        else:
            raise NotImplementedError(f"{type(self).__name__}.execute() not implemented for {self}")
        return None

    def __str__(self) -> str:
        return self.name


@dataclass(frozen=True)
class DependencyRuleApplication:
    action: DependencyRuleAction
    rule_description: str
    origin_address: Address
    origin_type: str
    dependency_address: Address
    dependency_type: str

    def execute(self) -> str | None:
        err = self.action.execute(
            description_of_origin=(
                f"{self.origin_address}'s dependency on {self.dependency_address}"
            ),
            return_exc=True,
        )
        if err is None:
            return None
        else:
            return str(self)

    def __str__(self) -> str:
        return (
            f"{self.rule_description} : {self.action.name}\n{self.origin_type} "
            f"{self.origin_address} -> {self.dependency_type} {self.dependency_address}"
        )


class DependencyRuleSet(Protocol):
    def peek(self) -> tuple[str, ...]:
        """Return a list of all rules in rule set."""


class BuildFileDependencyRules(ABC):
    @staticmethod
    @abstractmethod
    def create_parser_state(
        path: str, parent: BuildFileDependencyRules | None
    ) -> BuildFileDependencyRulesParserState: ...

    @staticmethod
    @abstractmethod
    def check_dependency_rules(
        *,
        origin_address: Address,
        origin_adaptor: TargetAdaptor,
        dependencies_rules: BuildFileDependencyRules | None,
        dependency_address: Address,
        dependency_adaptor: TargetAdaptor,
        dependents_rules: BuildFileDependencyRules | None,
    ) -> DependencyRuleApplication:
        """Check all rules for any that apply to the relation between the two targets.

        The `__dependencies_rules__` are the rules applicable for the origin target.
        The `__dependents_rules__` are the rules applicable for the dependency target.

        Return dependency rule application describing the resulting action to take: ALLOW, DENY or
        WARN. WARN is effectively the same as ALLOW, but with a logged warning.
        """

    @abstractmethod
    def get_ruleset(self, address: Address, target: TargetAdaptor) -> DependencyRuleSet | None: ...


class BuildFileDependencyRulesParserState(ABC):
    @abstractmethod
    def get_frozen_dependency_rules(self) -> BuildFileDependencyRules | None:
        pass

    @abstractmethod
    def set_dependency_rules(
        self,
        build_file: str,
        *args,
        **kwargs,
    ) -> None:
        pass


@union
class BuildFileDependencyRulesImplementationRequest:
    pass


@dataclass(frozen=True)
class BuildFileDependencyRulesImplementation:
    build_file_dependency_rules_class: type[BuildFileDependencyRules]


@rule(polymorphic=True)
async def _get_build_file_dependency_rules_implementation(
    req: BuildFileDependencyRulesImplementationRequest,
) -> BuildFileDependencyRulesImplementation:
    raise NotImplementedError()


@dataclass(frozen=True)
class MaybeBuildFileDependencyRulesImplementation:
    build_file_dependency_rules_class: type[BuildFileDependencyRules] | None


@rule
async def get_build_file_dependency_rules_implementation(
    union_membership: UnionMembership,
) -> MaybeBuildFileDependencyRulesImplementation:
    request_types = union_membership.get(BuildFileDependencyRulesImplementationRequest)
    if len(request_types) > 1:
        impls = ", ".join(map(str, request_types))
        raise AssertionError(
            f"There must be at most one BUILD file dependency rules implementation, got: {impls}"
        )
    if len(request_types) == 1:
        request_type = next(iter(request_types))
        impl = await _get_build_file_dependency_rules_implementation(
            **implicitly({request_type(): BuildFileDependencyRulesImplementationRequest})
        )
        return MaybeBuildFileDependencyRulesImplementation(impl.build_file_dependency_rules_class)
    return MaybeBuildFileDependencyRulesImplementation(None)


def rules():
    return collect_rules()

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	12✔
4
5	import logging	12✔
6	from abc import ABC, abstractmethod	12✔
7	from dataclasses import dataclass	12✔
8	from enum import Enum	12✔
9	from typing import Protocol	12✔
10
11	from pants.engine.addresses import Address	12✔
12	from pants.engine.internals.target_adaptor import TargetAdaptor	12✔
13	from pants.engine.rules import collect_rules, implicitly, rule	12✔
14	from pants.engine.unions import UnionMembership, union	12✔
15
16	logger = logging.getLogger(__name__)	12✔
17
18
19	class DependencyRulesError(Exception):	12✔
20	pass	12✔
21
22
23	class DependencyRuleActionDeniedError(DependencyRulesError):	12✔
24	@classmethod	12✔
25	def create(cls, description_of_origin: str) -> DependencyRuleActionDeniedError:	12✔
UNCOV 26	return cls(f"Dependency rule violation for {description_of_origin}")	2✔
27
28
29	class DependencyRuleAction(Enum):	12✔
30	ALLOW = "allow"	12✔
31	DENY = "deny"	12✔
32	WARN = "warn"	12✔
33
34	def execute(	12✔
35	self, *, description_of_origin: str, return_exc: bool = False
36	) -> DependencyRuleActionDeniedError \| None:
UNCOV 37	if self is DependencyRuleAction.ALLOW:	2✔
UNCOV 38	return None	2✔
UNCOV 39	err = DependencyRuleActionDeniedError.create(description_of_origin)	2✔
UNCOV 40	if self is DependencyRuleAction.DENY:	2✔
UNCOV 41	if return_exc:	2✔
UNCOV 42	return err	1✔
43	else:
UNCOV 44	raise err	1✔
UNCOV 45	if self is DependencyRuleAction.WARN:	1✔
UNCOV 46	logger.warning(str(err))	1✔
47	else:
48	raise NotImplementedError(f"{type(self).__name__}.execute() not implemented for {self}")	×
UNCOV 49	return None	1✔
50
51	def __str__(self) -> str:	12✔
UNCOV 52	return self.name	1✔
53
54
55	@dataclass(frozen=True)	12✔
56	class DependencyRuleApplication:	12✔
57	action: DependencyRuleAction	12✔
58	rule_description: str	12✔
59	origin_address: Address	12✔
60	origin_type: str	12✔
61	dependency_address: Address	12✔
62	dependency_type: str	12✔
63
64	def execute(self) -> str \| None:	12✔
UNCOV 65	err = self.action.execute(	1✔
66	description_of_origin=(
67	f"{self.origin_address}'s dependency on {self.dependency_address}"
68	),
69	return_exc=True,
70	)
UNCOV 71	if err is None:	1✔
UNCOV 72	return None	1✔
73	else:
UNCOV 74	return str(self)	1✔
75
76	def __str__(self) -> str:	12✔
UNCOV 77	return (	1✔
78	f"{self.rule_description} : {self.action.name}\n{self.origin_type} "
79	f"{self.origin_address} -> {self.dependency_type} {self.dependency_address}"
80	)
81
82
83	class DependencyRuleSet(Protocol):	12✔
84	def peek(self) -> tuple[str, ...]:	12✔
85	"""Return a list of all rules in rule set."""
86
87
88	class BuildFileDependencyRules(ABC):	12✔
89	@staticmethod
90	@abstractmethod
91	def create_parser_state(
92	path: str, parent: BuildFileDependencyRules \| None
93	) -> BuildFileDependencyRulesParserState: ...
94
95	@staticmethod	12✔
96	@abstractmethod	12✔
97	def check_dependency_rules(	12✔
98	*,
99	origin_address: Address,
100	origin_adaptor: TargetAdaptor,
101	dependencies_rules: BuildFileDependencyRules \| None,
102	dependency_address: Address,
103	dependency_adaptor: TargetAdaptor,
104	dependents_rules: BuildFileDependencyRules \| None,
105	) -> DependencyRuleApplication:
106	"""Check all rules for any that apply to the relation between the two targets.
107
108	The `__dependencies_rules__` are the rules applicable for the origin target.
109	The `__dependents_rules__` are the rules applicable for the dependency target.
110
111	Return dependency rule application describing the resulting action to take: ALLOW, DENY or
112	WARN. WARN is effectively the same as ALLOW, but with a logged warning.
113	"""
114
115	@abstractmethod
116	def get_ruleset(self, address: Address, target: TargetAdaptor) -> DependencyRuleSet \| None: ...
117
118
119	class BuildFileDependencyRulesParserState(ABC):	12✔
120	@abstractmethod	12✔
121	def get_frozen_dependency_rules(self) -> BuildFileDependencyRules \| None:	12✔
122	pass	×
123
124	@abstractmethod	12✔
125	def set_dependency_rules(	12✔
126	self,
127	build_file: str,
128	*args,
129	**kwargs,
130	) -> None:
131	pass	×
132
133
134	@union	12✔
135	class BuildFileDependencyRulesImplementationRequest:	12✔
136	pass	12✔
137
138
139	@dataclass(frozen=True)	12✔
140	class BuildFileDependencyRulesImplementation:	12✔
141	build_file_dependency_rules_class: type[BuildFileDependencyRules]	12✔
142
143
144	@rule(polymorphic=True)	12✔
145	async def _get_build_file_dependency_rules_implementation(	12✔
146	req: BuildFileDependencyRulesImplementationRequest,
147	) -> BuildFileDependencyRulesImplementation:
148	raise NotImplementedError()	×
149
150
151	@dataclass(frozen=True)	12✔
152	class MaybeBuildFileDependencyRulesImplementation:	12✔
153	build_file_dependency_rules_class: type[BuildFileDependencyRules] \| None	12✔
154
155
156	@rule	12✔
157	async def get_build_file_dependency_rules_implementation(	12✔
158	union_membership: UnionMembership,
159	) -> MaybeBuildFileDependencyRulesImplementation:
160	request_types = union_membership.get(BuildFileDependencyRulesImplementationRequest)	×
161	if len(request_types) > 1:	×
162	impls = ", ".join(map(str, request_types))	×
163	raise AssertionError(	×
164	f"There must be at most one BUILD file dependency rules implementation, got: {impls}"
165	)
166	if len(request_types) == 1:	×
167	request_type = next(iter(request_types))	×
168	impl = await _get_build_file_dependency_rules_implementation(	×
169	**implicitly({request_type(): BuildFileDependencyRulesImplementationRequest})
170	)
171	return MaybeBuildFileDependencyRulesImplementation(impl.build_file_dependency_rules_class)	×
172	return MaybeBuildFileDependencyRulesImplementation(None)	×
173
174
175	def rules():	12✔
176	return collect_rules()	12✔

pantsbuild / pants / 21552830208

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