21374897774

Committed 26 Jan 2026 09:37PM UTC coverage: 80.008% (-0.3%) from 80.269%

Build # 21374897774

Build Type

Pull #23037

github

Committed by

web-flow

Commit Message

Merge 4023b9eee into 09b8ecaa1

Pull Request Pull Request #23037: Enable publish without package 2

Run Details

105 of 178 new or added lines in 11 files covered. (58.99%)

238 existing lines in 14 files now uncovered.

78628 of 98275 relevant lines covered (80.01%)

3.35 hits per line

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

93.51

/src/python/pants/backend/python/goals/publish_test.py

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

from __future__ import annotations

from collections.abc import Callable
from textwrap import dedent

import pytest

from pants.backend.python.goals.publish import (
    PublishPythonPackageFieldSet,
    PublishPythonPackageRequest,
    rules,
)
from pants.backend.python.macros.python_artifact import PythonArtifact
from pants.backend.python.subsystems.setuptools import PythonDistributionFieldSet
from pants.backend.python.target_types import PythonDistribution, PythonSourcesGeneratorTarget
from pants.backend.python.util_rules import pex_from_targets
from pants.core.goals.package import BuiltPackage, BuiltPackageArtifact
from pants.core.goals.publish import PublishPackages, PublishProcesses
from pants.core.util_rules.config_files import rules as config_files_rules
from pants.engine.addresses import Address
from pants.engine.fs import EMPTY_DIGEST
from pants.engine.process import InteractiveProcess, Process
from pants.testutil.process_util import process_assertion
from pants.testutil.python_rule_runner import PythonRuleRunner
from pants.testutil.rule_runner import QueryRule
from pants.util.frozendict import FrozenDict


@pytest.fixture
def rule_runner() -> PythonRuleRunner:
    rule_runner = PythonRuleRunner(
        preserve_tmpdirs=True,
        rules=[
            *config_files_rules(),
            *pex_from_targets.rules(),
            *rules(),
            QueryRule(PublishProcesses, [PublishPythonPackageRequest]),
        ],
        target_types=[PythonSourcesGeneratorTarget, PythonDistribution],
        objects={"python_artifact": PythonArtifact},
    )
    return set_options(rule_runner)


def set_options(rule_runner: PythonRuleRunner, options: list | None = None) -> PythonRuleRunner:
    rule_runner.set_options(
        options or [],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
        env={
            "TWINE_USERNAME": "whoami",
            "TWINE_USERNAME_PYPI": "whoareyou",
            "TWINE_PASSWORD_PYPI": "secret",
        },
    )
    return rule_runner


@pytest.fixture
def packages():
    return (
        BuiltPackage(
            EMPTY_DIGEST,
            (
                BuiltPackageArtifact("my-package-0.1.0.tar.gz"),
                BuiltPackageArtifact("my_package-0.1.0-py3-none-any.whl"),
            ),
        ),
    )


def project_files(
    skip_twine: bool = False, repositories: list[str] = ["@pypi", "@private"]
) -> dict[str, str]:
    return {
        "src/BUILD": dedent(
            f"""\
            python_sources()
            python_distribution(
              name="dist",
              provides=python_artifact(
                name="my-package",
                version="0.1.0",
              ),
              repositories={repositories!r},
              skip_twine={skip_twine},
            )
            """
        ),
        "src/hello.py": """print("hello")""",
        ".pypirc": "",
    }


def request_publish_processes(rule_runner: PythonRuleRunner, packages) -> PublishProcesses:
    tgt = rule_runner.get_target(Address("src", target_name="dist"))
    fs = PublishPythonPackageFieldSet.create(tgt)
    return rule_runner.request(PublishProcesses, [fs._request(packages)])


def assert_package(
    package: PublishPackages,
    expect_names: tuple[str, ...],
    expect_description: str,
    expect_process: Callable[[Process], None] | None,
) -> None:
    assert package.names == expect_names
    assert package.description == expect_description
    if expect_process:
        assert package.process
        assert isinstance(package.process, InteractiveProcess)
        expect_process(package.process.process)
    else:
        assert package.process is None


