18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

71.5

/src/python/pants/testutil/pants_integration_test.py

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

from __future__ import annotations

import errno
import glob
import os
import subprocess
import sys
from collections.abc import Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from io import BytesIO
from threading import Thread
from typing import Any, TextIO, Union, cast

import pytest
import toml

from pants.base.build_environment import get_buildroot
from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE
from pants.option.options_bootstrapper import OptionsBootstrapper
from pants.pantsd.pants_daemon_client import PantsDaemonClient
from pants.util.contextutil import temporary_dir
from pants.util.dirutil import fast_relpath, safe_file_dump, safe_mkdir, safe_open
from pants.util.osutil import Pid
from pants.util.strutil import ensure_binary

# NB: If `shell=True`, it's a single `str`.
Command = Union[str, list[str]]

# Sometimes we mix strings and bytes as keys and/or values, but in most
# cases we pass strict str->str, and we want both to typecheck.
# TODO: The complexity of this type, and the casting and # type: ignoring we have to do below,
#  is a code smell. We should use bytes everywhere, and convert lazily as needed.
Env = Union[Mapping[str, str], Mapping[bytes, bytes], Mapping[Union[str, bytes], Union[str, bytes]]]


@dataclass(frozen=True)
class PantsResult:
    command: Command
    exit_code: int
    stdout: str
    stderr: str
    workdir: str
    pid: Pid

    def _format_unexpected_error_code_msg(self, msg: str | None) -> str:
        details = [msg] if msg else []
        details.append(" ".join(self.command))
        details.append(f"exit_code: {self.exit_code}")

        def indent(content):
            return "\n\t".join(content.splitlines())

        details.append(f"stdout:\n\t{indent(self.stdout)}")
        details.append(f"stderr:\n\t{indent(self.stderr)}")
        return "\n".join(details)

    def assert_success(self, msg: str | None = None) -> None:
        assert self.exit_code == 0, self._format_unexpected_error_code_msg(msg)

    def assert_failure(self, msg: str | None = None) -> None:
        assert self.exit_code != 0, self._format_unexpected_error_code_msg(msg)


@dataclass(frozen=True)
class PantsJoinHandle:
    command: Command
    process: subprocess.Popen
    workdir: str

    # Write data to the child's stdin pipe and then close the pipe. (Copied from Python source
    # at https://github.com/python/cpython/blob/e41ec8e18b078024b02a742272e675ae39778536/Lib/subprocess.py#L1151
    # to handle the same edge cases handled by `subprocess.Popen.communicate`.)
    def _stdin_write(self, input: bytes | str | None):
        assert self.process.stdin

        if input:
            try:
                binary_input = ensure_binary(input)
                self.process.stdin.write(binary_input)
            except BrokenPipeError:
                pass  # communicate() must ignore broken pipe errors.
            except OSError as exc:
                if exc.errno == errno.EINVAL:
                    # bpo-19612, bpo-30418: On Windows, stdin.write() fails
                    # with EINVAL if the child process exited or if the child
                    # process is still running but closed the pipe.
                    pass
                else:
                    raise

        try:
            self.process.stdin.close()
        except BrokenPipeError:
            pass  # communicate() must ignore broken pipe errors.
        except OSError as exc:
            if exc.errno == errno.EINVAL:
                pass
            else:
                raise

    def join(
        self, stdin_data: bytes | str | None = None, stream_output: bool = False
    ) -> PantsResult:
        """Wait for the pants process to complete, and return a PantsResult for it."""

        def worker(in_stream: BytesIO, buffer: bytearray, out_stream: TextIO) -> None:
            while data := in_stream.read1(1024):
                buffer.extend(data)
                out_stream.write(data.decode(errors="ignore"))
                out_stream.flush()

        if stream_output:
            stdout_buffer = bytearray()
            stdout_thread = Thread(
                target=worker, args=(self.process.stdout, stdout_buffer, sys.stdout)
            )
            stdout_thread.daemon = True
            stdout_thread.start()

            stderr_buffer = bytearray()
            stderr_thread = Thread(
                target=worker, args=(self.process.stderr, stderr_buffer, sys.stderr)
            )
            stderr_thread.daemon = True
            stderr_thread.start()

            self._stdin_write(stdin_data)
            self.process.wait()
            stdout, stderr = (bytes(stdout_buffer), bytes(stderr_buffer))
        else:
            if stdin_data is not None:
                stdin_data = ensure_binary(stdin_data)
            stdout, stderr = self.process.communicate(stdin_data)

        if self.process.returncode != PANTS_SUCCEEDED_EXIT_CODE or stream_output:
            render_logs(self.workdir)

        return PantsResult(
            command=self.command,
            exit_code=self.process.returncode,
            stdout=stdout.decode(),
            stderr=stderr.decode(),
            workdir=self.workdir,
            pid=self.process.pid,
        )


