19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

64.77

/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	1✔
4
5	import logging	1✔
6	from abc import ABC, abstractmethod	1✔
7	from dataclasses import dataclass	1✔
8	from enum import Enum	1✔
9	from typing import Protocol	1✔
10
11	from pants.engine.addresses import Address	1✔
12	from pants.engine.internals.target_adaptor import TargetAdaptor	1✔
13	from pants.engine.rules import collect_rules, implicitly, rule	1✔
14	from pants.engine.unions import UnionMembership, union	1✔
15
16	logger = logging.getLogger(__name__)	1✔
17
18
19	class DependencyRulesError(Exception):	1✔
20	pass	1✔
21
22
23	class DependencyRuleActionDeniedError(DependencyRulesError):	1✔
24	@classmethod	1✔
25	def create(cls, description_of_origin: str) -> DependencyRuleActionDeniedError:	1✔
UNCOV 26	return cls(f"Dependency rule violation for {description_of_origin}")	×
27
28
29	class DependencyRuleAction(Enum):	1✔
30	ALLOW = "allow"	1✔
31	DENY = "deny"	1✔
32	WARN = "warn"	1✔
33
34	def execute(	1✔
35	self, *, description_of_origin: str, return_exc: bool = False
36	) -> DependencyRuleActionDeniedError \| None:
UNCOV 37	if self is DependencyRuleAction.ALLOW:	×
UNCOV 38	return None	×
UNCOV 39	err = DependencyRuleActionDeniedError.create(description_of_origin)	×
UNCOV 40	if self is DependencyRuleAction.DENY:	×
UNCOV 41	if return_exc:	×
UNCOV 42	return err	×
43	else:
UNCOV 44	raise err	×
UNCOV 45	if self is DependencyRuleAction.WARN:	×
UNCOV 46	logger.warning(str(err))	×
47	else:
48	raise NotImplementedError(f"{type(self).__name__}.execute() not implemented for {self}")	×
UNCOV 49	return None	×
50
51	def __str__(self) -> str:	1✔
UNCOV 52	return self.name	×
53
54
55	@dataclass(frozen=True)	1✔
56	class DependencyRuleApplication:	1✔
57	action: DependencyRuleAction	1✔
58	rule_description: str	1✔
59	origin_address: Address	1✔
60	origin_type: str	1✔
61	dependency_address: Address	1✔
62	dependency_type: str	1✔
63
64	def execute(self) -> str \| None:	1✔
UNCOV 65	err = self.action.execute(	×
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:	×
UNCOV 72	return None	×
73	else:
UNCOV 74	return str(self)	×
75
76	def __str__(self) -> str:	1✔
UNCOV 77	return (	×
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):	1✔
84	def peek(self) -> tuple[str, ...]:	1✔
85	"""Return a list of all rules in rule set."""
86
87
88	class BuildFileDependencyRules(ABC):	1✔
89	@staticmethod
90	@abstractmethod
91	def create_parser_state(
92	path: str, parent: BuildFileDependencyRules \| None
93	) -> BuildFileDependencyRulesParserState: ...
94
95	@staticmethod	1✔
96	@abstractmethod	1✔
97	def check_dependency_rules(	1✔
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):	1✔
120	@abstractmethod	1✔
121	def get_frozen_dependency_rules(self) -> BuildFileDependencyRules \| None:	1✔
122	pass	×
123
124	@abstractmethod	1✔
125	def set_dependency_rules(	1✔
126	self,
127	build_file: str,
128	*args,
129	**kwargs,
130	) -> None:
131	pass	×
132
133
134	@union	1✔
135	class BuildFileDependencyRulesImplementationRequest:	1✔
136	pass	1✔
137
138
139	@dataclass(frozen=True)	1✔
140	class BuildFileDependencyRulesImplementation:	1✔
141	build_file_dependency_rules_class: type[BuildFileDependencyRules]	1✔
142
143
144	@rule(polymorphic=True)	1✔
145	async def _get_build_file_dependency_rules_implementation(	1✔
146	req: BuildFileDependencyRulesImplementationRequest,
147	) -> BuildFileDependencyRulesImplementation:
148	raise NotImplementedError()	×
149
150
151	@dataclass(frozen=True)	1✔
152	class MaybeBuildFileDependencyRulesImplementation:	1✔
153	build_file_dependency_rules_class: type[BuildFileDependencyRules] \| None	1✔
154
155
156	@rule	1✔
157	async def get_build_file_dependency_rules_implementation(	1✔
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():	1✔
UNCOV 176	return collect_rules()	×

pantsbuild / pants / 19015773527

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