24055979590

Committed 06 Apr 2026 11:17PM UTC coverage: 52.37% (-40.5%) from 92.908%

Build # 24055979590

Build Type

Pull #23225

github

Committed by

web-flow

Commit Message

Merge 67474653c into 542ca048d

Pull Request Pull Request #23225: Add --test-show-all-batch-targets to expose all targets in batched pytest

Coverage Stats

6 of 17 new or added lines in 2 files covered. (35.29%)

23030 existing lines in 605 files now uncovered.

31643 of 60422 relevant lines covered (52.37%)

1.05 hits per line

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

81.91

/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

# pants: infer-dep(k8s_parser.lock*)

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

pantsbuild / pants / 24055979590

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