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

63.16

/src/python/pants/engine/internals/engine_testutil.py

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

from __future__ import annotations

import re
from collections.abc import Callable
from dataclasses import dataclass, field

from pants.engine.streaming_workunit_handler import WorkunitsCallback


def assert_equal_with_printing(
    expected, actual, uniform_formatter: Callable[[str], str] | None = None
):
    """Asserts equality, but also prints the values so they can be compared on failure."""
    str_actual = str(actual)
    print("Expected:")
    print(expected)
    print("Actual:")
    print(str_actual)

    if uniform_formatter is not None:
        expected = uniform_formatter(expected)
        str_actual = uniform_formatter(str_actual)
    assert expected == str_actual


def remove_locations_from_traceback(trace: str) -> str:
    location_pattern = re.compile(r'"/.*", line \d+')
    address_pattern = re.compile(r"0x[0-9a-f]+")
    new_trace = location_pattern.sub("LOCATION-INFO", trace)
    new_trace = address_pattern.sub("0xEEEEEEEEE", new_trace)
    return new_trace


@dataclass
class WorkunitTracker(WorkunitsCallback):
    """This class records every non-empty batch of started and completed workunits received from the
    engine."""

    finished_workunit_chunks: list[list[dict]] = field(default_factory=list)
    started_workunit_chunks: list[list[dict]] = field(default_factory=list)
    finished: bool = False

    @property
    def can_finish_async(self) -> bool:
        return False

    def __call__(self, **kwargs) -> None:
        if kwargs["finished"] is True:
            self.finished = True

        started_workunits = kwargs.get("started_workunits")
        if started_workunits:
            self.started_workunit_chunks.append(started_workunits)

        completed_workunits = kwargs.get("completed_workunits")
        if completed_workunits:
            self.finished_workunit_chunks.append(completed_workunits)

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 re	1✔
7	from collections.abc import Callable	1✔
8	from dataclasses import dataclass, field	1✔
9
10	from pants.engine.streaming_workunit_handler import WorkunitsCallback	1✔
11
12
13	def assert_equal_with_printing(	1✔
14	expected, actual, uniform_formatter: Callable[[str], str] \| None = None
15	):
16	"""Asserts equality, but also prints the values so they can be compared on failure."""
17	str_actual = str(actual)	1✔
18	print("Expected:")	1✔
19	print(expected)	1✔
20	print("Actual:")	1✔
21	print(str_actual)	1✔
22
23	if uniform_formatter is not None:	1✔
24	expected = uniform_formatter(expected)	1✔
25	str_actual = uniform_formatter(str_actual)	1✔
26	assert expected == str_actual	1✔
27
28
29	def remove_locations_from_traceback(trace: str) -> str:	1✔
30	location_pattern = re.compile(r'"/.*", line \d+')	×
31	address_pattern = re.compile(r"0x[0-9a-f]+")	×
32	new_trace = location_pattern.sub("LOCATION-INFO", trace)	×
33	new_trace = address_pattern.sub("0xEEEEEEEEE", new_trace)	×
34	return new_trace	×
35
36
37	@dataclass	1✔
38	class WorkunitTracker(WorkunitsCallback):	1✔
39	"""This class records every non-empty batch of started and completed workunits received from the
40	engine."""
41
42	finished_workunit_chunks: list[list[dict]] = field(default_factory=list)	1✔
43	started_workunit_chunks: list[list[dict]] = field(default_factory=list)	1✔
44	finished: bool = False	1✔
45
46	@property	1✔
47	def can_finish_async(self) -> bool:	1✔
48	return False	×
49
50	def __call__(self, **kwargs) -> None:	1✔
51	if kwargs["finished"] is True:	×
52	self.finished = True	×
53
54	started_workunits = kwargs.get("started_workunits")	×
55	if started_workunits:	×
56	self.started_workunit_chunks.append(started_workunits)	×
57
58	completed_workunits = kwargs.get("completed_workunits")	×
59	if completed_workunits:	×
60	self.finished_workunit_chunks.append(completed_workunits)	×

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