def run_pants_with_workdir_without_waiting(
    command: Command,
    *,
    workdir: str,
    hermetic: bool = True,
    use_pantsd: bool = True,
    config: Mapping | None = None,
    extra_env: Env | None = None,
    shell: bool = False,
    set_pants_ignore: bool = True,
) -> PantsJoinHandle:
    args = [
        "--no-pantsrc",
        f"--pants-workdir={workdir}",
    ]
    if set_pants_ignore:
        # FIXME: For some reason, Pants's CI adds the coverage file and it is not ignored by default. Why?
        args.append("--pants-ignore=+['.coverage.*', '.python-build-standalone']")

    pantsd_in_command = "--no-pantsd" in command or "--pantsd" in command
    pantsd_in_config = config and "GLOBAL" in config and "pantsd" in config["GLOBAL"]
    if not pantsd_in_command and not pantsd_in_config:
        args.append("--pantsd" if use_pantsd else "--no-pantsd")

    if hermetic:
        args.append("--pants-config-files=[]")
        if set_pants_ignore:
            # Certain tests may be invoking `./pants test` for a pytest test with conftest discovery
            # enabled. We should ignore the root conftest.py for these cases.
            args.append("--pants-ignore=+['/conftest.py']")

    if config:
        toml_file_name = os.path.join(workdir, "pants.toml")
        with safe_open(toml_file_name, mode="w") as fp:
            fp.write(_TomlSerializer(config).serialize())
        args.append(f"--pants-config-files={toml_file_name}")

    # The python backend requires setting ICs explicitly.
    # We do this centrally here for convenience.
    if any("pants.backend.python" in arg for arg in command) and not any(
        "--python-interpreter-constraints" in arg for arg in command
    ):
        args.append("--python-interpreter-constraints=['>=3.9,<3.14']")

    pants_script = [sys.executable, "-m", "pants"]

    # Permit usage of shell=True and string-based commands to allow e.g. `./pants | head`.
    pants_command: Command
    if shell:
        assert not isinstance(command, list), "must pass command as a string when using shell=True"
        pants_command = " ".join([*pants_script, " ".join(args), command])
    else:
        pants_command = [*pants_script, *args, *command]

    # Only allow-listed entries will be included in the environment if hermetic=True. Note that
    # the env will already be fairly hermetic thanks to the v2 engine; this provides an
    # additional layer of hermiticity.
    env: dict[str | bytes, str | bytes]
    if hermetic:
        # With an empty environment, we would generally get the true underlying system default
        # encoding, which is unlikely to be what we want (it's generally ASCII, still). So we
        # explicitly set an encoding here.
        env = {"LC_ALL": "en_US.UTF-8"}
        # Apply our allowlist.
        for h in (
            "HOME",
            "PATH",  # Needed to find Python interpreters and other binaries.
        ):
            value = os.getenv(h)
            if value is not None:
                env[h] = value
        hermetic_env = os.getenv("HERMETIC_ENV")
        if hermetic_env:
            for h in hermetic_env.strip(",").split(","):
                value = os.getenv(h)
                if value is not None:
                    env[h] = value
    else:
        env = cast(dict[Union[str, bytes], Union[str, bytes]], os.environ.copy())

    env.update(PYTHONPATH=os.pathsep.join(sys.path), NO_SCIE_WARNING="1")
    if extra_env:
        env.update(cast(dict[Union[str, bytes], Union[str, bytes]], extra_env))

    # Pants command that was called from the test shouldn't have a parent.
    if "PANTS_PARENT_BUILD_ID" in env:
        del env["PANTS_PARENT_BUILD_ID"]

    return PantsJoinHandle(
        command=pants_command,
        process=subprocess.Popen(
            pants_command,
            # The type stub for the env argument is unnecessarily restrictive: it requires
            # all keys to be str or all to be bytes. But in practice Popen supports a mix,
            # which is what we pass. So we silence the typechecking error.
            env=env,  # type: ignore
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=shell,
        ),
        workdir=workdir,
    )


