22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

85.11

/src/python/pants/backend/helm/subsystems/k8s_parser.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import json
import logging
import pkgutil
from dataclasses import dataclass
from pathlib import PurePath
from typing import Any

from pants.backend.helm.utils.yaml import YamlPath
from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase
from pants.backend.python.target_types import EntryPoint
from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.pex import (
    VenvPex,
    VenvPexProcess,
    VenvPexRequest,
    create_venv_pex,
)
from pants.backend.python.util_rules.pex_environment import PexEnvironment
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.fs import CreateDigest, FileContent, FileEntry
from pants.engine.intrinsics import create_digest, execute_process
from pants.engine.rules import collect_rules, implicitly, rule
from pants.option.option_types import DictOption
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize, softwrap

logger = logging.getLogger(__name__)

_HELM_K8S_PARSER_SOURCE = "k8s_parser_main.py"
_HELM_K8S_PARSER_PACKAGE = "pants.backend.helm.subsystems"


class HelmKubeParserSubsystem(PythonToolRequirementsBase):
    options_scope = "helm-k8s-parser"
    help_short = "Analyses K8S manifests rendered by Helm."

    default_requirements = [
        "hikaru>=1.1.0",
        "hikaru-model-28",
        "hikaru-model-27",
        "hikaru-model-26",
        "hikaru-model-25",
        "hikaru-model-24",
        "hikaru-model-23",
    ]

    register_interpreter_constraints = True
    crd = DictOption[str](
        help=softwrap(
            """
            Additional custom resource definitions be made available to all Helm processes
            or during value interpolation.
            Example:
                [helm-k8s-parser.crd]
                "filename1"="classname1"
                "filename2"="classname2"
            """
        ),
        default={},
    )

    default_lockfile_resource = (_HELM_K8S_PARSER_PACKAGE, "k8s_parser.lock")


@dataclass(frozen=True)
class _HelmKubeParserTool:
    pex: VenvPex
    crd: str


@rule
async def build_k8s_parser_tool(
    k8s_parser: HelmKubeParserSubsystem,
    pex_environment: PexEnvironment,
) -> _HelmKubeParserTool:
    parser_sources = pkgutil.get_data(_HELM_K8S_PARSER_PACKAGE, _HELM_K8S_PARSER_SOURCE)

    if not parser_sources:
        raise ValueError(
            f"Unable to find source to {_HELM_K8S_PARSER_SOURCE!r} in {_HELM_K8S_PARSER_PACKAGE}"
        )

    parser_file_content = FileContent(
        path="__k8s_parser.py", content=parser_sources, is_executable=True
    )

    digest_sources = [parser_file_content]

    modulename_classname = []
    if k8s_parser.crd != {}:
        for file, classname in k8s_parser.crd.items():
            crd_sources = open(file, "rb").read()
            if not crd_sources:
                raise ValueError(
                    f"Unable to find source to customer resource definition in {_HELM_K8S_PARSER_PACKAGE}"
                )
            unique_name = f"_crd_source_{hash(file)}"
            parser_file_content_source = FileContent(
                path=f"{unique_name}.py", content=crd_sources, is_executable=False
            )
            digest_sources.append(parser_file_content_source)
            modulename_classname.append((unique_name, classname))
    parser_digest = await create_digest(CreateDigest(digest_sources))

    # We use copies of site packages because hikaru gets confused with symlinked packages
    # The core hikaru package tries to load the packages containing the kubernetes-versioned models
    # using the __path__ attribute of the core package,
    # which doesn't work when the packages are symlinked from inside the namespace-handling dirs in the PEX
    use_site_packages_copies = True

    parser_pex = await create_venv_pex(
        VenvPexRequest(
            k8s_parser.to_pex_request(
                main=EntryPoint(PurePath(parser_file_content.path).stem),
                sources=parser_digest,
            ),
            pex_environment.in_sandbox(working_directory=None),
            site_packages_copies=use_site_packages_copies,
        ),
        **implicitly(),
    )
    return _HelmKubeParserTool(parser_pex, json.dumps(modulename_classname))


@dataclass(frozen=True)
class ParseKubeManifestRequest(EngineAwareParameter):
    file: FileEntry

    def debug_hint(self) -> str | None:
        return self.file.path

    def metadata(self) -> dict[str, Any] | None:
        return {"file": self.file}


@dataclass(frozen=True)
class ParsedImageRefEntry:
    document_index: int
    path: YamlPath
    unparsed_image_ref: str


@dataclass(frozen=True)
class ParsedKubeManifest(EngineAwareReturnType):
    filename: str
    found_image_refs: tuple[ParsedImageRefEntry, ...]

    def level(self) -> LogLevel | None:
        return LogLevel.DEBUG

    def message(self) -> str | None:
        return f"Found {pluralize(len(self.found_image_refs), 'image reference')} in file {self.filename}"

    def metadata(self) -> dict[str, Any] | None:
        return {
            "filename": self.filename,
            "found_image_refs": self.found_image_refs,
        }


