25259185675

Committed 02 May 2026 06:47PM UTC coverage: 92.141% (-0.8%) from 92.955%

Build # 25259185675

Build Type

push

github

Committed by

web-flow

Commit Message

Fix the dynamic UI. (#23306)

In #23114 we upgraded to indicatif 0.18.4,
which included a fix to respect TERM, and 
display nothing if it's unset.

Since we did not pass TERM through pantsd, the
dynamic ui is now not shown. 

This change fixes that, and also pass NO_COLOR
through, since indicatif inspects it too.

Coverage Stats

88773 of 96345 relevant lines covered (92.14%)

3.83 hits per line

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

95.17

/src/python/pants/backend/javascript/nodejs_project.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import itertools
import os.path
from collections.abc import Iterable
from dataclasses import dataclass, replace
from pathlib import PurePath

from pants.backend.javascript import package_json
from pants.backend.javascript.package_json import (
    AllPackageJson,
    PackageJson,
    PnpmWorkspaceGlobs,
    PnpmWorkspaces,
)
from pants.backend.javascript.package_manager import PackageManager
from pants.backend.javascript.subsystems import nodejs
from pants.backend.javascript.subsystems.nodejs import NodeJS, UserChosenNodeJSResolveAliases
from pants.core.util_rules import stripped_source_files
from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name
from pants.engine.collection import Collection
from pants.engine.internals.native_engine import MergeDigests
from pants.engine.rules import Rule, collect_rules, rule
from pants.engine.unions import UnionRule
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.strutil import bullet_list, softwrap


@dataclass(frozen=True)
class _TentativeProject:
    root_dir: str
    workspaces: FrozenOrderedSet[PackageJson]
    default_resolve_name: str

    def is_parent(self, project: _TentativeProject) -> bool:
        return self.root_dir != project.root_dir and any(
            project.root_dir == workspace.root_dir for workspace in self.workspaces
        )

    def including_workspaces_from(self, child: _TentativeProject) -> _TentativeProject:
        return replace(self, workspaces=self.workspaces | child.workspaces)

    def root_workspace(self) -> PackageJson | None:
        for ws in self.workspaces:
            if ws.root_dir == self.root_dir:
                return ws
        return None


@dataclass(frozen=True)
class NodeJSProject:
    root_dir: str
    workspaces: FrozenOrderedSet[PackageJson]
    default_resolve_name: str
    package_manager: PackageManager
    pnpm_workspace: PnpmWorkspaceGlobs | None = None

    @property
    def lockfile_name(self) -> str:
        return self.package_manager.lockfile_name

    @property
    def generate_lockfile_args(self) -> tuple[str, ...]:
        return self.package_manager.generate_lockfile_args

    @property
    def immutable_install_args(self) -> tuple[str, ...]:
        return self.package_manager.immutable_install_args

    @property
    def workspace_specifier_arg(self) -> str:
        return self.package_manager.workspace_specifier_arg

    @property
    def args_separator(self) -> tuple[str, ...]:
        return self.package_manager.run_arg_separator

    def extra_env(self) -> dict[str, str]:
        return dict(self.package_manager.extra_env)

    @property
    def pack_archive_format(self) -> str:
        return self.package_manager.pack_archive_format

    def extra_caches(self) -> dict[str, str]:
        return dict(self.package_manager.extra_caches)

    def get_project_digest(self) -> MergeDigests:
        return MergeDigests(
            itertools.chain(
                (ws.digest for ws in self.workspaces),
                [self.pnpm_workspace.digest] if self.pnpm_workspace else [],
            )
        )

    @property
    def single_workspace(self) -> bool:
        return len(self.workspaces) == 1 and next(iter(self.workspaces)).root_dir == self.root_dir

    @classmethod
    def from_tentative(
        cls,
        project: _TentativeProject,
        nodejs: NodeJS,
        pnpm_workspaces: PnpmWorkspaces,
    ) -> NodeJSProject:
        root_ws = project.root_workspace()
        package_manager: str | None = None
        if root_ws:
            package_manager = root_ws.package_manager or nodejs.default_package_manager
        if not package_manager:
            raise ValueError(
                softwrap(
                    f"""
                    Could not determine package manager for project {project.default_resolve_name}.

                    Either configure a default [{NodeJS.name}].package_manager, or set the root
                    `package.json#packageManager` property.
                    """
                )
            )

        for workspace in project.workspaces:
            if workspace.package_manager:
                if not package_manager == workspace.package_manager:
                    ws_ref = f"{workspace.name}@{workspace.version}"
                    raise ValueError(
                        softwrap(
                            f"""
                            Workspace {ws_ref}'s package manager
                            {workspace.package_manager} is not compatible with
                            project {project.default_resolve_name}'s package manager {package_manager}.

                            Move or duplicate the `package.json#packageManager` entry from the
                            workspace {ws_ref} to the root package to resolve this error.
                            """
                        )
                    )

        return NodeJSProject(
            root_dir=project.root_dir,
            workspaces=project.workspaces,
            default_resolve_name=project.default_resolve_name or "nodejs-default",
            package_manager=PackageManager.from_string(package_manager),
            pnpm_workspace=pnpm_workspaces.for_root(project.root_dir),
        )


class AllNodeJSProjects(Collection[NodeJSProject]):
    def project_for_directory(self, directory: str) -> NodeJSProject:
        for project in self:
            if directory in (workspace.root_dir for workspace in project.workspaces):
                return project
        raise ValueError(
            f"{directory} is not a package directory that is part of a project. This is likely a bug."
        )


@dataclass(frozen=True)
class ProjectPaths:
    root: str
    project_globs: list[str]

    def full_globs(self) -> Iterable[str]:
        return (os.path.join(self.root, project) for project in self.project_globs)

    def matches_glob(self, pkg_json: PackageJson) -> bool:
        path = PurePath(pkg_json.root_dir)

        def safe_match(glob: str) -> bool:
            if glob == "":
                return pkg_json.root_dir == ""
            return path.match(glob)

        return any(safe_match(glob) for glob in self.full_globs())


async def _get_default_resolve_name(path: str) -> str:
    stripped = await strip_file_name(StrippedFileNameRequest(path))
    return stripped.value.replace(os.path.sep, ".")


@rule
async def find_node_js_projects(
    package_workspaces: AllPackageJson,
    pnpm_workspaces: PnpmWorkspaces,
    nodejs: NodeJS,
    resolve_names: UserChosenNodeJSResolveAliases,
) -> AllNodeJSProjects:
    project_paths = []
    for pkg in package_workspaces:
        package_manager = pkg.package_manager or nodejs.default_package_manager
        if (
            package_manager and PackageManager.from_string(package_manager).name == "pnpm"
        ):  # case for pnpm
            if pkg in pnpm_workspaces:
                project_paths.append(
                    ProjectPaths(pkg.root_dir, ["", *pnpm_workspaces[pkg].packages])
                )
            else:
                project_paths.append(ProjectPaths(pkg.root_dir, [""]))
        else:  # case for npm, yarn
            project_paths.append(ProjectPaths(pkg.root_dir, ["", *pkg.workspaces]))

    node_js_projects = {
        _TentativeProject(
            paths.root,
            FrozenOrderedSet(pkg for pkg in package_workspaces if paths.matches_glob(pkg)),
            await _get_default_resolve_name(paths.root),
        )
        for paths in project_paths
    }
    merged_projects = _merge_workspaces(node_js_projects)
    all_projects = AllNodeJSProjects(
        NodeJSProject.from_tentative(p, nodejs, pnpm_workspaces) for p in merged_projects
    )
    _ensure_resolve_names_are_unique(all_projects, resolve_names)

    return all_projects


_AMBIGUOUS_RESOLVE_SOLUTIONS = [
    f"Configure [{NodeJS.options_scope}].resolves to grant the package.json directories different names.",
    "Make one package a workspace of the other.",
    "Re-configure your source root(s).",
]


def _ensure_resolve_names_are_unique(
    all_projects: AllNodeJSProjects, resolve_names: UserChosenNodeJSResolveAliases
) -> None:
    seen: dict[str, NodeJSProject] = {}
    for project in all_projects:
        resolve_name = resolve_names.get(project.root_dir, project.default_resolve_name)
        seen_project = seen.get(resolve_name)
        if seen_project:
            raise ValueError(
                softwrap(
                    f"""
                    Projects with root directories '{project.root_dir}' and '{seen_project.root_dir}'
                    have the same resolve name {resolve_name}. This will cause ambiguities.

                    To disambiguate, either:\n\n
                    {bullet_list(_AMBIGUOUS_RESOLVE_SOLUTIONS)}
                    """
                )
            )
        seen[resolve_name] = project


def _project_to_parents(
    projects: set[_TentativeProject],
) -> dict[_TentativeProject, list[_TentativeProject]]:
    return {
        project: [
            candidate_parent for candidate_parent in projects if candidate_parent.is_parent(project)
        ]
        for project in sorted(projects, key=lambda p: p.root_dir, reverse=False)
    }


def _merge_workspaces(node_js_projects: set[_TentativeProject]) -> Iterable[_TentativeProject]:
    project_to_parents = _project_to_parents(node_js_projects)
    while any(parents for parents in project_to_parents.values()):
        _ensure_one_parent(project_to_parents)
        node_js_projects = set()
        for project, parents in project_to_parents.items():
            node_js_projects -= {project, *parents}
            node_js_projects.add(
                parents[0].including_workspaces_from(project) if parents else project
            )
        project_to_parents = _project_to_parents(node_js_projects)
    return node_js_projects


def _ensure_one_parent(
    project_to_parents: dict[_TentativeProject, list[_TentativeProject]],
) -> None:
    for project, parents in project_to_parents.items():
        if len(parents) > 1:
            raise ValueError(
                softwrap(
                    f"""
                    Nodejs projects {", ".join(parent.root_dir for parent in parents)}
                    are specifying {project.root_dir} to be part of their workspaces.

                    A package can only be part of one project.
                    """
                )
            )


def rules() -> Iterable[Rule | UnionRule]:
    return [
        *nodejs.rules(),
        *package_json.rules(),
        *stripped_source_files.rules(),
        *collect_rules(),
    ]

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	10✔
4
5	import itertools	10✔
6	import os.path	10✔
7	from collections.abc import Iterable	10✔
8	from dataclasses import dataclass, replace	10✔
9	from pathlib import PurePath	10✔
10
11	from pants.backend.javascript import package_json	10✔
12	from pants.backend.javascript.package_json import (	10✔
13	AllPackageJson,
14	PackageJson,
15	PnpmWorkspaceGlobs,
16	PnpmWorkspaces,
17	)
18	from pants.backend.javascript.package_manager import PackageManager	10✔
19	from pants.backend.javascript.subsystems import nodejs	10✔
20	from pants.backend.javascript.subsystems.nodejs import NodeJS, UserChosenNodeJSResolveAliases	10✔
21	from pants.core.util_rules import stripped_source_files	10✔
22	from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name	10✔
23	from pants.engine.collection import Collection	10✔
24	from pants.engine.internals.native_engine import MergeDigests	10✔
25	from pants.engine.rules import Rule, collect_rules, rule	10✔
26	from pants.engine.unions import UnionRule	10✔
27	from pants.util.ordered_set import FrozenOrderedSet	10✔
28	from pants.util.strutil import bullet_list, softwrap	10✔
29
30
31	@dataclass(frozen=True)	10✔
32	class _TentativeProject:	10✔
33	root_dir: str	10✔
34	workspaces: FrozenOrderedSet[PackageJson]	10✔
35	default_resolve_name: str	10✔
36
37	def is_parent(self, project: _TentativeProject) -> bool:	10✔
38	return self.root_dir != project.root_dir and any(	8✔
39	project.root_dir == workspace.root_dir for workspace in self.workspaces
40	)
41
42	def including_workspaces_from(self, child: _TentativeProject) -> _TentativeProject:	10✔
43	return replace(self, workspaces=self.workspaces \| child.workspaces)	4✔
44
45	def root_workspace(self) -> PackageJson \| None:	10✔
46	for ws in self.workspaces:	8✔
47	if ws.root_dir == self.root_dir:	8✔
48	return ws	8✔
49	return None	×
50
51
52	@dataclass(frozen=True)	10✔
53	class NodeJSProject:	10✔
54	root_dir: str	10✔
55	workspaces: FrozenOrderedSet[PackageJson]	10✔
56	default_resolve_name: str	10✔
57	package_manager: PackageManager	10✔
58	pnpm_workspace: PnpmWorkspaceGlobs \| None = None	10✔
59
60	@property	10✔
61	def lockfile_name(self) -> str:	10✔
62	return self.package_manager.lockfile_name	8✔
63
64	@property	10✔
65	def generate_lockfile_args(self) -> tuple[str, ...]:	10✔
66	return self.package_manager.generate_lockfile_args	1✔
67
68	@property	10✔
69	def immutable_install_args(self) -> tuple[str, ...]:	10✔
70	return self.package_manager.immutable_install_args	7✔
71
72	@property	10✔
73	def workspace_specifier_arg(self) -> str:	10✔
74	return self.package_manager.workspace_specifier_arg	2✔
75
76	@property	10✔
77	def args_separator(self) -> tuple[str, ...]:	10✔
78	return self.package_manager.run_arg_separator	1✔
79
80	def extra_env(self) -> dict[str, str]:	10✔
81	return dict(self.package_manager.extra_env)	7✔
82
83	@property	10✔
84	def pack_archive_format(self) -> str:	10✔
85	return self.package_manager.pack_archive_format	1✔
86
87	def extra_caches(self) -> dict[str, str]:	10✔
88	return dict(self.package_manager.extra_caches)	7✔
89
90	def get_project_digest(self) -> MergeDigests:	10✔
91	return MergeDigests(	7✔
92	itertools.chain(
93	(ws.digest for ws in self.workspaces),
94	[self.pnpm_workspace.digest] if self.pnpm_workspace else [],
95	)
96	)
97
98	@property	10✔
99	def single_workspace(self) -> bool:	10✔
100	return len(self.workspaces) == 1 and next(iter(self.workspaces)).root_dir == self.root_dir	7✔
101
102	@classmethod	10✔
103	def from_tentative(	10✔
104	cls,
105	project: _TentativeProject,
106	nodejs: NodeJS,
107	pnpm_workspaces: PnpmWorkspaces,
108	) -> NodeJSProject:
109	root_ws = project.root_workspace()	8✔
110	package_manager: str \| None = None	8✔
111	if root_ws:	8✔
112	package_manager = root_ws.package_manager or nodejs.default_package_manager	8✔
113	if not package_manager:	8✔
114	raise ValueError(	×
115	softwrap(
116	f"""
117	Could not determine package manager for project {project.default_resolve_name}.
118
119	Either configure a default [{NodeJS.name}].package_manager, or set the root
120	`package.json#packageManager` property.
121	"""
122	)
123	)
124
125	for workspace in project.workspaces:	8✔
126	if workspace.package_manager:	8✔
127	if not package_manager == workspace.package_manager:	6✔
128	ws_ref = f"{workspace.name}@{workspace.version}"	×
129	raise ValueError(	×
130	softwrap(
131	f"""
132	Workspace {ws_ref}'s package manager
133	{workspace.package_manager} is not compatible with
134	project {project.default_resolve_name}'s package manager {package_manager}.
135
136	Move or duplicate the `package.json#packageManager` entry from the
137	workspace {ws_ref} to the root package to resolve this error.
138	"""
139	)
140	)
141
142	return NodeJSProject(	8✔
143	root_dir=project.root_dir,
144	workspaces=project.workspaces,
145	default_resolve_name=project.default_resolve_name or "nodejs-default",
146	package_manager=PackageManager.from_string(package_manager),
147	pnpm_workspace=pnpm_workspaces.for_root(project.root_dir),
148	)
149
150
151	class AllNodeJSProjects(Collection[NodeJSProject]):	10✔
152	def project_for_directory(self, directory: str) -> NodeJSProject:	10✔
153	for project in self:	8✔
154	if directory in (workspace.root_dir for workspace in project.workspaces):	8✔
155	return project	8✔
156	raise ValueError(	×
157	f"{directory} is not a package directory that is part of a project. This is likely a bug."
158	)
159
160
161	@dataclass(frozen=True)	10✔
162	class ProjectPaths:	10✔
163	root: str	10✔
164	project_globs: list[str]	10✔
165
166	def full_globs(self) -> Iterable[str]:	10✔
167	return (os.path.join(self.root, project) for project in self.project_globs)	8✔
168
169	def matches_glob(self, pkg_json: PackageJson) -> bool:	10✔
170	path = PurePath(pkg_json.root_dir)	8✔
171
172	def safe_match(glob: str) -> bool:	8✔
173	if glob == "":	8✔
174	return pkg_json.root_dir == ""	4✔
175	return path.match(glob)	6✔
176
177	return any(safe_match(glob) for glob in self.full_globs())	8✔
178
179
180	async def _get_default_resolve_name(path: str) -> str:	10✔
181	stripped = await strip_file_name(StrippedFileNameRequest(path))	8✔
182	return stripped.value.replace(os.path.sep, ".")	8✔
183
184
185	@rule	10✔
186	async def find_node_js_projects(	10✔
187	package_workspaces: AllPackageJson,
188	pnpm_workspaces: PnpmWorkspaces,
189	nodejs: NodeJS,
190	resolve_names: UserChosenNodeJSResolveAliases,
191	) -> AllNodeJSProjects:
192	project_paths = []	8✔
193	for pkg in package_workspaces:	8✔
194	package_manager = pkg.package_manager or nodejs.default_package_manager	8✔
195	if (	8✔
196	package_manager and PackageManager.from_string(package_manager).name == "pnpm"
197	): # case for pnpm
198	if pkg in pnpm_workspaces:	4✔
199	project_paths.append(	3✔
200	ProjectPaths(pkg.root_dir, ["", *pnpm_workspaces[pkg].packages])
201	)
202	else:
203	project_paths.append(ProjectPaths(pkg.root_dir, [""]))	4✔
204	else: # case for npm, yarn
205	project_paths.append(ProjectPaths(pkg.root_dir, ["", *pkg.workspaces]))	8✔
206
207	node_js_projects = {	8✔
208	_TentativeProject(
209	paths.root,
210	FrozenOrderedSet(pkg for pkg in package_workspaces if paths.matches_glob(pkg)),
211	await _get_default_resolve_name(paths.root),
212	)
213	for paths in project_paths
214	}
215	merged_projects = _merge_workspaces(node_js_projects)	8✔
216	all_projects = AllNodeJSProjects(	8✔
217	NodeJSProject.from_tentative(p, nodejs, pnpm_workspaces) for p in merged_projects
218	)
219	_ensure_resolve_names_are_unique(all_projects, resolve_names)	8✔
220
221	return all_projects	8✔
222
223
224	_AMBIGUOUS_RESOLVE_SOLUTIONS = [	10✔
225	f"Configure [{NodeJS.options_scope}].resolves to grant the package.json directories different names.",
226	"Make one package a workspace of the other.",
227	"Re-configure your source root(s).",
228	]
229
230
231	def _ensure_resolve_names_are_unique(	10✔
232	all_projects: AllNodeJSProjects, resolve_names: UserChosenNodeJSResolveAliases
233	) -> None:
234	seen: dict[str, NodeJSProject] = {}	8✔
235	for project in all_projects:	8✔
236	resolve_name = resolve_names.get(project.root_dir, project.default_resolve_name)	8✔
237	seen_project = seen.get(resolve_name)	8✔
238	if seen_project:	8✔
239	raise ValueError(	×
240	softwrap(
241	f"""
242	Projects with root directories '{project.root_dir}' and '{seen_project.root_dir}'
243	have the same resolve name {resolve_name}. This will cause ambiguities.
244
245	To disambiguate, either:\n\n
246	{bullet_list(_AMBIGUOUS_RESOLVE_SOLUTIONS)}
247	"""
248	)
249	)
250	seen[resolve_name] = project	8✔
251
252
253	def _project_to_parents(	10✔
254	projects: set[_TentativeProject],
255	) -> dict[_TentativeProject, list[_TentativeProject]]:
256	return {	8✔
257	project: [
258	candidate_parent for candidate_parent in projects if candidate_parent.is_parent(project)
259	]
260	for project in sorted(projects, key=lambda p: p.root_dir, reverse=False)
261	}
262
263
264	def _merge_workspaces(node_js_projects: set[_TentativeProject]) -> Iterable[_TentativeProject]:	10✔
265	project_to_parents = _project_to_parents(node_js_projects)	8✔
266	while any(parents for parents in project_to_parents.values()):	8✔
267	_ensure_one_parent(project_to_parents)	4✔
268	node_js_projects = set()	4✔
269	for project, parents in project_to_parents.items():	4✔
270	node_js_projects -= {project, *parents}	4✔
271	node_js_projects.add(	4✔
272	parents[0].including_workspaces_from(project) if parents else project
273	)
274	project_to_parents = _project_to_parents(node_js_projects)	4✔
275	return node_js_projects	8✔
276
277
278	def _ensure_one_parent(	10✔
279	project_to_parents: dict[_TentativeProject, list[_TentativeProject]],
280	) -> None:
281	for project, parents in project_to_parents.items():	4✔
282	if len(parents) > 1:	4✔
283	raise ValueError(	×
284	softwrap(
285	f"""
286	Nodejs projects {", ".join(parent.root_dir for parent in parents)}
287	are specifying {project.root_dir} to be part of their workspaces.
288
289	A package can only be part of one project.
290	"""
291	)
292	)
293
294
295	def rules() -> Iterable[Rule \| UnionRule]:	10✔
296	return [	10✔
297	*nodejs.rules(),
298	*package_json.rules(),
299	*stripped_source_files.rules(),
300	*collect_rules(),
301	]

pantsbuild / pants / 25259185675

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