def run_pants_with_workdir(
    command: Command,
    *,
    workdir: str,
    hermetic: bool = True,
    use_pantsd: bool = True,
    config: Mapping | None = None,
    extra_env: Env | None = None,
    stdin_data: bytes | str | None = None,
    shell: bool = False,
    set_pants_ignore: bool = True,
    stream_output: bool = False,
) -> PantsResult:
    handle = run_pants_with_workdir_without_waiting(
        command,
        workdir=workdir,
        hermetic=hermetic,
        use_pantsd=use_pantsd,
        shell=shell,
        config=config,
        extra_env=extra_env,
        set_pants_ignore=set_pants_ignore,
    )
    return handle.join(stdin_data=stdin_data, stream_output=stream_output)


def run_pants(
    command: Command,
    *,
    hermetic: bool = True,
    use_pantsd: bool = False,
    config: Mapping | None = None,
    extra_env: Env | None = None,
    stdin_data: bytes | str | None = None,
    stream_output: bool = False,
) -> PantsResult:
    """Runs Pants in a subprocess.

    :param command: A list of command line arguments coming after `./pants`.
    :param hermetic: If hermetic, your actual `pants.toml` will not be used.
    :param use_pantsd: If True, the Pants process will use pantsd.
    :param config: Optional data for a generated TOML file. A map of <section-name> ->
        map of key -> value.
    :param extra_env: Set these env vars in the Pants process's environment.
    :param stdin_data: Make this data available to be read from the process's stdin.
    """
    with temporary_workdir() as workdir:
        return run_pants_with_workdir(
            command,
            workdir=workdir,
            hermetic=hermetic,
            use_pantsd=use_pantsd,
            config=config,
            stdin_data=stdin_data,
            extra_env=extra_env,
            stream_output=stream_output,
        )


# -----------------------------------------------------------------------------------------------
# Environment setup.
# -----------------------------------------------------------------------------------------------


@contextmanager
def setup_tmpdir(
    files: Mapping[str, str], raw_files: Mapping[str, bytes] | None = None
) -> Iterator[str]:
    """Create a temporary directory with the given files and return the tmpdir (relative to the
    build root).

    The `files` parameter is a dictionary of file paths to content. All file paths will be prefixed
    with the tmpdir. The file content can use `{tmpdir}` to have it substituted with the actual
    tmpdir via a format string.

    The `raw_files` parameter can be used to write binary files. These
    files will not go through formatting in any way.


    This is useful to set up controlled test environments, such as setting up source files and
    BUILD files.
    """

    raw_files = raw_files or {}

    with temporary_dir(root_dir=get_buildroot()) as tmpdir:
        rel_tmpdir = os.path.relpath(tmpdir, get_buildroot())
        for path, content in files.items():
            safe_file_dump(
                os.path.join(tmpdir, path), content.format(tmpdir=rel_tmpdir), makedirs=True
            )

        for path, data in raw_files.items():
            safe_file_dump(os.path.join(tmpdir, path), data, makedirs=True, mode="wb")

        yield rel_tmpdir


@contextmanager
def temporary_workdir(cleanup: bool = True) -> Iterator[str]:
    # We can hard-code '.pants.d' here because we know that will always be its value
    # in the pantsbuild/pants repo (e.g., that's what we .gitignore in that repo).
    # Grabbing the pants_workdir config would require this pants's config object,
    # which we don't have a reference to here.
    root = os.path.join(get_buildroot(), ".pants.d", "tmp")
    safe_mkdir(root)
    with temporary_dir(root_dir=root, cleanup=cleanup, suffix=".pants.d") as tmpdir:
        yield tmpdir


# -----------------------------------------------------------------------------------------------
# Pantsd and logs.
# -----------------------------------------------------------------------------------------------


def kill_daemon(pid_dir=None):
    args = ["./pants"]
    if pid_dir:
        args.append(f"--pants-subprocessdir={pid_dir}")
    pantsd_client = PantsDaemonClient(
        OptionsBootstrapper.create(env=os.environ, args=args, allow_pantsrc=False).bootstrap_options
    )
    with pantsd_client.lifecycle_lock:
        pantsd_client.terminate()


