25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

1294 existing lines in 76 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

94.92

/src/python/pants/backend/helm/goals/tailor.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
import os
from dataclasses import dataclass

from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import (
    HelmChartTarget,
    HelmUnitTestGeneratingSourcesField,
    HelmUnitTestTestsGeneratorTarget,
)
from pants.backend.helm.util_rules.chart_metadata import HELM_CHART_METADATA_FILENAMES
from pants.core.goals.tailor import (
    AllOwnedSources,
    PutativeTarget,
    PutativeTargets,
    PutativeTargetsRequest,
)
from pants.core.target_types import ResourcesGeneratorTarget
from pants.engine.fs import PathGlobs
from pants.engine.intrinsics import path_globs_to_paths
from pants.engine.rules import collect_rules, rule
from pants.engine.unions import UnionRule
from pants.source.filespec import FilespecMatcher
from pants.util.dirutil import group_by_dir
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class PutativeHelmTargetsRequest(PutativeTargetsRequest):
    pass


_TESTS_FOLDER_NAME = "tests"
_SNAPSHOT_FOLDER_NAME = "__snapshot__"
_SNAPSHOT_FILE_GLOBS = ("*_test.yaml.snap", "*_test.yml.snap")


@rule(desc="Determine candidate Helm chart targets to create", level=LogLevel.DEBUG)
async def find_putative_helm_targets(
    request: PutativeHelmTargetsRequest,
    all_owned_sources: AllOwnedSources,
    helm_subsystem: HelmSubsystem,
) -> PutativeTargets:
    putative_targets = []

    if helm_subsystem.tailor_charts:
        all_chart_files = await path_globs_to_paths(
            request.path_globs(*HELM_CHART_METADATA_FILENAMES)
        )
        unowned_chart_files = set(all_chart_files.files) - set(all_owned_sources)

        for chart_file in sorted(unowned_chart_files):
            dirname, filename = os.path.split(chart_file)
            putative_targets.append(
                PutativeTarget.for_target_type(
                    HelmChartTarget,
                    name=os.path.basename(dirname),
                    path=dirname,
                    triggering_sources=[filename],
                )
            )

        if helm_subsystem.tailor_unittests:
            chart_folders = {os.path.dirname(path) for path in all_chart_files.files}
            # Helm charts have a rigid folder structure and we rely on it
            # to successfully identify unit tests without false positives.
            all_unittest_files = await path_globs_to_paths(
                PathGlobs(
                    [
                        os.path.join(chart_root, _TESTS_FOLDER_NAME, glob)
                        for glob in HelmUnitTestGeneratingSourcesField.default
                        for chart_root in chart_folders
                    ]
                )
            )
            unonwned_unittest_files = set(all_unittest_files.files) - set(all_owned_sources)
            unittest_filespec_matcher = FilespecMatcher(
                HelmUnitTestGeneratingSourcesField.default, ()
            )
            unittest_files = {
                path
                for path in unonwned_unittest_files
                if os.path.basename(path)
                in set(
                    unittest_filespec_matcher.matches(
                        [os.path.basename(path) for path in unonwned_unittest_files]
                    )
                )
            }
            grouped_unittest_files = group_by_dir(unittest_files)

            # To prevent false positives, we look for snapshot files relative to the unit test sources
            all_snapshot_files = await path_globs_to_paths(
                PathGlobs(
                    [
                        os.path.join(dirname, _SNAPSHOT_FOLDER_NAME, glob)
                        for glob in _SNAPSHOT_FILE_GLOBS
                        for dirname in grouped_unittest_files.keys()
                    ]
                )
            )
            unowned_snapshot_files = set(all_snapshot_files.files) - set(all_owned_sources)
            grouped_snapshot_files = group_by_dir(unowned_snapshot_files)
            snapshot_folders = list(grouped_snapshot_files.keys())

            def find_snapshot_files_for(unittest_dir: str) -> set[str]:
                key = [
                    folder for folder in snapshot_folders if os.path.dirname(folder) == unittest_dir
                ]
                if not key:
                    return set()
                return grouped_snapshot_files[key[0]]

            for dirname, filenames in grouped_unittest_files.items():
                putative_targets.append(
                    PutativeTarget.for_target_type(
                        HelmUnitTestTestsGeneratorTarget,
                        path=dirname,
                        name=None,
                        triggering_sources=sorted(filenames),
                    )
                )

                snapshot_files = find_snapshot_files_for(dirname)
                if snapshot_files:
                    putative_targets.append(
                        PutativeTarget.for_target_type(
                            ResourcesGeneratorTarget,
                            path=os.path.join(dirname, _SNAPSHOT_FOLDER_NAME),
                            name=None,
                            triggering_sources=sorted(snapshot_files),
                            kwargs={"sources": _SNAPSHOT_FILE_GLOBS},
                        )
                    )

    elif helm_subsystem.tailor_unittests:
        logging.warning(
            softwrap(
                """
                Pants can not identify Helm unit tests when `[helm].tailor_charts` has been set to `False`.

                If this is intentional, then set `[helm].tailor_unittests` to `False` to disable this warning.
                """
            )
        )

    return PutativeTargets(putative_targets)