@rule(desc="Parse Kubernetes resource manifest")
async def parse_kube_manifest(
    request: ParseKubeManifestRequest, tool: _HelmKubeParserTool
) -> ParsedKubeManifest:
    file_digest = await create_digest(CreateDigest([request.file]))

    result = await execute_process(
        **implicitly(
            VenvPexProcess(
                tool.pex,
                argv=[request.file.path, tool.crd],
                input_digest=file_digest,
                description=f"Analyzing Kubernetes manifest {request.file.path}",
                level=LogLevel.DEBUG,
            )
        )
    )

    if result.exit_code == 0:
        output = result.stdout.decode("utf-8").splitlines()
        image_refs: list[ParsedImageRefEntry] = []
        for line in output:
            parts = line.split(",")
            if len(parts) != 3:
                raise Exception(
                    softwrap(
                        f"""Unexpected output from k8s parser when parsing file {request.file.path}:

                        {line}
                        """
                    )
                )

            image_refs.append(
                ParsedImageRefEntry(
                    document_index=int(parts[0]),
                    path=YamlPath.parse(parts[1]),
                    unparsed_image_ref=parts[2],
                )
            )

        return ParsedKubeManifest(filename=request.file.path, found_image_refs=tuple(image_refs))
    else:
        parser_error = result.stderr.decode("utf-8")
        raise Exception(
            softwrap(
                f"""
                Could not parse Kubernetes manifests in file: {request.file.path}.
                {parser_error}
                """
            )
        )


def rules():
    return [
        *collect_rules(),
        *pex.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	3✔
5
6	import json	3✔
7	import logging	3✔
8	import pkgutil	3✔
9	from dataclasses import dataclass	3✔
10	from pathlib import PurePath	3✔
11	from typing import Any	3✔
12
13	from pants.backend.helm.utils.yaml import YamlPath	3✔
14	from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase	3✔
15	from pants.backend.python.target_types import EntryPoint	3✔
16	from pants.backend.python.util_rules import pex	3✔
17	from pants.backend.python.util_rules.pex import (	3✔
18	VenvPex,
19	VenvPexProcess,
20	VenvPexRequest,
21	create_venv_pex,
22	)
23	from pants.backend.python.util_rules.pex_environment import PexEnvironment	3✔
24	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	3✔
25	from pants.engine.fs import CreateDigest, FileContent, FileEntry	3✔
26	from pants.engine.intrinsics import create_digest, execute_process	3✔
27	from pants.engine.rules import collect_rules, implicitly, rule	3✔
28	from pants.option.option_types import DictOption	3✔
29	from pants.util.logging import LogLevel	3✔
30	from pants.util.strutil import pluralize, softwrap	3✔
31
32	logger = logging.getLogger(__name__)	3✔
33
34	_HELM_K8S_PARSER_SOURCE = "k8s_parser_main.py"	3✔
35	_HELM_K8S_PARSER_PACKAGE = "pants.backend.helm.subsystems"	3✔
36
37
38	class HelmKubeParserSubsystem(PythonToolRequirementsBase):	3✔
39	options_scope = "helm-k8s-parser"	3✔
40	help_short = "Analyses K8S manifests rendered by Helm."	3✔
41
42	default_requirements = [	3✔
43	"hikaru>=1.1.0",
44	"hikaru-model-28",
45	"hikaru-model-27",
46	"hikaru-model-26",
47	"hikaru-model-25",
48	"hikaru-model-24",
49	"hikaru-model-23",
50	]
51
52	register_interpreter_constraints = True	3✔
53	crd = DictOption[str](	3✔
54	help=softwrap(
55	"""
56	Additional custom resource definitions be made available to all Helm processes
57	or during value interpolation.
58	Example:
59	[helm-k8s-parser.crd]
60	"filename1"="classname1"
61	"filename2"="classname2"
62	"""
63	),
64	default={},
65	)
66
67	default_lockfile_resource = (_HELM_K8S_PARSER_PACKAGE, "k8s_parser.lock")	3✔
68
69
70	@dataclass(frozen=True)	3✔
71	class _HelmKubeParserTool:	3✔
72	pex: VenvPex	3✔
73	crd: str	3✔
74
75
76	@rule	3✔
77	async def build_k8s_parser_tool(	3✔
78	k8s_parser: HelmKubeParserSubsystem,
79	pex_environment: PexEnvironment,
80	) -> _HelmKubeParserTool:
81	parser_sources = pkgutil.get_data(_HELM_K8S_PARSER_PACKAGE, _HELM_K8S_PARSER_SOURCE)	3✔
82
83	if not parser_sources:	3✔
84	raise ValueError(	×
85	f"Unable to find source to {_HELM_K8S_PARSER_SOURCE!r} in {_HELM_K8S_PARSER_PACKAGE}"
86	)
87
88	parser_file_content = FileContent(	3✔
89	path="__k8s_parser.py", content=parser_sources, is_executable=True
90	)
91
92	digest_sources = [parser_file_content]	3✔
93
94	modulename_classname = []	3✔
95	if k8s_parser.crd != {}:	3✔
UNCOV 96	for file, classname in k8s_parser.crd.items():	×
UNCOV 97	crd_sources = open(file, "rb").read()	×
UNCOV 98	if not crd_sources:	×
99	raise ValueError(	×
100	f"Unable to find source to customer resource definition in {_HELM_K8S_PARSER_PACKAGE}"
101	)
UNCOV 102	unique_name = f"_crd_source_{hash(file)}"	×
UNCOV 103	parser_file_content_source = FileContent(	×
104	path=f"{unique_name}.py", content=crd_sources, is_executable=False
105	)
UNCOV 106	digest_sources.append(parser_file_content_source)	×
UNCOV 107	modulename_classname.append((unique_name, classname))	×
108	parser_digest = await create_digest(CreateDigest(digest_sources))	3✔
109
110	# We use copies of site packages because hikaru gets confused with symlinked packages
111	# The core hikaru package tries to load the packages containing the kubernetes-versioned models
112	# using the __path__ attribute of the core package,
113	# which doesn't work when the packages are symlinked from inside the namespace-handling dirs in the PEX
114	use_site_packages_copies = True	3✔
115
116	parser_pex = await create_venv_pex(	3✔
117	VenvPexRequest(
118	k8s_parser.to_pex_request(
119	main=EntryPoint(PurePath(parser_file_content.path).stem),
120	sources=parser_digest,
121	),
122	pex_environment.in_sandbox(working_directory=None),
123	site_packages_copies=use_site_packages_copies,
124	),
125	**implicitly(),
126	)
127	return _HelmKubeParserTool(parser_pex, json.dumps(modulename_classname))	3✔
128
129
130	@dataclass(frozen=True)	3✔
131	class ParseKubeManifestRequest(EngineAwareParameter):	3✔
132	file: FileEntry	3✔
133
134	def debug_hint(self) -> str \| None:	3✔
UNCOV 135	return self.file.path	×
136
137	def metadata(self) -> dict[str, Any] \| None:	3✔
138	return {"file": self.file}	×
139
140
141	@dataclass(frozen=True)	3✔
142	class ParsedImageRefEntry:	3✔
143	document_index: int	3✔
144	path: YamlPath	3✔
145	unparsed_image_ref: str	3✔
146
147
148	@dataclass(frozen=True)	3✔
149	class ParsedKubeManifest(EngineAwareReturnType):	3✔
150	filename: str	3✔
151	found_image_refs: tuple[ParsedImageRefEntry, ...]	3✔
152
153	def level(self) -> LogLevel \| None:	3✔
154	return LogLevel.DEBUG	3✔
155
156	def message(self) -> str \| None:	3✔
157	return f"Found {pluralize(len(self.found_image_refs), 'image reference')} in file {self.filename}"	3✔
158
159	def metadata(self) -> dict[str, Any] \| None:	3✔
160	return {	3✔
161	"filename": self.filename,
162	"found_image_refs": self.found_image_refs,
163	}
164
165
166	@rule(desc="Parse Kubernetes resource manifest")	3✔
167	async def parse_kube_manifest(	3✔
168	request: ParseKubeManifestRequest, tool: _HelmKubeParserTool
169	) -> ParsedKubeManifest:
170	file_digest = await create_digest(CreateDigest([request.file]))	3✔
171
172	result = await execute_process(	3✔
173	**implicitly(
174	VenvPexProcess(
175	tool.pex,
176	argv=[request.file.path, tool.crd],
177	input_digest=file_digest,
178	description=f"Analyzing Kubernetes manifest {request.file.path}",
179	level=LogLevel.DEBUG,
180	)
181	)
182	)
183
184	if result.exit_code == 0:	3✔
185	output = result.stdout.decode("utf-8").splitlines()	3✔
186	image_refs: list[ParsedImageRefEntry] = []	3✔
187	for line in output:	3✔
188	parts = line.split(",")	2✔
189	if len(parts) != 3:	2✔
190	raise Exception(	×
191	softwrap(
192	f"""Unexpected output from k8s parser when parsing file {request.file.path}:
193
194	{line}
195	"""
196	)
197	)
198
199	image_refs.append(	2✔
200	ParsedImageRefEntry(
201	document_index=int(parts[0]),
202	path=YamlPath.parse(parts[1]),
203	unparsed_image_ref=parts[2],
204	)
205	)
206
207	return ParsedKubeManifest(filename=request.file.path, found_image_refs=tuple(image_refs))	3✔
208	else:
UNCOV 209	parser_error = result.stderr.decode("utf-8")	×
UNCOV 210	raise Exception(	×
211	softwrap(
212	f"""
213	Could not parse Kubernetes manifests in file: {request.file.path}.
214	{parser_error}
215	"""
216	)
217	)
218
219
220	def rules():	3✔
221	return [	3✔
222	*collect_rules(),
223	*pex.rules(),
224	]

pantsbuild / pants / 22285099215

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