21457998286

Committed 28 Jan 2026 10:32PM UTC coverage: 80.281% (+0.01%) from 80.269%

Build # 21457998286

Build Type

Pull #23037

github

Committed by

web-flow

Commit Message

Merge 43b38d939 into 0fdb40370

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

Run Details

275 of 328 new or added lines in 13 files covered. (83.84%)

46 existing lines in 9 files now uncovered.

78960 of 98355 relevant lines covered (80.28%)

3.36 hits per line

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

98.7

/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,
    PublishPythonPackageSkipRequest,
    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, SkippedPublishPackages
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]),
            QueryRule(SkippedPublishPackages, [PublishPythonPackageSkipRequest]),
        ],
        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() -> tuple[BuiltPackage, ...]:
    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: tuple[BuiltPackage, ...]
) -> 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: PythonRuleRunner, packages: tuple[BuiltPackage, ...]) -> 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"}),
        ),
    )


@pytest.mark.parametrize("skip_twine", [True, False])
@pytest.mark.parametrize("repositories", [[], ["@pypi", "@private"]])
@pytest.mark.parametrize("skip_twine_config", [True, False])
def test_check_if_skip_upload(
    rule_runner: PythonRuleRunner,
    skip_twine: bool,
    repositories: list[str],
    skip_twine_config: bool,
) -> None:
    expected = bool(repositories) and not (skip_twine or skip_twine_config)
    rule_runner.set_options(["--twine-skip" if skip_twine_config else "--no-twine-skip"])
    rule_runner.write_files(project_files(skip_twine=skip_twine, repositories=repositories))
    tgt = rule_runner.get_target(Address("src", target_name="dist"))
    request = PublishPythonPackageSkipRequest(
        publish_fs=PublishPythonPackageFieldSet.create(tgt),
        package_fs=PythonDistributionFieldSet.create(tgt),
    )
    result = rule_runner.request(SkippedPublishPackages, [request])
    if expected:
        assert not result.inner
    else:
        assert len(result.inner) == 1
        assert result.inner[0].names == ("my-package",)


@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: PythonRuleRunner,
    packages: tuple[BuiltPackage, ...],
    options: list[str],
    cert_arg: str | None,
) -> 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)

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

pantsbuild / pants / 21457998286

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