18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

65.91

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

pantsbuild / pants / 18252174847

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