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

79.43

/src/python/pants/pantsd/pantsd_integration_test_base.py

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

from __future__ import annotations

import functools
import os
import time
import unittest
from collections.abc import Callable, Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any

from colors import bold, cyan, magenta

from pants.pantsd.process_manager import ProcessManager
from pants.testutil.pants_integration_test import (
    PantsJoinHandle,
    PantsResult,
    kill_daemon,
    read_pants_log,
    run_pants,
    run_pants_with_workdir,
    run_pants_with_workdir_without_waiting,
)
from pants.util.collections import recursively_update
from pants.util.contextutil import temporary_dir
from pants.util.dirutil import maybe_read_file


def banner(s):
    print(cyan("=" * 63))
    print(cyan(f"- {s} {('-' * (60 - len(s)))}"))
    print(cyan("=" * 63))


def attempts(
    msg: str,
    *,
    delay: float = 0.5,
    timeout: float = 30,
    backoff: float = 1.2,
) -> Iterator[None]:
    """A generator that yields a number of times before failing.

    A caller should break out of a loop on the generator in order to succeed.
    """
    count = 0
    deadline = time.time() + timeout
    while time.time() < deadline:
        count += 1
        yield
        time.sleep(delay)
        delay *= backoff
    raise AssertionError(f"After {count} attempts in {timeout} seconds: {msg}")


def launch_waiter(
    *, workdir: str, config: Mapping | None = None, cleanup_wait_time: int = 0
) -> tuple[PantsJoinHandle, int, int, str]:
    """Launch a process that will wait forever for a file to be created.

    Returns the pants client handle, the pid of the waiting process, the pid of a child of the
    waiting process, and the file to create to cause the waiting child to exit.
    """
    file_to_make = os.path.join(workdir, "some_magic_file")
    waiter_pid_file = os.path.join(workdir, "pid_file")
    child_pid_file = os.path.join(workdir, "child_pid_file")

    argv = [
        "run",
        "testprojects/src/python/coordinated_runs:waiter",
        "--",
        file_to_make,
        waiter_pid_file,
        child_pid_file,
        str(cleanup_wait_time),
    ]
    client_handle = run_pants_with_workdir_without_waiting(argv, workdir=workdir, config=config)
    waiter_pid = -1
    for _ in attempts("The waiter process should have written its pid."):
        waiter_pid_str = maybe_read_file(waiter_pid_file)
        child_pid_str = maybe_read_file(child_pid_file)
        if waiter_pid_str and child_pid_str:
            waiter_pid = int(waiter_pid_str)
            child_pid = int(child_pid_str)
            break
    return client_handle, waiter_pid, child_pid, file_to_make


class PantsDaemonMonitor(ProcessManager):
    def __init__(self, metadata_base_dir: str):
        super().__init__(name="pantsd", metadata_base_dir=metadata_base_dir)
        self._started = False

    def _log(self):
        print(magenta(f"PantsDaemonMonitor: pid is {self.pid} is_alive={self.is_alive()}"))

    def assert_started_and_stopped(self, timeout: int = 30) -> None:
        """Asserts that pantsd was alive (it wrote a pid file), but that it stops afterward."""
        self.await_pid(timeout)
        self._started = True
        self.assert_stopped()

    def assert_started(self, timeout=30):
        self.await_pid(timeout)
        self._started = True
        self._check_pantsd_is_alive()
        return self.pid

    def _check_pantsd_is_alive(self):
        self._log()
        assert self._started, (
            "cannot assert that pantsd is running. Try calling assert_started before calling this method."
        )
        assert self.is_alive(), "pantsd was not alive."
        return self.pid

    def current_memory_usage(self):
        """Return the current memory usage of the pantsd process (which must be running)

        :return: memory usage in bytes
        """
        self.assert_running()
        return self._as_process().memory_info()[0]

    def assert_running(self):
        if not self._started:
            return self.assert_started()
        else:
            return self._check_pantsd_is_alive()

    def assert_stopped(self):
        self._log()
        assert self._started, (
            "cannot assert pantsd stoppage. Try calling assert_started before calling this method."
        )
        for _ in attempts("pantsd should be stopped!"):
            if self.is_dead():
                break


