26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

88.64

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

pantsbuild / pants / 26342152999

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