def test_twine_upload(rule_runner, packages) -> None:
    rule_runner.write_files(project_files(skip_twine=False))
    result = request_publish_processes(rule_runner, packages)

    assert len(result) == 2
    assert_package(
        result[0],
        expect_names=(
            "my-package-0.1.0.tar.gz",
            "my_package-0.1.0-py3-none-any.whl",
        ),
        expect_description="@pypi",
        expect_process=process_assertion(
            argv=(
                "./twine.pex_pex_shim.sh",
                "upload",
                "--non-interactive",
                "--config-file=.pypirc",
                "--repository=pypi",
                "my-package-0.1.0.tar.gz",
                "my_package-0.1.0-py3-none-any.whl",
            ),
            env=FrozenDict(
                {
                    "TWINE_USERNAME": "whoareyou",
                    "TWINE_PASSWORD": "secret",
                }
            ),
        ),
    )
    assert_package(
        result[1],
        expect_names=(
            "my-package-0.1.0.tar.gz",
            "my_package-0.1.0-py3-none-any.whl",
        ),
        expect_description="@private",
        expect_process=process_assertion(
            argv=(
                "./twine.pex_pex_shim.sh",
                "upload",
                "--non-interactive",
                "--config-file=.pypirc",
                "--repository=private",
                "my-package-0.1.0.tar.gz",
                "my_package-0.1.0-py3-none-any.whl",
            ),
            env=FrozenDict({"TWINE_USERNAME": "whoami"}),
        ),
    )


def test_skip_twine(rule_runner, packages) -> None:
    rule_runner.write_files(project_files(skip_twine=True))
    result = request_publish_processes(rule_runner, packages)

    assert len(result) == 1
    assert_package(
        result[0],
        expect_names=(
            "my-package-0.1.0.tar.gz",
            "my_package-0.1.0-py3-none-any.whl",
        ),
        expect_description="(by `skip_twine` on src:dist)",
        expect_process=None,
    )

    # Skip twine globally from config option.
    rule_runner.set_options(["--twine-skip"])
    result = request_publish_processes(rule_runner, packages)
    assert len(result) == 0


@pytest.mark.parametrize(
    "options, cert_arg",
    [
        pytest.param(
            [],
            None,
            id="No ca cert",
        ),
        pytest.param(
            ["--twine-ca-certs-path={}"],
            "--cert=ca_certs.pem",
            id="[twine].ca_certs_path",
        ),
        # This test needs a working ca bundle to work. Verified manually for now.
        # pytest.param(
        #     ["--ca-certs-path={}"],
        #     "--cert=ca_certs.pem",
        #     id="[GLOBAL].ca_certs_path",
        # ),
    ],
)
def test_twine_cert_arg(rule_runner, packages, options, cert_arg) -> None:
    ca_cert_path = rule_runner.write_files({"conf/ca_certs.pem": ""})[0]
    rule_runner.write_files(project_files(repositories=["@private"]))
    set_options(rule_runner, [opt.format(ca_cert_path) for opt in options])
    result = request_publish_processes(rule_runner, packages)
    assert len(result) == 1
    process = result[0].process
    assert process
    if cert_arg:
        assert isinstance(process, InteractiveProcess)
        assert cert_arg in process.process.argv
    else:
        assert isinstance(process, InteractiveProcess)
        assert not any(arg.startswith("--cert") for arg in process.process.argv)


@pytest.mark.parametrize("skip_twine", [True, False])
def test_publish_field_set_package_before_publish(
    skip_twine: bool, rule_runner: PythonRuleRunner
) -> None:
    rule_runner.write_files(project_files(skip_twine=skip_twine))
    tgt = rule_runner.get_target(Address("src", target_name="dist"))
    fs = PublishPythonPackageFieldSet.create(tgt)
    assert fs.package_before_publish(PythonDistributionFieldSet.create(tgt)) is (not skip_twine)

