• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 18913303678

29 Oct 2025 03:29PM UTC coverage: 80.004% (-0.3%) from 80.283%
18913303678

push

github

web-flow
Updating 3rd party lockfiles for PBS script and MyPy (#22796)

Also small bumps to fastapi and starlette, which updated over the last few days. Starlette updated for features, and then a security vulnerability.

Fastapi bumped just for starlette.

88 of 93 new or added lines in 73 files covered. (94.62%)

221 existing lines in 15 files now uncovered.

77334 of 96663 relevant lines covered (80.0%)

5.2 hits per line

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

0.0
/src/python/pants/backend/helm/goals/publish.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 logging
×
UNCOV
7
from dataclasses import dataclass
×
8

UNCOV
9
from pants.backend.helm.goals.package import BuiltHelmArtifact
×
UNCOV
10
from pants.backend.helm.subsystems.helm import HelmSubsystem
×
UNCOV
11
from pants.backend.helm.target_types import (
×
12
    HelmChartFieldSet,
13
    HelmChartRepositoryField,
14
    HelmRegistriesField,
15
    HelmSkipPushField,
16
)
UNCOV
17
from pants.backend.helm.util_rules.tool import HelmProcess, helm_process
×
UNCOV
18
from pants.core.goals.publish import (
×
19
    PublishFieldSet,
20
    PublishOutputData,
21
    PublishPackages,
22
    PublishProcesses,
23
    PublishRequest,
24
)
UNCOV
25
from pants.engine.process import InteractiveProcess
×
UNCOV
26
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
×
UNCOV
27
from pants.util.logging import LogLevel
×
28

UNCOV
29
logger = logging.getLogger(__name__)
×
30

31

UNCOV
32
class PublishHelmChartRequest(PublishRequest):
×
UNCOV
33
    pass
×
34

35

UNCOV
36
@dataclass(frozen=True)
×
UNCOV
37
class HelmPublishFieldSet(HelmChartFieldSet, PublishFieldSet):
×
UNCOV
38
    publish_request_type = PublishHelmChartRequest
×
39

UNCOV
40
    registries: HelmRegistriesField
×
UNCOV
41
    repository: HelmChartRepositoryField
×
UNCOV
42
    skip_push: HelmSkipPushField
×
43

UNCOV
44
    def get_output_data(self) -> PublishOutputData:
×
45
        return PublishOutputData(
×
46
            {
47
                "publisher": "helm",
48
                "registries": self.registries.value or (),
49
                **super().get_output_data(),
50
            }
51
        )
52

53

UNCOV
54
@rule(desc="Push Helm chart to OCI registries", level=LogLevel.DEBUG)
×
UNCOV
55
async def publish_helm_chart(
×
56
    request: PublishHelmChartRequest, helm_subsystem: HelmSubsystem
57
) -> PublishProcesses:
58
    remotes = helm_subsystem.remotes()
×
59
    built_artifacts = [
×
60
        (pkg, artifact, artifact.info)
61
        for pkg in request.packages
62
        for artifact in pkg.artifacts
63
        if isinstance(artifact, BuiltHelmArtifact) and artifact.info
64
    ]
65

66
    registries_to_push = list(remotes.get(*(request.field_set.registries.value or [])))
×
67
    if not registries_to_push:
×
68
        return PublishProcesses(
×
69
            [
70
                PublishPackages(
71
                    names=tuple(metadata.artifact_name for _, _, metadata in built_artifacts),
72
                    description=f"(by missing `{request.field_set.registries.alias}` on {request.field_set.address})",
73
                )
74
            ]
75
        )
76

77
    push_repository = (
×
78
        request.field_set.repository.value or helm_subsystem.default_registry_repository
79
    )
80
    publish_refs = [
×
81
        registry.package_ref(metadata.artifact_name, repository=push_repository)
82
        for _, _, metadata in built_artifacts
83
        for registry in registries_to_push
84
    ]
85
    if request.field_set.skip_push.value:
×
86
        return PublishProcesses(
×
87
            [
88
                PublishPackages(
89
                    names=tuple(publish_refs),
90
                    description=f"(by `{request.field_set.skip_push.alias}` on {request.field_set.address})",
91
                )
92
            ]
93
        )
94

95
    processes = await concurrently(
×
96
        helm_process(
97
            HelmProcess(
98
                ["push", artifact.relpath, registry.repository_ref(push_repository)],
99
                input_digest=pkg.digest,
100
                description=f"Pushing Helm chart '{metadata.name}' with version '{metadata.version}' into OCI registry: {registry.address}",
101
            ),
102
            **implicitly(),
103
        )
104
        for pkg, artifact, metadata in built_artifacts
105
        if artifact.relpath
106
        for registry in registries_to_push
107
    )
108

109
    return PublishProcesses(
×
110
        [
111
            PublishPackages(names=(package_ref,), process=InteractiveProcess.from_process(process))
112
            for package_ref, process in zip(publish_refs, processes)
113
        ]
114
    )
115

116

UNCOV
117
def rules():
×
UNCOV
118
    return [*collect_rules(), *HelmPublishFieldSet.rules()]
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc