18847018991

Committed 27 Oct 2025 03:45PM UTC coverage: 92.254% (+12.0%) from 80.282%

Build # 18847018991

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge f1312fa87 into 06e216752

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

Run Details

39 of 40 new or added lines in 11 files covered. (97.5%)

382 existing lines in 22 files now uncovered.

89230 of 96722 relevant lines covered (92.25%)

3.72 hits per line

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

95.65

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

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

from __future__ import annotations

from textwrap import dedent
from typing import cast

import pytest

from pants.backend.helm.subsystems import k8s_parser
from pants.backend.helm.subsystems.k8s_parser import (
    ParsedImageRefEntry,
    ParsedKubeManifest,
    ParseKubeManifestRequest,
)
from pants.backend.helm.testutil import K8S_CRD_FILE_IMAGE, K8S_POD_FILE
from pants.backend.helm.utils.yaml import YamlPath
from pants.engine.fs import CreateDigest, Digest, DigestEntries, FileContent, FileEntry
from pants.engine.rules import QueryRule
from pants.testutil.pants_integration_test import setup_tmpdir
from pants.testutil.rule_runner import PYTHON_BOOTSTRAP_ENV, RuleRunner


@pytest.fixture
def rule_runner() -> RuleRunner:
    rule_runner = RuleRunner(
        rules=[
            *k8s_parser.rules(),
            QueryRule(ParsedKubeManifest, (ParseKubeManifestRequest,)),
            QueryRule(DigestEntries, (Digest,)),
        ]
    )
    rule_runner.set_options(
        [],
        env_inherit=PYTHON_BOOTSTRAP_ENV,
    )
    return rule_runner


def test_parser_can_run(rule_runner: RuleRunner) -> None:
    file_digest = rule_runner.request(
        Digest, [CreateDigest([FileContent("pod.yaml", K8S_POD_FILE.encode("utf-8"))])]
    )
    file_entries = rule_runner.request(DigestEntries, [file_digest])

    parsed_manifest = rule_runner.request(
        ParsedKubeManifest,
        [ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))],
    )

    expected_image_refs = [
        ParsedImageRefEntry(0, YamlPath.parse("/spec/containers/0/image"), "busybox:1.28"),
        ParsedImageRefEntry(0, YamlPath.parse("/spec/initContainers/0/image"), "busybox:1.29"),
    ]

    assert parsed_manifest.found_image_refs == tuple(expected_image_refs)


def test_parser_returns_no_image_refs(rule_runner: RuleRunner) -> None:
    config_map_contents = dedent(
        """\
        apiVersion: v1
        kind: ConfigMap
        metadata:
          name: foo
        data:
          key: value
        """
    )

    file_digest = rule_runner.request(
        Digest,
        [CreateDigest([FileContent("config_map.yaml", config_map_contents.encode("utf-8"))])],
    )
    file_entries = rule_runner.request(DigestEntries, [file_digest])

    parsed_manifest = rule_runner.request(
        ParsedKubeManifest,
        [ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))],
    )

    assert len(parsed_manifest.found_image_refs) == 0


def test_crd_parser_can_run(rule_runner: RuleRunner) -> None:
    file_digest_python = {
        "crd.py": dedent(
            """

        from __future__ import annotations
        from hikaru.model.rel_1_28.v1 import *

        from hikaru import (HikaruBase, HikaruDocumentBase,
                            set_default_release)
        from hikaru.crd import HikaruCRDDocumentMixin
        from typing import Optional, List
        from dataclasses import dataclass

        set_default_release("rel_1_28")

        @dataclass
        class ContainersSpec(HikaruBase):
            containers: List[Container]
            initContainers: List[Container]

        @dataclass
        class CRD(HikaruDocumentBase, HikaruCRDDocumentMixin):
            spec: ContainersSpec
            metadata: ObjectMeta
            apiVersion: str = f"pants/v1alpha1"
            kind: str = "CustomResourceDefinition"

    """
        )
    }

    file_digest = rule_runner.request(
        Digest, [CreateDigest([FileContent("crd.yaml", K8S_CRD_FILE_IMAGE.encode("utf-8"))])]
    )
    file_entries = rule_runner.request(DigestEntries, [file_digest])
    with setup_tmpdir(file_digest_python) as tmpdir:
        rule_runner.set_options([f'--helm-k8s-parser-crd={{"{tmpdir}/crd.py": "CRD"}}'])
        parsed_manifest = rule_runner.request(
            ParsedKubeManifest, [ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))]
        )

    expected_image_refs = [
        ParsedImageRefEntry(0, YamlPath.parse("/spec/containers/0/image"), "busybox:1.28"),
        ParsedImageRefEntry(0, YamlPath.parse("/spec/initContainers/0/image"), "busybox:1.29"),
    ]
    assert parsed_manifest.found_image_refs == tuple(expected_image_refs)


