18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

79.55

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

pantsbuild / pants / 18517631058

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