20826906489

Committed 08 Jan 2026 06:09PM UTC coverage: 80.192% (-0.08%) from 80.274%

Build # 20826906489

Build Type

Pull #22987

github

Committed by

web-flow

Commit Message

Merge aa52dd80f into 0d471f924

Pull Request Pull Request #22987: WIP DRAFT: Pex 2.77.1 tests

Coverage Stats

3 of 3 new or added lines in 1 file covered. (100.0%)

90 existing lines in 15 files now uncovered.

78714 of 98157 relevant lines covered (80.19%)

3.36 hits per line

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

90.24

/src/python/pants/engine/streaming_workunit_handler_integration_test.py

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

from __future__ import annotations

import os
import signal
from collections.abc import Mapping

import pytest
from workunit_logger.register import FINISHED_SUCCESSFULLY

from pants.pantsd.pantsd_integration_test_base import attempts, launch_waiter
from pants.testutil.pants_integration_test import (
    PantsResult,
    run_pants,
    setup_tmpdir,
    temporary_workdir,
)
from pants.util.dirutil import maybe_read_file


def workunit_logger_config(log_dest: str, *, pantsd: bool = True) -> Mapping:
    return {
        "GLOBAL": {
            "pantsd": pantsd,
            "backend_packages.add": ["workunit_logger", "pants.backend.python"],
        },
        "workunit-logger": {"dest": log_dest},
        "python": {"interpreter_constraints": "['>=3.8,<3.10']"},
    }


def run(
    args: list[str], success: bool = True, *, files: dict[str, str] | None = None
) -> tuple[PantsResult, str | None]:
    with setup_tmpdir(files or {}) as tmpdir:
        dest = os.path.join(tmpdir, "dest.log")
        normalized_args = [arg.format(tmpdir=tmpdir) for arg in args]
        pants_run = run_pants(normalized_args, config=workunit_logger_config(dest))
        if success:
            pants_run.assert_success()
            confirm_eventual_success(dest)
        else:
            pants_run.assert_failure()
        return pants_run, maybe_read_file(dest)


def confirm_eventual_success(log_dest: str) -> None:
    for _ in attempts("The log should eventually show that the SWH shut down."):
        content = maybe_read_file(log_dest)
        if content and FINISHED_SUCCESSFULLY in content:
            break


def test_list() -> None:
    run(["list", "{tmpdir}/foo::"], files={"foo/BUILD": "target()"})


def test_help() -> None:
    run(["help"])
    run(["--version"])


@pytest.mark.parametrize("pantsd", [True, False])
def test_ctrl_c(pantsd: bool) -> None:
    with temporary_workdir() as workdir:
        dest = os.path.join(workdir, "dest.log")

        # Start a pantsd run that will wait forever, then kill the pantsd client.
        client_handle, _, _, _ = launch_waiter(
            workdir=workdir, config=workunit_logger_config(dest, pantsd=pantsd)
        )
        client_pid = client_handle.process.pid
        os.kill(client_pid, signal.SIGINT)

        # Confirm that finish is still called (even though it may be backgrounded in the server).
        confirm_eventual_success(dest)


def test_restart() -> None:
    # Will trigger a restart
    run(["--pantsd-max-memory-usage=1", "roots"])

1	# Copyright 2020 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 os	1✔
7	import signal	1✔
8	from collections.abc import Mapping	1✔
9
10	import pytest	1✔
11	from workunit_logger.register import FINISHED_SUCCESSFULLY	1✔
12
13	from pants.pantsd.pantsd_integration_test_base import attempts, launch_waiter	1✔
14	from pants.testutil.pants_integration_test import (	1✔
15	PantsResult,
16	run_pants,
17	setup_tmpdir,
18	temporary_workdir,
19	)
20	from pants.util.dirutil import maybe_read_file	1✔
21
22
23	def workunit_logger_config(log_dest: str, *, pantsd: bool = True) -> Mapping:	1✔
24	return {	1✔
25	"GLOBAL": {
26	"pantsd": pantsd,
27	"backend_packages.add": ["workunit_logger", "pants.backend.python"],
28	},
29	"workunit-logger": {"dest": log_dest},
30	"python": {"interpreter_constraints": "['>=3.8,<3.10']"},
31	}
32
33
34	def run(	1✔
35	args: list[str], success: bool = True, *, files: dict[str, str] \| None = None
36	) -> tuple[PantsResult, str \| None]:
37	with setup_tmpdir(files or {}) as tmpdir:	1✔
38	dest = os.path.join(tmpdir, "dest.log")	1✔
39	normalized_args = [arg.format(tmpdir=tmpdir) for arg in args]	1✔
40	pants_run = run_pants(normalized_args, config=workunit_logger_config(dest))	1✔
41	if success:	1✔
42	pants_run.assert_success()	1✔
43	confirm_eventual_success(dest)	1✔
44	else:
45	pants_run.assert_failure()	×
46	return pants_run, maybe_read_file(dest)	1✔
47
48
49	def confirm_eventual_success(log_dest: str) -> None:	1✔
50	for _ in attempts("The log should eventually show that the SWH shut down."):	1✔
51	content = maybe_read_file(log_dest)	1✔
52	if content and FINISHED_SUCCESSFULLY in content:	1✔
53	break	1✔
54
55
56	def test_list() -> None:	1✔
57	run(["list", "{tmpdir}/foo::"], files={"foo/BUILD": "target()"})	1✔
58
59
60	def test_help() -> None:	1✔
61	run(["help"])	1✔
62	run(["--version"])	1✔
63
64
65	@pytest.mark.parametrize("pantsd", [True, False])	1✔
66	def test_ctrl_c(pantsd: bool) -> None:	1✔
67	with temporary_workdir() as workdir:	1✔
68	dest = os.path.join(workdir, "dest.log")	1✔
69
70	# Start a pantsd run that will wait forever, then kill the pantsd client.
71	client_handle, _, _, _ = launch_waiter(	1✔
72	workdir=workdir, config=workunit_logger_config(dest, pantsd=pantsd)
73	)
UNCOV 74	client_pid = client_handle.process.pid	×
UNCOV 75	os.kill(client_pid, signal.SIGINT)	×
76
77	# Confirm that finish is still called (even though it may be backgrounded in the server).
UNCOV 78	confirm_eventual_success(dest)	×
79
80
81	def test_restart() -> None:	1✔
82	# Will trigger a restart
83	run(["--pantsd-max-memory-usage=1", "roots"])	1✔

pantsbuild / pants / 20826906489

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