@dataclass(frozen=True)
class PantsdRunContext:
    runner: Callable[..., Any]
    checker: PantsDaemonMonitor
    workdir: str
    pantsd_config: dict[str, Any]


class PantsDaemonIntegrationTestBase(unittest.TestCase):
    @staticmethod
    def run_pants(*args, **kwargs) -> PantsResult:
        # We set our own ad-hoc pantsd configuration in most of these tests.
        return run_pants(*args, **{**kwargs, **{"use_pantsd": False}})

    @staticmethod
    def run_pants_with_workdir(*args, **kwargs) -> PantsResult:
        # We set our own ad-hoc pantsd configuration in most of these tests.
        return run_pants_with_workdir(*args, **{**kwargs, **{"use_pantsd": False}})

    @staticmethod
    def run_pants_with_workdir_without_waiting(*args, **kwargs) -> PantsJoinHandle:
        # We set our own ad-hoc pantsd configuration in most of these tests.
        return run_pants_with_workdir_without_waiting(*args, **{**kwargs, **{"use_pantsd": False}})

    @contextmanager
    def pantsd_test_context(
        self, *, log_level: str = "info", extra_config: dict[str, Any] | None = None
    ) -> Iterator[tuple[str, dict[str, Any], PantsDaemonMonitor]]:
        with temporary_dir(root_dir=os.getcwd()) as dot_pants_dot_d:
            pid_dir = os.path.join(dot_pants_dot_d, "pids")
            workdir = os.path.join(dot_pants_dot_d, "workdir")
            print(f"\npantsd log is {workdir}/pants.log")
            pantsd_config = {
                "GLOBAL": {
                    "pantsd": True,
                    "level": log_level,
                    "pants_subprocessdir": pid_dir,
                    "backend_packages": [
                        # Provide goals used by various tests.
                        "pants.backend.python",
                        "pants.backend.python.lint.flake8",
                    ],
                },
                "python": {
                    "interpreter_constraints": "['>=3.11,<3.12']",
                },
            }

            if extra_config:
                recursively_update(pantsd_config, extra_config)
            print(f">>> config: \n{pantsd_config}\n")

            checker = PantsDaemonMonitor(pid_dir)
            kill_daemon(pid_dir)
            try:
                yield workdir, pantsd_config, checker
                kill_daemon(pid_dir)
                checker.assert_stopped()
            finally:
                banner("BEGIN pants.log")
                for line in read_pants_log(workdir):
                    print(line)
                banner("END pants.log")

    @contextmanager
    def pantsd_successful_run_context(self, *args, **kwargs) -> Iterator[PantsdRunContext]:
        with self.pantsd_run_context(*args, success=True, **kwargs) as context:  # type: ignore[misc]
            yield context

    @contextmanager
    def pantsd_run_context(
        self,
        log_level: str = "info",
        extra_config: dict[str, Any] | None = None,
        extra_env: dict[str, str] | None = None,
        success: bool = True,
    ) -> Iterator[PantsdRunContext]:
        with self.pantsd_test_context(log_level=log_level, extra_config=extra_config) as (
            workdir,
            pantsd_config,
            checker,
        ):
            runner = functools.partial(
                self.assert_runner,
                workdir,
                pantsd_config,
                extra_env=extra_env,
                success=success,
            )
            yield PantsdRunContext(
                runner=runner, checker=checker, workdir=workdir, pantsd_config=pantsd_config
            )

    def _run_count(self, workdir):
        run_tracker_dir = os.path.join(workdir, "run-tracker")
        if os.path.isdir(run_tracker_dir):
            return len([f for f in os.listdir(run_tracker_dir) if f != "latest"])
        else:
            return 0

    def assert_runner(
        self,
        workdir: str,
        config,
        cmd,
        extra_config=None,
        extra_env=None,
        success=True,
        expected_runs: int = 1,
    ):
        combined_config = config.copy()
        recursively_update(combined_config, extra_config or {})
        print(
            bold(
                cyan(
                    "\nrunning: ./pants {} (config={}) (extra_env={})".format(
                        " ".join(cmd), combined_config, extra_env
                    )
                )
            )
        )
        run_count = self._run_count(workdir)
        start_time = time.time()
        run = self.run_pants_with_workdir(
            cmd, workdir=workdir, config=combined_config, extra_env=extra_env or {}
        )
        elapsed = time.time() - start_time
        print(bold(cyan(f"\ncompleted in {elapsed} seconds")))

        if success:
            run.assert_success()
        else:
            run.assert_failure()

        runs_created = self._run_count(workdir) - run_count
        self.assertEqual(
            runs_created,
            expected_runs,
            "Expected {} RunTracker run(s) to be created per pantsd run: was {}".format(
                expected_runs, runs_created
            ),
        )

        return run