def test_crd_parser_class_not_found_error(rule_runner: RuleRunner) -> None:
    file_digest_python = {
        "crd.py": dedent(
            """

        from __future__ import annotations
        from hikaru.model.rel_1_28.v1 import *

        from hikaru import (HikaruBase, HikaruDocumentBase,
                            set_default_release)
        from hikaru.crd import HikaruCRDDocumentMixin
        from typing import Optional, List
        from dataclasses import dataclass

        set_default_release("rel_1_28")

        @dataclass
        class ContainersSpec(HikaruBase):
            containers: List[Container]
            initContainers: List[Container]

        @dataclass
        class CustomResourceDefinition(HikaruDocumentBase, HikaruCRDDocumentMixin):
            spec: ContainersSpec
            metadata: ObjectMeta
            apiVersion: str = f"pants/v1alpha1"
            kind: str = "CustomResourceDefinition"

    """
        )
    }

    file_digest = rule_runner.request(
        Digest, [CreateDigest([FileContent("crd.yaml", K8S_CRD_FILE_IMAGE.encode("utf-8"))])]
    )
    file_entries = rule_runner.request(DigestEntries, [file_digest])
    with setup_tmpdir(file_digest_python) as tmpdir:
        rule_runner.set_options([f'--helm-k8s-parser-crd={{"{tmpdir}/crd.py": "CRD"}}'])
        with pytest.raises(Exception):
            rule_runner.request(
                ParsedKubeManifest,
                [ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))],
            )

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	1✔
5
6	from textwrap import dedent	1✔
7	from typing import cast	1✔
8
9	import pytest	1✔
10
11	from pants.backend.helm.subsystems import k8s_parser	1✔
12	from pants.backend.helm.subsystems.k8s_parser import (	1✔
13	ParsedImageRefEntry,
14	ParsedKubeManifest,
15	ParseKubeManifestRequest,
16	)
17	from pants.backend.helm.testutil import K8S_CRD_FILE_IMAGE, K8S_POD_FILE	1✔
18	from pants.backend.helm.utils.yaml import YamlPath	1✔
19	from pants.engine.fs import CreateDigest, Digest, DigestEntries, FileContent, FileEntry	1✔
20	from pants.engine.rules import QueryRule	1✔
21	from pants.testutil.pants_integration_test import setup_tmpdir	1✔
22	from pants.testutil.rule_runner import PYTHON_BOOTSTRAP_ENV, RuleRunner	1✔
23
24
25	@pytest.fixture	1✔
26	def rule_runner() -> RuleRunner:	1✔
27	rule_runner = RuleRunner(	1✔
28	rules=[
29	*k8s_parser.rules(),
30	QueryRule(ParsedKubeManifest, (ParseKubeManifestRequest,)),
31	QueryRule(DigestEntries, (Digest,)),
32	]
33	)
34	rule_runner.set_options(	1✔
35	[],
36	env_inherit=PYTHON_BOOTSTRAP_ENV,
37	)
38	return rule_runner	1✔
39
40
41	def test_parser_can_run(rule_runner: RuleRunner) -> None:	1✔
42	file_digest = rule_runner.request(	1✔
43	Digest, [CreateDigest([FileContent("pod.yaml", K8S_POD_FILE.encode("utf-8"))])]
44	)
45	file_entries = rule_runner.request(DigestEntries, [file_digest])	1✔
46
47	parsed_manifest = rule_runner.request(	1✔
48	ParsedKubeManifest,
49	[ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))],
50	)
51
52	expected_image_refs = [	1✔
53	ParsedImageRefEntry(0, YamlPath.parse("/spec/containers/0/image"), "busybox:1.28"),
54	ParsedImageRefEntry(0, YamlPath.parse("/spec/initContainers/0/image"), "busybox:1.29"),
55	]
56
57	assert parsed_manifest.found_image_refs == tuple(expected_image_refs)	1✔
58
59
60	def test_parser_returns_no_image_refs(rule_runner: RuleRunner) -> None:	1✔
61	config_map_contents = dedent(	1✔
62	"""\
63	apiVersion: v1
64	kind: ConfigMap
65	metadata:
66	name: foo
67	data:
68	key: value
69	"""
70	)
71
72	file_digest = rule_runner.request(	1✔
73	Digest,
74	[CreateDigest([FileContent("config_map.yaml", config_map_contents.encode("utf-8"))])],
75	)
76	file_entries = rule_runner.request(DigestEntries, [file_digest])	1✔
77
78	parsed_manifest = rule_runner.request(	1✔
79	ParsedKubeManifest,
80	[ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))],
81	)
82
83	assert len(parsed_manifest.found_image_refs) == 0	1✔
84
85
86	def test_crd_parser_can_run(rule_runner: RuleRunner) -> None:	1✔
87	file_digest_python = {	1✔
88	"crd.py": dedent(
89	"""
90
91	from __future__ import annotations
92	from hikaru.model.rel_1_28.v1 import *
93
94	from hikaru import (HikaruBase, HikaruDocumentBase,
95	set_default_release)
96	from hikaru.crd import HikaruCRDDocumentMixin
97	from typing import Optional, List
98	from dataclasses import dataclass
99
100	set_default_release("rel_1_28")
101
102	@dataclass
103	class ContainersSpec(HikaruBase):
104	containers: List[Container]
105	initContainers: List[Container]
106
107	@dataclass
108	class CRD(HikaruDocumentBase, HikaruCRDDocumentMixin):
109	spec: ContainersSpec
110	metadata: ObjectMeta
111	apiVersion: str = f"pants/v1alpha1"
112	kind: str = "CustomResourceDefinition"
113
114	"""
115	)
116	}
117
118	file_digest = rule_runner.request(	1✔
119	Digest, [CreateDigest([FileContent("crd.yaml", K8S_CRD_FILE_IMAGE.encode("utf-8"))])]
120	)
121	file_entries = rule_runner.request(DigestEntries, [file_digest])	1✔
122	with setup_tmpdir(file_digest_python) as tmpdir:	1✔
123	rule_runner.set_options([f'--helm-k8s-parser-crd={{"{tmpdir}/crd.py": "CRD"}}'])	1✔
124	parsed_manifest = rule_runner.request(	1✔
125	ParsedKubeManifest, [ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))]
126	)
127
UNCOV 128	expected_image_refs = [	×
129	ParsedImageRefEntry(0, YamlPath.parse("/spec/containers/0/image"), "busybox:1.28"),
130	ParsedImageRefEntry(0, YamlPath.parse("/spec/initContainers/0/image"), "busybox:1.29"),
131	]
UNCOV 132	assert parsed_manifest.found_image_refs == tuple(expected_image_refs)	×
133
134
135	def test_crd_parser_class_not_found_error(rule_runner: RuleRunner) -> None:	1✔
136	file_digest_python = {	1✔
137	"crd.py": dedent(
138	"""
139
140	from __future__ import annotations
141	from hikaru.model.rel_1_28.v1 import *
142
143	from hikaru import (HikaruBase, HikaruDocumentBase,
144	set_default_release)
145	from hikaru.crd import HikaruCRDDocumentMixin
146	from typing import Optional, List
147	from dataclasses import dataclass
148
149	set_default_release("rel_1_28")
150
151	@dataclass
152	class ContainersSpec(HikaruBase):
153	containers: List[Container]
154	initContainers: List[Container]
155
156	@dataclass
157	class CustomResourceDefinition(HikaruDocumentBase, HikaruCRDDocumentMixin):
158	spec: ContainersSpec
159	metadata: ObjectMeta
160	apiVersion: str = f"pants/v1alpha1"
161	kind: str = "CustomResourceDefinition"
162
163	"""
164	)
165	}
166
167	file_digest = rule_runner.request(	1✔
168	Digest, [CreateDigest([FileContent("crd.yaml", K8S_CRD_FILE_IMAGE.encode("utf-8"))])]
169	)
170	file_entries = rule_runner.request(DigestEntries, [file_digest])	1✔
171	with setup_tmpdir(file_digest_python) as tmpdir:	1✔
172	rule_runner.set_options([f'--helm-k8s-parser-crd={{"{tmpdir}/crd.py": "CRD"}}'])	1✔
173	with pytest.raises(Exception):	1✔
174	rule_runner.request(	1✔
175	ParsedKubeManifest,
176	[ParseKubeManifestRequest(file=cast(FileEntry, file_entries[0]))],
177	)

pantsbuild / pants / 18847018991

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