19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0

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

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	from __future__ import annotations	×
5
UNCOV 6	import json	×
UNCOV 7	import logging	×
UNCOV 8	import pkgutil	×
UNCOV 9	from dataclasses import dataclass	×
UNCOV 10	from pathlib import PurePath	×
UNCOV 11	from typing import Any	×
12
UNCOV 13	from pants.backend.helm.utils.yaml import YamlPath	×
UNCOV 14	from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase	×
UNCOV 15	from pants.backend.python.target_types import EntryPoint	×
UNCOV 16	from pants.backend.python.util_rules import pex	×
UNCOV 17	from pants.backend.python.util_rules.pex import (	×
18	VenvPex,
19	VenvPexProcess,
20	VenvPexRequest,
21	create_venv_pex,
22	)
UNCOV 23	from pants.backend.python.util_rules.pex_environment import PexEnvironment	×
UNCOV 24	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	×
UNCOV 25	from pants.engine.fs import CreateDigest, FileContent, FileEntry	×
UNCOV 26	from pants.engine.intrinsics import create_digest, execute_process	×
UNCOV 27	from pants.engine.rules import collect_rules, implicitly, rule	×
UNCOV 28	from pants.option.option_types import DictOption	×
UNCOV 29	from pants.util.logging import LogLevel	×
UNCOV 30	from pants.util.strutil import pluralize, softwrap	×
31
UNCOV 32	logger = logging.getLogger(__name__)	×
33
UNCOV 34	_HELM_K8S_PARSER_SOURCE = "k8s_parser_main.py"	×
UNCOV 35	_HELM_K8S_PARSER_PACKAGE = "pants.backend.helm.subsystems"	×
36
37
UNCOV 38	class HelmKubeParserSubsystem(PythonToolRequirementsBase):	×
UNCOV 39	options_scope = "helm-k8s-parser"	×
UNCOV 40	help_short = "Analyses K8S manifests rendered by Helm."	×
41
UNCOV 42	default_requirements = [	×
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
UNCOV 52	register_interpreter_constraints = True	×
UNCOV 53	crd = DictOption[str](	×
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
UNCOV 67	default_lockfile_resource = (_HELM_K8S_PARSER_PACKAGE, "k8s_parser.lock")	×
68
69
UNCOV 70	@dataclass(frozen=True)	×
UNCOV 71	class _HelmKubeParserTool:	×
UNCOV 72	pex: VenvPex	×
UNCOV 73	crd: str	×
74
75
UNCOV 76	@rule	×
UNCOV 77	async def build_k8s_parser_tool(	×
78	k8s_parser: HelmKubeParserSubsystem,
79	pex_environment: PexEnvironment,
80	) -> _HelmKubeParserTool:
81	parser_sources = pkgutil.get_data(_HELM_K8S_PARSER_PACKAGE, _HELM_K8S_PARSER_SOURCE)	×
82
83	if not parser_sources:	×
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(	×
89	path="__k8s_parser.py", content=parser_sources, is_executable=True
90	)
91
92	digest_sources = [parser_file_content]	×
93
94	modulename_classname = []	×
95	if k8s_parser.crd != {}:	×
96	for file, classname in k8s_parser.crd.items():	×
97	crd_sources = open(file, "rb").read()	×
98	if not crd_sources:	×
99	raise ValueError(	×
100	f"Unable to find source to customer resource definition in {_HELM_K8S_PARSER_PACKAGE}"
101	)
102	unique_name = f"_crd_source_{hash(file)}"	×
103	parser_file_content_source = FileContent(	×
104	path=f"{unique_name}.py", content=crd_sources, is_executable=False
105	)
106	digest_sources.append(parser_file_content_source)	×
107	modulename_classname.append((unique_name, classname))	×
108	parser_digest = await create_digest(CreateDigest(digest_sources))	×
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	×
115
116	parser_pex = await create_venv_pex(	×
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))	×
128
129
UNCOV 130	@dataclass(frozen=True)	×
UNCOV 131	class ParseKubeManifestRequest(EngineAwareParameter):	×
UNCOV 132	file: FileEntry	×
133
UNCOV 134	def debug_hint(self) -> str \| None:	×
135	return self.file.path	×
136
UNCOV 137	def metadata(self) -> dict[str, Any] \| None:	×
138	return {"file": self.file}	×
139
140
UNCOV 141	@dataclass(frozen=True)	×
UNCOV 142	class ParsedImageRefEntry:	×
UNCOV 143	document_index: int	×
UNCOV 144	path: YamlPath	×
UNCOV 145	unparsed_image_ref: str	×
146
147
UNCOV 148	@dataclass(frozen=True)	×
UNCOV 149	class ParsedKubeManifest(EngineAwareReturnType):	×
UNCOV 150	filename: str	×
UNCOV 151	found_image_refs: tuple[ParsedImageRefEntry, ...]	×
152
UNCOV 153	def level(self) -> LogLevel \| None:	×
154	return LogLevel.DEBUG	×
155
UNCOV 156	def message(self) -> str \| None:	×
157	return f"Found {pluralize(len(self.found_image_refs), 'image reference')} in file {self.filename}"	×
158
UNCOV 159	def metadata(self) -> dict[str, Any] \| None:	×
160	return {	×
161	"filename": self.filename,
162	"found_image_refs": self.found_image_refs,
163	}
164
165
UNCOV 166	@rule(desc="Parse Kubernetes resource manifest")	×
UNCOV 167	async def parse_kube_manifest(	×
168	request: ParseKubeManifestRequest, tool: _HelmKubeParserTool
169	) -> ParsedKubeManifest:
170	file_digest = await create_digest(CreateDigest([request.file]))	×
171
172	result = await execute_process(	×
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:	×
185	output = result.stdout.decode("utf-8").splitlines()	×
186	image_refs: list[ParsedImageRefEntry] = []	×
187	for line in output:	×
188	parts = line.split(",")	×
189	if len(parts) != 3:	×
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(	×
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))	×
208	else:
209	parser_error = result.stderr.decode("utf-8")	×
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
UNCOV 220	def rules():	×
UNCOV 221	return [	×
222	*collect_rules(),
223	*pex.rules(),
224	]

pantsbuild / pants / 19015773527

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