1	# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	3✔
5
6	import functools	3✔
7	import os	3✔
8	import time	3✔
9	import unittest	3✔
10	from collections.abc import Callable, Iterator, Mapping	3✔
11	from contextlib import contextmanager	3✔
12	from dataclasses import dataclass	3✔
13	from typing import Any	3✔
14
15	from colors import bold, cyan, magenta	3✔
16
17	from pants.pantsd.process_manager import ProcessManager	3✔
18	from pants.testutil.pants_integration_test import (	3✔
19	PantsJoinHandle,
20	PantsResult,
21	kill_daemon,
22	read_pants_log,
23	run_pants,
24	run_pants_with_workdir,
25	run_pants_with_workdir_without_waiting,
26	)
27	from pants.util.collections import recursively_update	3✔
28	from pants.util.contextutil import temporary_dir	3✔
29	from pants.util.dirutil import maybe_read_file	3✔
30
31
32	def banner(s):	3✔
33	print(cyan("=" * 63))	3✔
34	print(cyan(f"- {s} {('-' * (60 - len(s)))}"))	3✔
35	print(cyan("=" * 63))	3✔
36
37
38	def attempts(	3✔
39	msg: str,
40	*,
41	delay: float = 0.5,
42	timeout: float = 30,
43	backoff: float = 1.2,
44	) -> Iterator[None]:
45	"""A generator that yields a number of times before failing.
46
47	A caller should break out of a loop on the generator in order to succeed.
48	"""
49	count = 0	3✔
50	deadline = time.time() + timeout	3✔
51	while time.time() < deadline:	3✔
52	count += 1	3✔
53	yield	3✔
54	time.sleep(delay)	×
55	delay *= backoff	×
56	raise AssertionError(f"After {count} attempts in {timeout} seconds: {msg}")	×
57
58
59	def launch_waiter(	3✔
60	*, workdir: str, config: Mapping \| None = None, cleanup_wait_time: int = 0
61	) -> tuple[PantsJoinHandle, int, int, str]:
62	"""Launch a process that will wait forever for a file to be created.
63
64	Returns the pants client handle, the pid of the waiting process, the pid of a child of the
65	waiting process, and the file to create to cause the waiting child to exit.
66	"""
67	file_to_make = os.path.join(workdir, "some_magic_file")	×
68	waiter_pid_file = os.path.join(workdir, "pid_file")	×
69	child_pid_file = os.path.join(workdir, "child_pid_file")	×
70
71	argv = [	×
72	"run",
73	"testprojects/src/python/coordinated_runs:waiter",
74	"--",
75	file_to_make,
76	waiter_pid_file,
77	child_pid_file,
78	str(cleanup_wait_time),
79	]
80	client_handle = run_pants_with_workdir_without_waiting(argv, workdir=workdir, config=config)	×
81	waiter_pid = -1	×
82	for _ in attempts("The waiter process should have written its pid."):	×
83	waiter_pid_str = maybe_read_file(waiter_pid_file)	×
84	child_pid_str = maybe_read_file(child_pid_file)	×
85	if waiter_pid_str and child_pid_str:	×
86	waiter_pid = int(waiter_pid_str)	×
87	child_pid = int(child_pid_str)	×
88	break	×
89	return client_handle, waiter_pid, child_pid, file_to_make	×
90
91
92	class PantsDaemonMonitor(ProcessManager):	3✔
93	def __init__(self, metadata_base_dir: str):	3✔
94	super().__init__(name="pantsd", metadata_base_dir=metadata_base_dir)	3✔
95	self._started = False	3✔
96
97	def _log(self):	3✔
98	print(magenta(f"PantsDaemonMonitor: pid is {self.pid} is_alive={self.is_alive()}"))	3✔
99
100	def assert_started_and_stopped(self, timeout: int = 30) -> None:	3✔
101	"""Asserts that pantsd was alive (it wrote a pid file), but that it stops afterward."""
102	self.await_pid(timeout)	×
103	self._started = True	×
104	self.assert_stopped()	×
105
106	def assert_started(self, timeout=30):	3✔
107	self.await_pid(timeout)	3✔
108	self._started = True	3✔
109	self._check_pantsd_is_alive()	3✔
110	return self.pid	3✔
111
112	def _check_pantsd_is_alive(self):	3✔
113	self._log()	3✔
114	assert self._started, (	3✔
115	"cannot assert that pantsd is running. Try calling assert_started before calling this method."
116	)
117	assert self.is_alive(), "pantsd was not alive."	3✔
118	return self.pid	3✔
119
120	def current_memory_usage(self):	3✔
121	"""Return the current memory usage of the pantsd process (which must be running)
122
123	:return: memory usage in bytes
124	"""
125	self.assert_running()	×
126	return self._as_process().memory_info()[0]	×
127
128	def assert_running(self):	3✔
129	if not self._started:	×
130	return self.assert_started()	×
131	else:
132	return self._check_pantsd_is_alive()	×
133
134	def assert_stopped(self):	3✔
135	self._log()	3✔
136	assert self._started, (	3✔
137	"cannot assert pantsd stoppage. Try calling assert_started before calling this method."
138	)
139	for _ in attempts("pantsd should be stopped!"):	3✔
140	if self.is_dead():	3✔
141	break	3✔
142
143
144	@dataclass(frozen=True)	3✔
145	class PantsdRunContext:	3✔
146	runner: Callable[..., Any]	3✔
147	checker: PantsDaemonMonitor	3✔
148	workdir: str	3✔
149	pantsd_config: dict[str, Any]	3✔
150
151
152	class PantsDaemonIntegrationTestBase(unittest.TestCase):	3✔
153	@staticmethod	3✔
154	def run_pants(args, *kwargs) -> PantsResult:	3✔
155	# We set our own ad-hoc pantsd configuration in most of these tests.
156	return run_pants(args, {kwargs, *{"use_pantsd": False}})	×
157
158	@staticmethod	3✔
159	def run_pants_with_workdir(args, *kwargs) -> PantsResult:	3✔
160	# We set our own ad-hoc pantsd configuration in most of these tests.
161	return run_pants_with_workdir(args, {kwargs, *{"use_pantsd": False}})	3✔
162
163	@staticmethod	3✔
164	def run_pants_with_workdir_without_waiting(args, *kwargs) -> PantsJoinHandle:	3✔
165	# We set our own ad-hoc pantsd configuration in most of these tests.
166	return run_pants_with_workdir_without_waiting(args, {kwargs, *{"use_pantsd": False}})	×
167
168	@contextmanager	3✔
169	def pantsd_test_context(	3✔
170	self, *, log_level: str = "info", extra_config: dict[str, Any] \| None = None
171	) -> Iterator[tuple[str, dict[str, Any], PantsDaemonMonitor]]:
172	with temporary_dir(root_dir=os.getcwd()) as dot_pants_dot_d:	3✔
173	pid_dir = os.path.join(dot_pants_dot_d, "pids")	3✔
174	workdir = os.path.join(dot_pants_dot_d, "workdir")	3✔
175	print(f"\npantsd log is {workdir}/pants.log")	3✔
176	pantsd_config = {	3✔
177	"GLOBAL": {
178	"pantsd": True,
179	"level": log_level,
180	"pants_subprocessdir": pid_dir,
181	"backend_packages": [
182	# Provide goals used by various tests.
183	"pants.backend.python",
184	"pants.backend.python.lint.flake8",
185	],
186	},
187	"python": {
188	"interpreter_constraints": "['>=3.11,<3.12']",
189	},
190	}
191
192	if extra_config:	3✔
193	recursively_update(pantsd_config, extra_config)	×
194	print(f">>> config: \n{pantsd_config}\n")	3✔
195
196	checker = PantsDaemonMonitor(pid_dir)	3✔
197	kill_daemon(pid_dir)	3✔
198	try:	3✔
199	yield workdir, pantsd_config, checker	3✔
200	kill_daemon(pid_dir)	3✔
201	checker.assert_stopped()	3✔
202	finally:
203	banner("BEGIN pants.log")	3✔
204	for line in read_pants_log(workdir):	3✔
205	print(line)	3✔
206	banner("END pants.log")	3✔
207
208	@contextmanager	3✔
209	def pantsd_successful_run_context(self, args, *kwargs) -> Iterator[PantsdRunContext]:	3✔
210	with self.pantsd_run_context(args, success=True, *kwargs) as context: # type: ignore[misc]	3✔
211	yield context	3✔
212
213	@contextmanager	3✔
214	def pantsd_run_context(	3✔
215	self,
216	log_level: str = "info",
217	extra_config: dict[str, Any] \| None = None,
218	extra_env: dict[str, str] \| None = None,
219	success: bool = True,
220	) -> Iterator[PantsdRunContext]:
221	with self.pantsd_test_context(log_level=log_level, extra_config=extra_config) as (	3✔
222	workdir,
223	pantsd_config,
224	checker,
225	):
226	runner = functools.partial(	3✔
227	self.assert_runner,
228	workdir,
229	pantsd_config,
230	extra_env=extra_env,
231	success=success,
232	)
233	yield PantsdRunContext(	3✔
234	runner=runner, checker=checker, workdir=workdir, pantsd_config=pantsd_config
235	)
236
237	def _run_count(self, workdir):	3✔
238	run_tracker_dir = os.path.join(workdir, "run-tracker")	3✔
239	if os.path.isdir(run_tracker_dir):	3✔
240	return len([f for f in os.listdir(run_tracker_dir) if f != "latest"])	3✔
241	else:
242	return 0	3✔
243
244	def assert_runner(	3✔
245	self,
246	workdir: str,
247	config,
248	cmd,
249	extra_config=None,
250	extra_env=None,
251	success=True,
252	expected_runs: int = 1,
253	):
254	combined_config = config.copy()	3✔
255	recursively_update(combined_config, extra_config or {})	3✔
256	print(	3✔
257	bold(
258	cyan(
259	"\nrunning: ./pants {} (config={}) (extra_env={})".format(
260	" ".join(cmd), combined_config, extra_env
261	)
262	)
263	)
264	)
265	run_count = self._run_count(workdir)	3✔
266	start_time = time.time()	3✔
267	run = self.run_pants_with_workdir(	3✔
268	cmd, workdir=workdir, config=combined_config, extra_env=extra_env or {}
269	)
270	elapsed = time.time() - start_time	3✔
271	print(bold(cyan(f"\ncompleted in {elapsed} seconds")))	3✔
272
273	if success:	3✔
274	run.assert_success()	3✔
275	else:
276	run.assert_failure()	×
277
278	runs_created = self._run_count(workdir) - run_count	3✔
279	self.assertEqual(	3✔
280	runs_created,
281	expected_runs,
282	"Expected {} RunTracker run(s) to be created per pantsd run: was {}".format(
283	expected_runs, runs_created
284	),
285	)
286
287	return run	3✔

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