23177125175

Committed 17 Mar 2026 03:32AM UTC coverage: 52.677% (-40.3%) from 92.932%

Build # 23177125175

Build Type

Pull #23177

github

Committed by

web-flow

Commit Message

Merge 1824dfbf4 into 0b9fdfb0e

Pull Request Pull Request #23177: Bump the gha-deps group across 1 directory with 4 updates

Coverage Stats

31687 of 60153 relevant lines covered (52.68%)

1.05 hits per line

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

0.0

/src/python/pants/backend/build_files/fix/deprecations/renamed_fields_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 tokenize
from collections import defaultdict
from collections.abc import Mapping
from dataclasses import dataclass
from typing import DefaultDict

from pants.backend.build_files.fix.base import FixBuildFilesRequest
from pants.backend.build_files.fix.deprecations.base import FixBUILDFileRequest, FixedBUILDFile
from pants.backend.build_files.fix.deprecations.subsystem import BUILDDeprecationsFixer
from pants.core.goals.fix import FixResult
from pants.engine.fs import CreateDigest, FileContent
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import RegisteredTargetTypes, TargetGenerator
from pants.engine.unions import UnionMembership
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel


class RenameFieldsInFilesRequest(FixBuildFilesRequest):
    tool_subsystem = BUILDDeprecationsFixer  # type: ignore[assignment]


class RenameFieldsInFileRequest(FixBUILDFileRequest):
    pass


@dataclass(frozen=True)
class RenamedFieldTypes:
    """Map deprecated field names to their new name, per target."""

    target_field_renames: FrozenDict[str, FrozenDict[str, str]]

    @classmethod
    def from_dict(cls, data: Mapping[str, Mapping[str, str]]) -> RenamedFieldTypes:
        return cls(
            FrozenDict(
                {
                    target_name: FrozenDict(field_renames.items())
                    for target_name, field_renames in data.items()
                }
            )
        )


@rule
async def determine_renamed_field_types(
    target_types: RegisteredTargetTypes, union_membership: UnionMembership
) -> RenamedFieldTypes:
    target_field_renames: DefaultDict[str, dict[str, str]] = defaultdict(dict)
    for tgt in target_types.types:
        field_types = list(tgt.class_field_types(union_membership))
        if issubclass(tgt, TargetGenerator):
            field_types.extend(tgt.moved_fields)

        for field_type in field_types:
            if field_type.deprecated_alias is not None:
                target_field_renames[tgt.alias][field_type.deprecated_alias] = field_type.alias

        # Make sure we also update deprecated fields in deprecated targets.
        if tgt.deprecated_alias is not None:
            target_field_renames[tgt.deprecated_alias] = target_field_renames[tgt.alias]

    return RenamedFieldTypes.from_dict(target_field_renames)


@rule
async def fix_single(
    request: RenameFieldsInFileRequest,
    renamed_field_types: RenamedFieldTypes,
) -> FixedBUILDFile:
    pants_target: str = ""
    level: int = 0
    tokens = iter(request.tokenize())

    def parse_level(token: tokenize.TokenInfo) -> bool:
        """Returns true if token was consumed."""
        nonlocal level

        if level == 0 or token.type is not tokenize.OP or token.string not in ["(", ")"]:
            return False

        if token.string == "(":
            level += 1
        elif token.string == ")":
            level -= 1

        return True

    def parse_target(token: tokenize.TokenInfo) -> bool:
        """Returns true if we're parsing a field name for a top level target."""
        nonlocal pants_target
        nonlocal level

        if parse_level(token):
            # Consumed parenthesis operator.
            return False

        if token.type is not tokenize.NAME:
            return False

        if level == 0 and next_token_is("("):
            level = 1
            pants_target = token.string
            # Current token consumed.
            return False

        return level == 1

    def next_token_is(string: str, token_type=tokenize.OP) -> bool:
        for next_token in tokens:
            if next_token.type is tokenize.NL:
                continue
            parse_level(next_token)
            return next_token.type is token_type and next_token.string == string
        return False

    def should_be_renamed(token: tokenize.TokenInfo) -> bool:
        nonlocal pants_target

        if not parse_target(token):
            return False

        if pants_target not in renamed_field_types.target_field_renames:
            return False

        return (
            next_token_is("=")
            and token.string in renamed_field_types.target_field_renames[pants_target]
        )

    updated_text_lines = list(request.lines)
    for token in tokens:
        if not should_be_renamed(token):
            continue
        line_index = token.start[0] - 1
        line = updated_text_lines[line_index]
        prefix = line[: token.start[1]]
        suffix = line[token.end[1] :]
        new_symbol = renamed_field_types.target_field_renames[pants_target][token.string]
        updated_text_lines[line_index] = f"{prefix}{new_symbol}{suffix}"

    return FixedBUILDFile(request.path, content="".join(updated_text_lines).encode("utf-8"))