def rules():
    return [*collect_rules(), UnionRule(PutativeTargetsRequest, PutativeHelmTargetsRequest)]

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	2✔
5
6	import logging	2✔
7	import os	2✔
8	from dataclasses import dataclass	2✔
9
10	from pants.backend.helm.subsystems.helm import HelmSubsystem	2✔
11	from pants.backend.helm.target_types import (	2✔
12	HelmChartTarget,
13	HelmUnitTestGeneratingSourcesField,
14	HelmUnitTestTestsGeneratorTarget,
15	)
16	from pants.backend.helm.util_rules.chart_metadata import HELM_CHART_METADATA_FILENAMES	2✔
17	from pants.core.goals.tailor import (	2✔
18	AllOwnedSources,
19	PutativeTarget,
20	PutativeTargets,
21	PutativeTargetsRequest,
22	)
23	from pants.core.target_types import ResourcesGeneratorTarget	2✔
24	from pants.engine.fs import PathGlobs	2✔
25	from pants.engine.intrinsics import path_globs_to_paths	2✔
26	from pants.engine.rules import collect_rules, rule	2✔
27	from pants.engine.unions import UnionRule	2✔
28	from pants.source.filespec import FilespecMatcher	2✔
29	from pants.util.dirutil import group_by_dir	2✔
30	from pants.util.logging import LogLevel	2✔
31	from pants.util.strutil import softwrap	2✔
32
33	logger = logging.getLogger(__name__)	2✔
34
35
36	@dataclass(frozen=True)	2✔
37	class PutativeHelmTargetsRequest(PutativeTargetsRequest):	2✔
38	pass	2✔
39
40
41	_TESTS_FOLDER_NAME = "tests"	2✔
42	_SNAPSHOT_FOLDER_NAME = "__snapshot__"	2✔
43	_SNAPSHOT_FILE_GLOBS = ("_test.yaml.snap", "_test.yml.snap")	2✔
44
45
46	@rule(desc="Determine candidate Helm chart targets to create", level=LogLevel.DEBUG)	2✔
47	async def find_putative_helm_targets(	2✔
48	request: PutativeHelmTargetsRequest,
49	all_owned_sources: AllOwnedSources,
50	helm_subsystem: HelmSubsystem,
51	) -> PutativeTargets:
UNCOV 52	putative_targets = []	1✔
53
UNCOV 54	if helm_subsystem.tailor_charts:	1✔
UNCOV 55	all_chart_files = await path_globs_to_paths(	1✔
56	request.path_globs(*HELM_CHART_METADATA_FILENAMES)
57	)
UNCOV 58	unowned_chart_files = set(all_chart_files.files) - set(all_owned_sources)	1✔
59
UNCOV 60	for chart_file in sorted(unowned_chart_files):	1✔
UNCOV 61	dirname, filename = os.path.split(chart_file)	1✔
UNCOV 62	putative_targets.append(	1✔
63	PutativeTarget.for_target_type(
64	HelmChartTarget,
65	name=os.path.basename(dirname),
66	path=dirname,
67	triggering_sources=[filename],
68	)
69	)
70
UNCOV 71	if helm_subsystem.tailor_unittests:	1✔
UNCOV 72	chart_folders = {os.path.dirname(path) for path in all_chart_files.files}	1✔
73	# Helm charts have a rigid folder structure and we rely on it
74	# to successfully identify unit tests without false positives.
UNCOV 75	all_unittest_files = await path_globs_to_paths(	1✔
76	PathGlobs(
77	[
78	os.path.join(chart_root, _TESTS_FOLDER_NAME, glob)
79	for glob in HelmUnitTestGeneratingSourcesField.default
80	for chart_root in chart_folders
81	]
82	)
83	)
UNCOV 84	unonwned_unittest_files = set(all_unittest_files.files) - set(all_owned_sources)	1✔
UNCOV 85	unittest_filespec_matcher = FilespecMatcher(	1✔
86	HelmUnitTestGeneratingSourcesField.default, ()
87	)
UNCOV 88	unittest_files = {	1✔
89	path
90	for path in unonwned_unittest_files
91	if os.path.basename(path)
92	in set(
93	unittest_filespec_matcher.matches(
94	[os.path.basename(path) for path in unonwned_unittest_files]
95	)
96	)
97	}
UNCOV 98	grouped_unittest_files = group_by_dir(unittest_files)	1✔
99
100	# To prevent false positives, we look for snapshot files relative to the unit test sources
UNCOV 101	all_snapshot_files = await path_globs_to_paths(	1✔
102	PathGlobs(
103	[
104	os.path.join(dirname, _SNAPSHOT_FOLDER_NAME, glob)
105	for glob in _SNAPSHOT_FILE_GLOBS
106	for dirname in grouped_unittest_files.keys()
107	]
108	)
109	)
UNCOV 110	unowned_snapshot_files = set(all_snapshot_files.files) - set(all_owned_sources)	1✔
UNCOV 111	grouped_snapshot_files = group_by_dir(unowned_snapshot_files)	1✔
UNCOV 112	snapshot_folders = list(grouped_snapshot_files.keys())	1✔
113
UNCOV 114	def find_snapshot_files_for(unittest_dir: str) -> set[str]:	1✔
UNCOV 115	key = [	1✔
116	folder for folder in snapshot_folders if os.path.dirname(folder) == unittest_dir
117	]
UNCOV 118	if not key:	1✔
119	return set()	×
UNCOV 120	return grouped_snapshot_files[key[0]]	1✔
121
UNCOV 122	for dirname, filenames in grouped_unittest_files.items():	1✔
UNCOV 123	putative_targets.append(	1✔
124	PutativeTarget.for_target_type(
125	HelmUnitTestTestsGeneratorTarget,
126	path=dirname,
127	name=None,
128	triggering_sources=sorted(filenames),
129	)
130	)
131
UNCOV 132	snapshot_files = find_snapshot_files_for(dirname)	1✔
UNCOV 133	if snapshot_files:	1✔
UNCOV 134	putative_targets.append(	1✔
135	PutativeTarget.for_target_type(
136	ResourcesGeneratorTarget,
137	path=os.path.join(dirname, _SNAPSHOT_FOLDER_NAME),
138	name=None,
139	triggering_sources=sorted(snapshot_files),
140	kwargs={"sources": _SNAPSHOT_FILE_GLOBS},
141	)
142	)
143
144	elif helm_subsystem.tailor_unittests:	×
145	logging.warning(	×
146	softwrap(
147	"""
148	Pants can not identify Helm unit tests when `[helm].tailor_charts` has been set to `False`.
149
150	If this is intentional, then set `[helm].tailor_unittests` to `False` to disable this warning.
151	"""
152	)
153	)
154
UNCOV 155	return PutativeTargets(putative_targets)	1✔
156
157
158	def rules():	2✔
UNCOV 159	return [*collect_rules(), UnionRule(PutativeTargetsRequest, PutativeHelmTargetsRequest)]	1✔

pantsbuild / pants / 25443604553

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