1	# Copyright 2021 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 collections.abc import Callable	1✔
7	from textwrap import dedent	1✔
8
9	import pytest	1✔
10
11	from pants.backend.python.goals.publish import (	1✔
12	PublishPythonPackageFieldSet,
13	PublishPythonPackageRequest,
14	rules,
15	)
16	from pants.backend.python.macros.python_artifact import PythonArtifact	1✔
17	from pants.backend.python.subsystems.setuptools import PythonDistributionFieldSet	1✔
18	from pants.backend.python.target_types import PythonDistribution, PythonSourcesGeneratorTarget	1✔
19	from pants.backend.python.util_rules import pex_from_targets	1✔
20	from pants.core.goals.package import BuiltPackage, BuiltPackageArtifact	1✔
21	from pants.core.goals.publish import PublishPackages, PublishProcesses	1✔
22	from pants.core.util_rules.config_files import rules as config_files_rules	1✔
23	from pants.engine.addresses import Address	1✔
24	from pants.engine.fs import EMPTY_DIGEST	1✔
25	from pants.engine.process import InteractiveProcess, Process	1✔
26	from pants.testutil.process_util import process_assertion	1✔
27	from pants.testutil.python_rule_runner import PythonRuleRunner	1✔
28	from pants.testutil.rule_runner import QueryRule	1✔
29	from pants.util.frozendict import FrozenDict	1✔
30
31
32	@pytest.fixture	1✔
33	def rule_runner() -> PythonRuleRunner:	1✔
34	rule_runner = PythonRuleRunner(	1✔
35	preserve_tmpdirs=True,
36	rules=[
37	*config_files_rules(),
38	*pex_from_targets.rules(),
39	*rules(),
40	QueryRule(PublishProcesses, [PublishPythonPackageRequest]),
41	],
42	target_types=[PythonSourcesGeneratorTarget, PythonDistribution],
43	objects={"python_artifact": PythonArtifact},
44	)
45	return set_options(rule_runner)	1✔
46
47
48	def set_options(rule_runner: PythonRuleRunner, options: list \| None = None) -> PythonRuleRunner:	1✔
49	rule_runner.set_options(	1✔
50	options or [],
51	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
52	env={
53	"TWINE_USERNAME": "whoami",
54	"TWINE_USERNAME_PYPI": "whoareyou",
55	"TWINE_PASSWORD_PYPI": "secret",
56	},
57	)
58	return rule_runner	1✔
59
60
61	@pytest.fixture	1✔
62	def packages():	1✔
63	return (	1✔
64	BuiltPackage(
65	EMPTY_DIGEST,
66	(
67	BuiltPackageArtifact("my-package-0.1.0.tar.gz"),
68	BuiltPackageArtifact("my_package-0.1.0-py3-none-any.whl"),
69	),
70	),
71	)
72
73
74	def project_files(	1✔
75	skip_twine: bool = False, repositories: list[str] = ["@pypi", "@private"]
76	) -> dict[str, str]:
77	return {	1✔
78	"src/BUILD": dedent(
79	f"""\
80	python_sources()
81	python_distribution(
82	name="dist",
83	provides=python_artifact(
84	name="my-package",
85	version="0.1.0",
86	),
87	repositories={repositories!r},
88	skip_twine={skip_twine},
89	)
90	"""
91	),
92	"src/hello.py": """print("hello")""",
93	".pypirc": "",
94	}
95
96
97	def request_publish_processes(rule_runner: PythonRuleRunner, packages) -> PublishProcesses:	1✔
98	tgt = rule_runner.get_target(Address("src", target_name="dist"))	1✔
99	fs = PublishPythonPackageFieldSet.create(tgt)	1✔
100	return rule_runner.request(PublishProcesses, [fs._request(packages)])	1✔
101
102
103	def assert_package(	1✔
104	package: PublishPackages,
105	expect_names: tuple[str, ...],
106	expect_description: str,
107	expect_process: Callable[[Process], None] \| None,
108	) -> None:
109	assert package.names == expect_names	1✔
110	assert package.description == expect_description	1✔
111	if expect_process:	1✔
112	assert package.process	1✔
113	assert isinstance(package.process, InteractiveProcess)	1✔
114	expect_process(package.process.process)	1✔
115	else:
UNCOV 116	assert package.process is None	×
117
118
119	def test_twine_upload(rule_runner, packages) -> None:	1✔
120	rule_runner.write_files(project_files(skip_twine=False))	1✔
121	result = request_publish_processes(rule_runner, packages)	1✔
122
123	assert len(result) == 2	1✔
124	assert_package(	1✔
125	result[0],
126	expect_names=(
127	"my-package-0.1.0.tar.gz",
128	"my_package-0.1.0-py3-none-any.whl",
129	),
130	expect_description="@pypi",
131	expect_process=process_assertion(
132	argv=(
133	"./twine.pex_pex_shim.sh",
134	"upload",
135	"--non-interactive",
136	"--config-file=.pypirc",
137	"--repository=pypi",
138	"my-package-0.1.0.tar.gz",
139	"my_package-0.1.0-py3-none-any.whl",
140	),
141	env=FrozenDict(
142	{
143	"TWINE_USERNAME": "whoareyou",
144	"TWINE_PASSWORD": "secret",
145	}
146	),
147	),
148	)
149	assert_package(	1✔
150	result[1],
151	expect_names=(
152	"my-package-0.1.0.tar.gz",
153	"my_package-0.1.0-py3-none-any.whl",
154	),
155	expect_description="@private",
156	expect_process=process_assertion(
157	argv=(
158	"./twine.pex_pex_shim.sh",
159	"upload",
160	"--non-interactive",
161	"--config-file=.pypirc",
162	"--repository=private",
163	"my-package-0.1.0.tar.gz",
164	"my_package-0.1.0-py3-none-any.whl",
165	),
166	env=FrozenDict({"TWINE_USERNAME": "whoami"}),
167	),
168	)
169
170
171	def test_skip_twine(rule_runner, packages) -> None:	1✔
172	rule_runner.write_files(project_files(skip_twine=True))	1✔
173	result = request_publish_processes(rule_runner, packages)	1✔
174
175	assert len(result) == 1	1✔
UNCOV 176	assert_package(	×
177	result[0],
178	expect_names=(
179	"my-package-0.1.0.tar.gz",
180	"my_package-0.1.0-py3-none-any.whl",
181	),
182	expect_description="(by `skip_twine` on src:dist)",
183	expect_process=None,
184	)
185
186	# Skip twine globally from config option.
UNCOV 187	rule_runner.set_options(["--twine-skip"])	×
UNCOV 188	result = request_publish_processes(rule_runner, packages)	×
UNCOV 189	assert len(result) == 0	×
190
191
192	@pytest.mark.parametrize(	1✔
193	"options, cert_arg",
194	[
195	pytest.param(
196	[],
197	None,
198	id="No ca cert",
199	),
200	pytest.param(
201	["--twine-ca-certs-path={}"],
202	"--cert=ca_certs.pem",
203	id="[twine].ca_certs_path",
204	),
205	# This test needs a working ca bundle to work. Verified manually for now.
206	# pytest.param(
207	# ["--ca-certs-path={}"],
208	# "--cert=ca_certs.pem",
209	# id="[GLOBAL].ca_certs_path",
210	# ),
211	],
212	)
213	def test_twine_cert_arg(rule_runner, packages, options, cert_arg) -> None:	1✔
214	ca_cert_path = rule_runner.write_files({"conf/ca_certs.pem": ""})[0]	1✔
215	rule_runner.write_files(project_files(repositories=["@private"]))	1✔
216	set_options(rule_runner, [opt.format(ca_cert_path) for opt in options])	1✔
217	result = request_publish_processes(rule_runner, packages)	1✔
218	assert len(result) == 1	1✔
219	process = result[0].process	1✔
220	assert process	1✔
221	if cert_arg:	1✔
222	assert isinstance(process, InteractiveProcess)	1✔
223	assert cert_arg in process.process.argv	1✔
224	else:
225	assert isinstance(process, InteractiveProcess)	1✔
226	assert not any(arg.startswith("--cert") for arg in process.process.argv)	1✔
227
228
229	@pytest.mark.parametrize("skip_twine", [True, False])	1✔
230	def test_publish_field_set_package_before_publish(	1✔
231	skip_twine: bool, rule_runner: PythonRuleRunner
232	) -> None:
233	rule_runner.write_files(project_files(skip_twine=skip_twine))	1✔
234	tgt = rule_runner.get_target(Address("src", target_name="dist"))	1✔
235	fs = PublishPythonPackageFieldSet.create(tgt)	1✔
236	assert fs.package_before_publish(PythonDistributionFieldSet.create(tgt)) is (not skip_twine)	1✔

pantsbuild / pants / 21374897774

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