@rule(desc="Fix deprecated field names", level=LogLevel.DEBUG)
async def fix(request: RenameFieldsInFilesRequest.Batch) -> FixResult:
    digest_contents = await get_digest_contents(request.snapshot.digest)
    fixed_contents = await concurrently(
        fix_single(
            RenameFieldsInFileRequest(file_content.path, file_content.content), **implicitly()
        )
        for file_content in digest_contents
    )
    snapshot = await digest_to_snapshot(
        **implicitly(
            CreateDigest(FileContent(content.path, content.content) for content in fixed_contents)
        )
    )
    return FixResult(
        request.snapshot, snapshot, "", "", tool_name=RenameFieldsInFilesRequest.tool_name
    )


def rules():
    return [
        *collect_rules(),
        *RenameFieldsInFilesRequest.rules(),
    ]

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	×
5
6	import tokenize	×
7	from collections import defaultdict	×
8	from collections.abc import Mapping	×
9	from dataclasses import dataclass	×
10	from typing import DefaultDict	×
11
12	from pants.backend.build_files.fix.base import FixBuildFilesRequest	×
13	from pants.backend.build_files.fix.deprecations.base import FixBUILDFileRequest, FixedBUILDFile	×
14	from pants.backend.build_files.fix.deprecations.subsystem import BUILDDeprecationsFixer	×
15	from pants.core.goals.fix import FixResult	×
16	from pants.engine.fs import CreateDigest, FileContent	×
17	from pants.engine.internals.selectors import concurrently	×
18	from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents	×
19	from pants.engine.rules import collect_rules, implicitly, rule	×
20	from pants.engine.target import RegisteredTargetTypes, TargetGenerator	×
21	from pants.engine.unions import UnionMembership	×
22	from pants.util.frozendict import FrozenDict	×
23	from pants.util.logging import LogLevel	×
24
25
26	class RenameFieldsInFilesRequest(FixBuildFilesRequest):	×
27	tool_subsystem = BUILDDeprecationsFixer # type: ignore[assignment]	×
28
29
30	class RenameFieldsInFileRequest(FixBUILDFileRequest):	×
31	pass	×
32
33
34	@dataclass(frozen=True)	×
35	class RenamedFieldTypes:	×
36	"""Map deprecated field names to their new name, per target."""
37
38	target_field_renames: FrozenDict[str, FrozenDict[str, str]]	×
39
40	@classmethod	×
41	def from_dict(cls, data: Mapping[str, Mapping[str, str]]) -> RenamedFieldTypes:	×
42	return cls(	×
43	FrozenDict(
44	{
45	target_name: FrozenDict(field_renames.items())
46	for target_name, field_renames in data.items()
47	}
48	)
49	)
50
51
52	@rule	×
53	async def determine_renamed_field_types(	×
54	target_types: RegisteredTargetTypes, union_membership: UnionMembership
55	) -> RenamedFieldTypes:
56	target_field_renames: DefaultDict[str, dict[str, str]] = defaultdict(dict)	×
57	for tgt in target_types.types:	×
58	field_types = list(tgt.class_field_types(union_membership))	×
59	if issubclass(tgt, TargetGenerator):	×
60	field_types.extend(tgt.moved_fields)	×
61
62	for field_type in field_types:	×
63	if field_type.deprecated_alias is not None:	×
64	target_field_renames[tgt.alias][field_type.deprecated_alias] = field_type.alias	×
65
66	# Make sure we also update deprecated fields in deprecated targets.
67	if tgt.deprecated_alias is not None:	×
68	target_field_renames[tgt.deprecated_alias] = target_field_renames[tgt.alias]	×
69
70	return RenamedFieldTypes.from_dict(target_field_renames)	×
71
72
73	@rule	×
74	async def fix_single(	×
75	request: RenameFieldsInFileRequest,
76	renamed_field_types: RenamedFieldTypes,
77	) -> FixedBUILDFile:
78	pants_target: str = ""	×
79	level: int = 0	×
80	tokens = iter(request.tokenize())	×
81
82	def parse_level(token: tokenize.TokenInfo) -> bool:	×
83	"""Returns true if token was consumed."""
84	nonlocal level
85
86	if level == 0 or token.type is not tokenize.OP or token.string not in ["(", ")"]:	×
87	return False	×
88
89	if token.string == "(":	×
90	level += 1	×
91	elif token.string == ")":	×
92	level -= 1	×
93
94	return True	×
95
96	def parse_target(token: tokenize.TokenInfo) -> bool:	×
97	"""Returns true if we're parsing a field name for a top level target."""
98	nonlocal pants_target
99	nonlocal level
100
101	if parse_level(token):	×
102	# Consumed parenthesis operator.
103	return False	×
104
105	if token.type is not tokenize.NAME:	×
106	return False	×
107
108	if level == 0 and next_token_is("("):	×
109	level = 1	×
110	pants_target = token.string	×
111	# Current token consumed.
112	return False	×
113
114	return level == 1	×
115
116	def next_token_is(string: str, token_type=tokenize.OP) -> bool:	×
117	for next_token in tokens:	×
118	if next_token.type is tokenize.NL:	×
119	continue	×
120	parse_level(next_token)	×
121	return next_token.type is token_type and next_token.string == string	×
122	return False	×
123
124	def should_be_renamed(token: tokenize.TokenInfo) -> bool:	×
125	nonlocal pants_target
126
127	if not parse_target(token):	×
128	return False	×
129
130	if pants_target not in renamed_field_types.target_field_renames:	×
131	return False	×
132
133	return (	×
134	next_token_is("=")
135	and token.string in renamed_field_types.target_field_renames[pants_target]
136	)
137
138	updated_text_lines = list(request.lines)	×
139	for token in tokens:	×
140	if not should_be_renamed(token):	×
141	continue	×
142	line_index = token.start[0] - 1	×
143	line = updated_text_lines[line_index]	×
144	prefix = line[: token.start[1]]	×
145	suffix = line[token.end[1] :]	×
146	new_symbol = renamed_field_types.target_field_renames[pants_target][token.string]	×
147	updated_text_lines[line_index] = f"{prefix}{new_symbol}{suffix}"	×
148
149	return FixedBUILDFile(request.path, content="".join(updated_text_lines).encode("utf-8"))	×
150
151
152	@rule(desc="Fix deprecated field names", level=LogLevel.DEBUG)	×
153	async def fix(request: RenameFieldsInFilesRequest.Batch) -> FixResult:	×
154	digest_contents = await get_digest_contents(request.snapshot.digest)	×
155	fixed_contents = await concurrently(	×
156	fix_single(
157	RenameFieldsInFileRequest(file_content.path, file_content.content), **implicitly()
158	)
159	for file_content in digest_contents
160	)
161	snapshot = await digest_to_snapshot(	×
162	**implicitly(
163	CreateDigest(FileContent(content.path, content.content) for content in fixed_contents)
164	)
165	)
166	return FixResult(	×
167	request.snapshot, snapshot, "", "", tool_name=RenameFieldsInFilesRequest.tool_name
168	)
169
170
171	def rules():	×
172	return [	×
173	*collect_rules(),
174	*RenameFieldsInFilesRequest.rules(),
175	]

pantsbuild / pants / 23177125175

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