21803785359

Committed 08 Feb 2026 07:13PM UTC coverage: 43.3% (-37.0%) from 80.277%

Build # 21803785359

Build Type

Pull #23085

github

Committed by

web-flow

Commit Message

Merge 7c1cd926d into 40389cc58

Pull Request Pull Request #23085: A helper method for indexing paths by source root

Run Details

2 of 6 new or added lines in 1 file covered. (33.33%)

17114 existing lines in 539 files now uncovered.

26075 of 60219 relevant lines covered (43.3%)

0.43 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,
            f"Expected {expected_runs} RunTracker run(s) to be created per pantsd run: was {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	1✔
5
6	import functools	1✔
7	import os	1✔
8	import time	1✔
9	import unittest	1✔
10	from collections.abc import Callable, Iterator, Mapping	1✔
11	from contextlib import contextmanager	1✔
12	from dataclasses import dataclass	1✔
13	from typing import Any	1✔
14
15	from colors import bold, cyan, magenta	1✔
16
17	from pants.pantsd.process_manager import ProcessManager	1✔
18	from pants.testutil.pants_integration_test import (	1✔
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	1✔
28	from pants.util.contextutil import temporary_dir	1✔
29	from pants.util.dirutil import maybe_read_file	1✔
30
31
32	def banner(s):	1✔
33	print(cyan("=" * 63))	1✔
34	print(cyan(f"- {s} {('-' * (60 - len(s)))}"))	1✔
35	print(cyan("=" * 63))	1✔
36
37
38	def attempts(	1✔
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	1✔
50	deadline = time.time() + timeout	1✔
51	while time.time() < deadline:	1✔
52	count += 1	1✔
53	yield	1✔
UNCOV 54	time.sleep(delay)	×
UNCOV 55	delay *= backoff	×
56	raise AssertionError(f"After {count} attempts in {timeout} seconds: {msg}")	×
57
58
59	def launch_waiter(	1✔
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	"""
UNCOV 67	file_to_make = os.path.join(workdir, "some_magic_file")	×
UNCOV 68	waiter_pid_file = os.path.join(workdir, "pid_file")	×
UNCOV 69	child_pid_file = os.path.join(workdir, "child_pid_file")	×
70
UNCOV 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	]
UNCOV 80	client_handle = run_pants_with_workdir_without_waiting(argv, workdir=workdir, config=config)	×
UNCOV 81	waiter_pid = -1	×
UNCOV 82	for _ in attempts("The waiter process should have written its pid."):	×
UNCOV 83	waiter_pid_str = maybe_read_file(waiter_pid_file)	×
UNCOV 84	child_pid_str = maybe_read_file(child_pid_file)	×
UNCOV 85	if waiter_pid_str and child_pid_str:	×
UNCOV 86	waiter_pid = int(waiter_pid_str)	×
UNCOV 87	child_pid = int(child_pid_str)	×
UNCOV 88	break	×
UNCOV 89	return client_handle, waiter_pid, child_pid, file_to_make	×
90
91
92	class PantsDaemonMonitor(ProcessManager):	1✔
93	def __init__(self, metadata_base_dir: str):	1✔
94	super().__init__(name="pantsd", metadata_base_dir=metadata_base_dir)	1✔
95	self._started = False	1✔
96
97	def _log(self):	1✔
98	print(magenta(f"PantsDaemonMonitor: pid is {self.pid} is_alive={self.is_alive()}"))	1✔
99
100	def assert_started_and_stopped(self, timeout: int = 30) -> None:	1✔
101	"""Asserts that pantsd was alive (it wrote a pid file), but that it stops afterward."""
UNCOV 102	self.await_pid(timeout)	×
UNCOV 103	self._started = True	×
UNCOV 104	self.assert_stopped()	×
105
106	def assert_started(self, timeout=30):	1✔
107	self.await_pid(timeout)	1✔
108	self._started = True	1✔
109	self._check_pantsd_is_alive()	1✔
110	return self.pid	1✔
111
112	def _check_pantsd_is_alive(self):	1✔
113	self._log()	1✔
114	assert self._started, (	1✔
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."	1✔
118	return self.pid	1✔
119
120	def current_memory_usage(self):	1✔
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):	1✔
UNCOV 129	if not self._started:	×
UNCOV 130	return self.assert_started()	×
131	else:
UNCOV 132	return self._check_pantsd_is_alive()	×
133
134	def assert_stopped(self):	1✔
135	self._log()	1✔
136	assert self._started, (	1✔
137	"cannot assert pantsd stoppage. Try calling assert_started before calling this method."
138	)
139	for _ in attempts("pantsd should be stopped!"):	1✔
140	if self.is_dead():	1✔
141	break	1✔
142
143
144	@dataclass(frozen=True)	1✔
145	class PantsdRunContext:	1✔
146	runner: Callable[..., Any]	1✔
147	checker: PantsDaemonMonitor	1✔
148	workdir: str	1✔
149	pantsd_config: dict[str, Any]	1✔
150
151
152	class PantsDaemonIntegrationTestBase(unittest.TestCase):	1✔
153	@staticmethod	1✔
154	def run_pants(args, *kwargs) -> PantsResult:	1✔
155	# We set our own ad-hoc pantsd configuration in most of these tests.
UNCOV 156	return run_pants(args, {kwargs, *{"use_pantsd": False}})	×
157
158	@staticmethod	1✔
159	def run_pants_with_workdir(args, *kwargs) -> PantsResult:	1✔
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}})	1✔
162
163	@staticmethod	1✔
164	def run_pants_with_workdir_without_waiting(args, *kwargs) -> PantsJoinHandle:	1✔
165	# We set our own ad-hoc pantsd configuration in most of these tests.
UNCOV 166	return run_pants_with_workdir_without_waiting(args, {kwargs, *{"use_pantsd": False}})	×
167
168	@contextmanager	1✔
169	def pantsd_test_context(	1✔
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:	1✔
173	pid_dir = os.path.join(dot_pants_dot_d, "pids")	1✔
174	workdir = os.path.join(dot_pants_dot_d, "workdir")	1✔
175	print(f"\npantsd log is {workdir}/pants.log")	1✔
176	pantsd_config = {	1✔
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:	1✔
UNCOV 193	recursively_update(pantsd_config, extra_config)	×
194	print(f">>> config: \n{pantsd_config}\n")	1✔
195
196	checker = PantsDaemonMonitor(pid_dir)	1✔
197	kill_daemon(pid_dir)	1✔
198	try:	1✔
199	yield workdir, pantsd_config, checker	1✔
200	kill_daemon(pid_dir)	1✔
201	checker.assert_stopped()	1✔
202	finally:
203	banner("BEGIN pants.log")	1✔
204	for line in read_pants_log(workdir):	1✔
205	print(line)	1✔
206	banner("END pants.log")	1✔
207
208	@contextmanager	1✔
209	def pantsd_successful_run_context(self, args, *kwargs) -> Iterator[PantsdRunContext]:	1✔
210	with self.pantsd_run_context(args, success=True, *kwargs) as context: # type: ignore[misc]	1✔
211	yield context	1✔
212
213	@contextmanager	1✔
214	def pantsd_run_context(	1✔
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 (	1✔
222	workdir,
223	pantsd_config,
224	checker,
225	):
226	runner = functools.partial(	1✔
227	self.assert_runner,
228	workdir,
229	pantsd_config,
230	extra_env=extra_env,
231	success=success,
232	)
233	yield PantsdRunContext(	1✔
234	runner=runner, checker=checker, workdir=workdir, pantsd_config=pantsd_config
235	)
236
237	def _run_count(self, workdir):	1✔
238	run_tracker_dir = os.path.join(workdir, "run-tracker")	1✔
239	if os.path.isdir(run_tracker_dir):	1✔
240	return len([f for f in os.listdir(run_tracker_dir) if f != "latest"])	1✔
241	else:
242	return 0	1✔
243
244	def assert_runner(	1✔
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()	1✔
255	recursively_update(combined_config, extra_config or {})	1✔
256	print(	1✔
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)	1✔
266	start_time = time.time()	1✔
267	run = self.run_pants_with_workdir(	1✔
268	cmd, workdir=workdir, config=combined_config, extra_env=extra_env or {}
269	)
270	elapsed = time.time() - start_time	1✔
271	print(bold(cyan(f"\ncompleted in {elapsed} seconds")))	1✔
272
273	if success:	1✔
274	run.assert_success()	1✔
275	else:
UNCOV 276	run.assert_failure()	×
277
278	runs_created = self._run_count(workdir) - run_count	1✔
279	self.assertEqual(	1✔
280	runs_created,
281	expected_runs,
282	f"Expected {expected_runs} RunTracker run(s) to be created per pantsd run: was {runs_created}",
283	)
284
285	return run	1✔

pantsbuild / pants / 21803785359

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