def ensure_daemon(func):
    """A decorator to assist with running tests with and without the daemon enabled."""
    return pytest.mark.parametrize("use_pantsd", [True, False])(func)


def render_logs(workdir: str) -> None:
    """Renders all potentially relevant logs from the given workdir to stdout."""
    filenames = list(glob.glob(os.path.join(workdir, "logs/exceptions*log"))) + list(
        glob.glob(os.path.join(workdir, "pants.log"))
    )
    for filename in filenames:
        rel_filename = fast_relpath(filename, workdir)
        print(f"{rel_filename} +++ ")
        for line in _read_log(filename):
            print(f"{rel_filename} >>> {line}")
        print(f"{rel_filename} --- ")


def read_pants_log(workdir: str) -> Iterator[str]:
    """Yields all lines from the pants log under the given workdir."""
    # Surface the pants log for easy viewing via pytest's `-s` (don't capture stdio) option.
    yield from _read_log(f"{workdir}/pants.log")


def _read_log(filename: str) -> Iterator[str]:
    with open(filename) as f:
        for line in f:
            yield line.rstrip()


@dataclass(frozen=True)
class _TomlSerializer:
    """Convert a dictionary of option scopes -> Python values into TOML understood by Pants.

    The constructor expects a dictionary of option scopes to their corresponding values as
    represented in Python. For example:

      {
        "GLOBAL": {
          "o1": True,
          "o2": "hello",
          "o3": [0, 1, 2],
        },
        "some-subsystem": {
          "dict_option": {
            "a": 0,
            "b": 0,
          },
        },
      }
    """

    parsed: Mapping[str, dict[str, int | float | str | bool | list | dict]]

    def normalize(self) -> dict:
        def normalize_section_value(option, option_value) -> tuple[str, Any]:
            # With TOML, we store dict values as strings (for now).
            if isinstance(option_value, dict):
                option_value = str(option_value)
            if option.endswith(".add"):
                option = option.rsplit(".", 1)[0]
                option_value = f"+{option_value!r}"
            elif option.endswith(".remove"):
                option = option.rsplit(".", 1)[0]
                option_value = f"-{option_value!r}"
            return option, option_value

        return {
            section: dict(
                normalize_section_value(option, option_value)
                for option, option_value in section_values.items()
            )
            for section, section_values in self.parsed.items()
        }

    def serialize(self) -> str:
        toml_values = self.normalize()
        return toml.dumps(toml_values)

1	# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	7✔
5
6	import errno	7✔
7	import glob	7✔
8	import os	7✔
9	import subprocess	7✔
10	import sys	7✔
11	from collections.abc import Iterator, Mapping	7✔
12	from contextlib import contextmanager	7✔
13	from dataclasses import dataclass	7✔
14	from io import BytesIO	7✔
15	from threading import Thread	7✔
16	from typing import Any, TextIO, Union, cast	7✔
17
18	import pytest	7✔
19	import toml	7✔
20
21	from pants.base.build_environment import get_buildroot	7✔
22	from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE	7✔
23	from pants.option.options_bootstrapper import OptionsBootstrapper	7✔
24	from pants.pantsd.pants_daemon_client import PantsDaemonClient	7✔
25	from pants.util.contextutil import temporary_dir	7✔
26	from pants.util.dirutil import fast_relpath, safe_file_dump, safe_mkdir, safe_open	7✔
27	from pants.util.osutil import Pid	7✔
28	from pants.util.strutil import ensure_binary	7✔
29
30	# NB: If `shell=True`, it's a single `str`.
31	Command = Union[str, list[str]]	7✔
32
33	# Sometimes we mix strings and bytes as keys and/or values, but in most
34	# cases we pass strict str->str, and we want both to typecheck.
35	# TODO: The complexity of this type, and the casting and # type: ignoring we have to do below,
36	# is a code smell. We should use bytes everywhere, and convert lazily as needed.
37	Env = Union[Mapping[str, str], Mapping[bytes, bytes], Mapping[Union[str, bytes], Union[str, bytes]]]	7✔
38
39
40	@dataclass(frozen=True)	7✔
41	class PantsResult:	7✔
42	command: Command	7✔
43	exit_code: int	7✔
44	stdout: str	7✔
45	stderr: str	7✔
46	workdir: str	7✔
47	pid: Pid	7✔
48
49	def _format_unexpected_error_code_msg(self, msg: str \| None) -> str:	7✔
50	details = [msg] if msg else []	×
51	details.append(" ".join(self.command))	×
52	details.append(f"exit_code: {self.exit_code}")	×
53
54	def indent(content):	×
55	return "\n\t".join(content.splitlines())	×
56
57	details.append(f"stdout:\n\t{indent(self.stdout)}")	×
58	details.append(f"stderr:\n\t{indent(self.stderr)}")	×
59	return "\n".join(details)	×
60
61	def assert_success(self, msg: str \| None = None) -> None:	7✔
62	assert self.exit_code == 0, self._format_unexpected_error_code_msg(msg)	7✔
63
64	def assert_failure(self, msg: str \| None = None) -> None:	7✔
65	assert self.exit_code != 0, self._format_unexpected_error_code_msg(msg)	7✔
66
67
68	@dataclass(frozen=True)	7✔
69	class PantsJoinHandle:	7✔
70	command: Command	7✔
71	process: subprocess.Popen	7✔
72	workdir: str	7✔
73
74	# Write data to the child's stdin pipe and then close the pipe. (Copied from Python source
75	# at https://github.com/python/cpython/blob/e41ec8e18b078024b02a742272e675ae39778536/Lib/subprocess.py#L1151
76	# to handle the same edge cases handled by `subprocess.Popen.communicate`.)
77	def _stdin_write(self, input: bytes \| str \| None):	7✔
78	assert self.process.stdin	×
79
80	if input:	×
81	try:	×
82	binary_input = ensure_binary(input)	×
83	self.process.stdin.write(binary_input)	×
84	except BrokenPipeError:	×
85	pass # communicate() must ignore broken pipe errors.	×
86	except OSError as exc:	×
87	if exc.errno == errno.EINVAL:	×
88	# bpo-19612, bpo-30418: On Windows, stdin.write() fails
89	# with EINVAL if the child process exited or if the child
90	# process is still running but closed the pipe.
91	pass	×
92	else:
93	raise	×
94
95	try:	×
96	self.process.stdin.close()	×
97	except BrokenPipeError:	×
98	pass # communicate() must ignore broken pipe errors.	×
99	except OSError as exc:	×
100	if exc.errno == errno.EINVAL:	×
101	pass	×
102	else:
103	raise	×
104
105	def join(	7✔
106	self, stdin_data: bytes \| str \| None = None, stream_output: bool = False
107	) -> PantsResult:
108	"""Wait for the pants process to complete, and return a PantsResult for it."""
109
110	def worker(in_stream: BytesIO, buffer: bytearray, out_stream: TextIO) -> None:	7✔
111	while data := in_stream.read1(1024):	×
112	buffer.extend(data)	×
113	out_stream.write(data.decode(errors="ignore"))	×
114	out_stream.flush()	×
115
116	if stream_output:	7✔
117	stdout_buffer = bytearray()	×
118	stdout_thread = Thread(	×
119	target=worker, args=(self.process.stdout, stdout_buffer, sys.stdout)
120	)
121	stdout_thread.daemon = True	×
122	stdout_thread.start()	×
123
124	stderr_buffer = bytearray()	×
125	stderr_thread = Thread(	×
126	target=worker, args=(self.process.stderr, stderr_buffer, sys.stderr)
127	)
128	stderr_thread.daemon = True	×
129	stderr_thread.start()	×
130
131	self._stdin_write(stdin_data)	×
132	self.process.wait()	×
133	stdout, stderr = (bytes(stdout_buffer), bytes(stderr_buffer))	×
134	else:
135	if stdin_data is not None:	7✔
136	stdin_data = ensure_binary(stdin_data)	×
137	stdout, stderr = self.process.communicate(stdin_data)	7✔
138
139	if self.process.returncode != PANTS_SUCCEEDED_EXIT_CODE or stream_output:	7✔
140	render_logs(self.workdir)	7✔
141
142	return PantsResult(	7✔
143	command=self.command,
144	exit_code=self.process.returncode,
145	stdout=stdout.decode(),
146	stderr=stderr.decode(),
147	workdir=self.workdir,
148	pid=self.process.pid,
149	)
150
151
152	def run_pants_with_workdir_without_waiting(	7✔
153	command: Command,
154	*,
155	workdir: str,
156	hermetic: bool = True,
157	use_pantsd: bool = True,
158	config: Mapping \| None = None,
159	extra_env: Env \| None = None,
160	shell: bool = False,
161	set_pants_ignore: bool = True,
162	) -> PantsJoinHandle:
163	args = [	7✔
164	"--no-pantsrc",
165	f"--pants-workdir={workdir}",
166	]
167	if set_pants_ignore:	7✔
168	# FIXME: For some reason, Pants's CI adds the coverage file and it is not ignored by default. Why?
169	args.append("--pants-ignore=+['.coverage.*', '.python-build-standalone']")	7✔
170
171	pantsd_in_command = "--no-pantsd" in command or "--pantsd" in command	7✔
172	pantsd_in_config = config and "GLOBAL" in config and "pantsd" in config["GLOBAL"]	7✔
173	if not pantsd_in_command and not pantsd_in_config:	7✔
174	args.append("--pantsd" if use_pantsd else "--no-pantsd")	7✔
175
176	if hermetic:	7✔
177	args.append("--pants-config-files=[]")	7✔
178	if set_pants_ignore:	7✔
179	# Certain tests may be invoking `./pants test` for a pytest test with conftest discovery
180	# enabled. We should ignore the root conftest.py for these cases.
181	args.append("--pants-ignore=+['/conftest.py']")	7✔
182
183	if config:	7✔
184	toml_file_name = os.path.join(workdir, "pants.toml")	5✔
185	with safe_open(toml_file_name, mode="w") as fp:	5✔
186	fp.write(_TomlSerializer(config).serialize())	5✔
187	args.append(f"--pants-config-files={toml_file_name}")	5✔
188
189	# The python backend requires setting ICs explicitly.
190	# We do this centrally here for convenience.
191	if any("pants.backend.python" in arg for arg in command) and not any(	7✔
192	"--python-interpreter-constraints" in arg for arg in command
193	):
194	args.append("--python-interpreter-constraints=['>=3.9,<3.14']")	4✔
195
196	pants_script = [sys.executable, "-m", "pants"]	7✔
197
198	# Permit usage of shell=True and string-based commands to allow e.g. `./pants \| head`.
199	pants_command: Command
200	if shell:	7✔
201	assert not isinstance(command, list), "must pass command as a string when using shell=True"	×
202	pants_command = " ".join([*pants_script, " ".join(args), command])	×
203	else:
204	pants_command = [pants_script, args, *command]	7✔
205
206	# Only allow-listed entries will be included in the environment if hermetic=True. Note that
207	# the env will already be fairly hermetic thanks to the v2 engine; this provides an
208	# additional layer of hermiticity.
209	env: dict[str \| bytes, str \| bytes]
210	if hermetic:	7✔
211	# With an empty environment, we would generally get the true underlying system default
212	# encoding, which is unlikely to be what we want (it's generally ASCII, still). So we
213	# explicitly set an encoding here.
214	env = {"LC_ALL": "en_US.UTF-8"}	7✔
215	# Apply our allowlist.
216	for h in (	7✔
217	"HOME",
218	"PATH", # Needed to find Python interpreters and other binaries.
219	):
220	value = os.getenv(h)	7✔
221	if value is not None:	7✔
222	env[h] = value	7✔
223	hermetic_env = os.getenv("HERMETIC_ENV")	7✔
224	if hermetic_env:	7✔
225	for h in hermetic_env.strip(",").split(","):	×
226	value = os.getenv(h)	×
227	if value is not None:	×
228	env[h] = value	×
229	else:
230	env = cast(dict[Union[str, bytes], Union[str, bytes]], os.environ.copy())	×
231
232	env.update(PYTHONPATH=os.pathsep.join(sys.path), NO_SCIE_WARNING="1")	7✔
233	if extra_env:	7✔
234	env.update(cast(dict[Union[str, bytes], Union[str, bytes]], extra_env))	5✔
235
236	# Pants command that was called from the test shouldn't have a parent.
237	if "PANTS_PARENT_BUILD_ID" in env:	7✔
238	del env["PANTS_PARENT_BUILD_ID"]	×
239
240	return PantsJoinHandle(	7✔
241	command=pants_command,
242	process=subprocess.Popen(
243	pants_command,
244	# The type stub for the env argument is unnecessarily restrictive: it requires
245	# all keys to be str or all to be bytes. But in practice Popen supports a mix,
246	# which is what we pass. So we silence the typechecking error.
247	env=env, # type: ignore
248	stdin=subprocess.PIPE,
249	stdout=subprocess.PIPE,
250	stderr=subprocess.PIPE,
251	shell=shell,
252	),
253	workdir=workdir,
254	)
255
256
257	def run_pants_with_workdir(	7✔
258	command: Command,
259	*,
260	workdir: str,
261	hermetic: bool = True,
262	use_pantsd: bool = True,
263	config: Mapping \| None = None,
264	extra_env: Env \| None = None,
265	stdin_data: bytes \| str \| None = None,
266	shell: bool = False,
267	set_pants_ignore: bool = True,
268	stream_output: bool = False,
269	) -> PantsResult:
270	handle = run_pants_with_workdir_without_waiting(	7✔
271	command,
272	workdir=workdir,
273	hermetic=hermetic,
274	use_pantsd=use_pantsd,
275	shell=shell,
276	config=config,
277	extra_env=extra_env,
278	set_pants_ignore=set_pants_ignore,
279	)
280	return handle.join(stdin_data=stdin_data, stream_output=stream_output)	7✔
281
282
283	def run_pants(	7✔
284	command: Command,
285	*,
286	hermetic: bool = True,
287	use_pantsd: bool = False,
288	config: Mapping \| None = None,
289	extra_env: Env \| None = None,
290	stdin_data: bytes \| str \| None = None,
291	stream_output: bool = False,
292	) -> PantsResult:
293	"""Runs Pants in a subprocess.
294
295	:param command: A list of command line arguments coming after `./pants`.
296	:param hermetic: If hermetic, your actual `pants.toml` will not be used.
297	:param use_pantsd: If True, the Pants process will use pantsd.
298	:param config: Optional data for a generated TOML file. A map of <section-name> ->
299	map of key -> value.
300	:param extra_env: Set these env vars in the Pants process's environment.
301	:param stdin_data: Make this data available to be read from the process's stdin.
302	"""
303	with temporary_workdir() as workdir:	7✔
304	return run_pants_with_workdir(	7✔
305	command,
306	workdir=workdir,
307	hermetic=hermetic,
308	use_pantsd=use_pantsd,
309	config=config,
310	stdin_data=stdin_data,
311	extra_env=extra_env,
312	stream_output=stream_output,
313	)
314
315
316	# -----------------------------------------------------------------------------------------------
317	# Environment setup.
318	# -----------------------------------------------------------------------------------------------
319
320
321	@contextmanager	7✔
322	def setup_tmpdir(	7✔
323	files: Mapping[str, str], raw_files: Mapping[str, bytes] \| None = None
324	) -> Iterator[str]:
325	"""Create a temporary directory with the given files and return the tmpdir (relative to the
326	build root).
327
328	The `files` parameter is a dictionary of file paths to content. All file paths will be prefixed
329	with the tmpdir. The file content can use `{tmpdir}` to have it substituted with the actual
330	tmpdir via a format string.
331
332	The `raw_files` parameter can be used to write binary files. These
333	files will not go through formatting in any way.
334
335
336	This is useful to set up controlled test environments, such as setting up source files and
337	BUILD files.
338	"""
339
340	raw_files = raw_files or {}	7✔
341
342	with temporary_dir(root_dir=get_buildroot()) as tmpdir:	7✔
343	rel_tmpdir = os.path.relpath(tmpdir, get_buildroot())	7✔
344	for path, content in files.items():	7✔
345	safe_file_dump(	7✔
346	os.path.join(tmpdir, path), content.format(tmpdir=rel_tmpdir), makedirs=True
347	)
348
349	for path, data in raw_files.items():	7✔
350	safe_file_dump(os.path.join(tmpdir, path), data, makedirs=True, mode="wb")	×
351
352	yield rel_tmpdir	7✔
353
354
355	@contextmanager	7✔
356	def temporary_workdir(cleanup: bool = True) -> Iterator[str]:	7✔
357	# We can hard-code '.pants.d' here because we know that will always be its value
358	# in the pantsbuild/pants repo (e.g., that's what we .gitignore in that repo).
359	# Grabbing the pants_workdir config would require this pants's config object,
360	# which we don't have a reference to here.
361	root = os.path.join(get_buildroot(), ".pants.d", "tmp")	7✔
362	safe_mkdir(root)	7✔
363	with temporary_dir(root_dir=root, cleanup=cleanup, suffix=".pants.d") as tmpdir:	7✔
364	yield tmpdir	7✔
365
366
367	# -----------------------------------------------------------------------------------------------
368	# Pantsd and logs.
369	# -----------------------------------------------------------------------------------------------
370
371
372	def kill_daemon(pid_dir=None):	7✔
373	args = ["./pants"]	3✔
374	if pid_dir:	3✔
375	args.append(f"--pants-subprocessdir={pid_dir}")	3✔
376	pantsd_client = PantsDaemonClient(	3✔
377	OptionsBootstrapper.create(env=os.environ, args=args, allow_pantsrc=False).bootstrap_options
378	)
379	with pantsd_client.lifecycle_lock:	3✔
380	pantsd_client.terminate()	3✔
381
382
383	def ensure_daemon(func):	7✔
384	"""A decorator to assist with running tests with and without the daemon enabled."""
385	return pytest.mark.parametrize("use_pantsd", [True, False])(func)	2✔
386
387
388	def render_logs(workdir: str) -> None:	7✔
389	"""Renders all potentially relevant logs from the given workdir to stdout."""
390	filenames = list(glob.glob(os.path.join(workdir, "logs/exceptions*log"))) + list(	7✔
391	glob.glob(os.path.join(workdir, "pants.log"))
392	)
393	for filename in filenames:	7✔
394	rel_filename = fast_relpath(filename, workdir)	7✔
395	print(f"{rel_filename} +++ ")	7✔
396	for line in _read_log(filename):	7✔
397	print(f"{rel_filename} >>> {line}")	7✔
398	print(f"{rel_filename} --- ")	7✔
399
400
401	def read_pants_log(workdir: str) -> Iterator[str]:	7✔
402	"""Yields all lines from the pants log under the given workdir."""
403	# Surface the pants log for easy viewing via pytest's `-s` (don't capture stdio) option.
404	yield from _read_log(f"{workdir}/pants.log")	4✔
405
406
407	def _read_log(filename: str) -> Iterator[str]:	7✔
408	with open(filename) as f:	7✔
409	for line in f:	7✔
410	yield line.rstrip()	7✔
411
412
413	@dataclass(frozen=True)	7✔
414	class _TomlSerializer:	7✔
415	"""Convert a dictionary of option scopes -> Python values into TOML understood by Pants.
416
417	The constructor expects a dictionary of option scopes to their corresponding values as
418	represented in Python. For example:
419
420	{
421	"GLOBAL": {
422	"o1": True,
423	"o2": "hello",
424	"o3": [0, 1, 2],
425	},
426	"some-subsystem": {
427	"dict_option": {
428	"a": 0,
429	"b": 0,
430	},
431	},
432	}
433	"""
434
435	parsed: Mapping[str, dict[str, int \| float \| str \| bool \| list \| dict]]	7✔
436
437	def normalize(self) -> dict:	7✔
438	def normalize_section_value(option, option_value) -> tuple[str, Any]:	5✔
439	# With TOML, we store dict values as strings (for now).
440	if isinstance(option_value, dict):	5✔
441	option_value = str(option_value)	×
442	if option.endswith(".add"):	5✔
443	option = option.rsplit(".", 1)[0]	×
444	option_value = f"+{option_value!r}"	×
445	elif option.endswith(".remove"):	5✔
446	option = option.rsplit(".", 1)[0]	×
447	option_value = f"-{option_value!r}"	×
448	return option, option_value	5✔
449
450	return {	5✔
451	section: dict(
452	normalize_section_value(option, option_value)
453	for option, option_value in section_values.items()
454	)
455	for section, section_values in self.parsed.items()
456	}
457
458	def serialize(self) -> str:	7✔
459	toml_values = self.normalize()	5✔
460	return toml.dumps(toml_values)	5✔

pantsbuild / pants / 18517631058

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