25259185675

Committed 02 May 2026 06:47PM UTC coverage: 92.141% (-0.8%) from 92.955%

Build # 25259185675

Build Type

push

github

Committed by

web-flow

Commit Message

Fix the dynamic UI. (#23306)

In #23114 we upgraded to indicatif 0.18.4,
which included a fix to respect TERM, and 
display nothing if it's unset.

Since we did not pass TERM through pantsd, the
dynamic ui is now not shown. 

This change fixes that, and also pass NO_COLOR
through, since indicatif inspects it too.

Coverage Stats

88773 of 96345 relevant lines covered (92.14%)

3.83 hits per line

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

95.74

/src/python/pants/backend/cue/goals/fix_test.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

from collections import namedtuple
from collections.abc import Iterable
from textwrap import dedent

import pytest

from pants.backend.cue.goals.fix import CueFmtRequest, rules
from pants.backend.cue.target_types import CueFieldSet, CuePackageTarget
from pants.core.goals.fmt import FmtResult, Partitions
from pants.core.util_rules import external_tool, source_files
from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
from pants.engine.addresses import AddressInput
from pants.engine.fs import DigestContents
from pants.testutil.rule_runner import QueryRule, RuleRunner


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            *rules(),
            *external_tool.rules(),
            *source_files.rules(),
            QueryRule(Partitions, [CueFmtRequest.PartitionRequest]),
            QueryRule(FmtResult, [CueFmtRequest.Batch]),
            QueryRule(SourceFiles, (SourceFilesRequest,)),
        ],
        target_types=[CuePackageTarget],
    )


def run_cue(
    rule_runner: RuleRunner,
    addresses: Iterable[str],
    *,
    extra_args: list[str] | None = None,
) -> tuple[FmtResult, ...]:
    targets = [
        rule_runner.get_target(
            AddressInput.parse(address, description_of_origin="cue tests").file_to_address()
        )
        for address in addresses
    ]
    rule_runner.set_options(
        extra_args or (),
    )
    field_sets = [CueFieldSet.create(tgt) for tgt in targets]
    kwargs = {}
    partitions = rule_runner.request(
        Partitions, [CueFmtRequest.PartitionRequest(tuple(field_sets))]
    )
    input_sources = rule_runner.request(
        SourceFiles, [SourceFilesRequest(field_set.sources for field_set in field_sets)]
    )
    kwargs["snapshot"] = input_sources.snapshot
    assert len(partitions) == 1
    results = []
    for partition in partitions:
        result = rule_runner.request(
            FmtResult,
            [CueFmtRequest.Batch("cue", partition.elements, partition.metadata, **kwargs)],
        )
        assert result.tool_name == "cue"
        results.append(result)
    return tuple(results)


ExpectedResult = namedtuple("ExpectedResult", "stdout, stderr, files", defaults=("", "", ()))


def assert_results(
    rule_runner: RuleRunner,
    results: tuple[FmtResult, ...],
    *expected_results: ExpectedResult,
) -> None:
    assert len(results) == len(expected_results)
    for result, expected in zip(results, expected_results):
        assert result.stdout == expected.stdout
        assert result.stderr == expected.stderr
        for filename, contents in expected.files:
            fc = next(
                fc
                for fc in rule_runner.request(DigestContents, [result.output.digest])
                if fc.path == filename
            )
            assert fc.content.decode() == contents


def test_simple_cue_fmt(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "src/BUILD": "cue_package()",
            "src/example.cue": dedent(
                """\
                package example

                config: "value"
                """
            ),
        }
    )
    assert_results(
        rule_runner,
        run_cue(rule_runner, ["src/example.cue"]),
        ExpectedResult(),
    )


@pytest.mark.platform_specific_behavior
def test_simple_cue_fmt_issue(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "src/BUILD": "cue_package()",
            "src/example.cue": dedent(
                """\
                package example

                config:{
                config: 42
                }
                """
            ),
        }
    )
    assert_results(
        rule_runner,
        run_cue(rule_runner, ["src/example.cue"]),
        # `cue fmt` does not output anything. so we have only the formatted files to go on. :/
        ExpectedResult(
            files=[
                (
                    "src/example.cue",
                    dedent(
                        """\
                        package example

                        config: {
                        \tconfig: 42
                        }
                        """
                    ),
                )
            ]
        ),
    )

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	2✔
4
5	from collections import namedtuple	2✔
6	from collections.abc import Iterable	2✔
7	from textwrap import dedent	2✔
8
9	import pytest	2✔
10
11	from pants.backend.cue.goals.fix import CueFmtRequest, rules	2✔
12	from pants.backend.cue.target_types import CueFieldSet, CuePackageTarget	2✔
13	from pants.core.goals.fmt import FmtResult, Partitions	2✔
14	from pants.core.util_rules import external_tool, source_files	2✔
15	from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest	2✔
16	from pants.engine.addresses import AddressInput	2✔
17	from pants.engine.fs import DigestContents	2✔
18	from pants.testutil.rule_runner import QueryRule, RuleRunner	2✔
19
20
21	@pytest.fixture	2✔
22	def rule_runner() -> RuleRunner:	2✔
23	return RuleRunner(	2✔
24	rules=[
25	*rules(),
26	*external_tool.rules(),
27	*source_files.rules(),
28	QueryRule(Partitions, [CueFmtRequest.PartitionRequest]),
29	QueryRule(FmtResult, [CueFmtRequest.Batch]),
30	QueryRule(SourceFiles, (SourceFilesRequest,)),
31	],
32	target_types=[CuePackageTarget],
33	)
34
35
36	def run_cue(	2✔
37	rule_runner: RuleRunner,
38	addresses: Iterable[str],
39	*,
40	extra_args: list[str] \| None = None,
41	) -> tuple[FmtResult, ...]:
42	targets = [	2✔
43	rule_runner.get_target(
44	AddressInput.parse(address, description_of_origin="cue tests").file_to_address()
45	)
46	for address in addresses
47	]
48	rule_runner.set_options(	2✔
49	extra_args or (),
50	)
51	field_sets = [CueFieldSet.create(tgt) for tgt in targets]	2✔
52	kwargs = {}	2✔
53	partitions = rule_runner.request(	2✔
54	Partitions, [CueFmtRequest.PartitionRequest(tuple(field_sets))]
55	)
56	input_sources = rule_runner.request(	2✔
57	SourceFiles, [SourceFilesRequest(field_set.sources for field_set in field_sets)]
58	)
59	kwargs["snapshot"] = input_sources.snapshot	2✔
60	assert len(partitions) == 1	2✔
61	results = []	2✔
62	for partition in partitions:	2✔
63	result = rule_runner.request(	2✔
64	FmtResult,
65	[CueFmtRequest.Batch("cue", partition.elements, partition.metadata, **kwargs)],
66	)
67	assert result.tool_name == "cue"	2✔
68	results.append(result)	2✔
69	return tuple(results)	2✔
70
71
72	ExpectedResult = namedtuple("ExpectedResult", "stdout, stderr, files", defaults=("", "", ()))	2✔
73
74
75	def assert_results(	2✔
76	rule_runner: RuleRunner,
77	results: tuple[FmtResult, ...],
78	*expected_results: ExpectedResult,
79	) -> None:
80	assert len(results) == len(expected_results)	2✔
81	for result, expected in zip(results, expected_results):	2✔
82	assert result.stdout == expected.stdout	2✔
83	assert result.stderr == expected.stderr	2✔
84	for filename, contents in expected.files:	2✔
85	fc = next(	2✔
86	fc
87	for fc in rule_runner.request(DigestContents, [result.output.digest])
88	if fc.path == filename
89	)
90	assert fc.content.decode() == contents	2✔
91
92
93	def test_simple_cue_fmt(rule_runner: RuleRunner) -> None:	2✔
94	rule_runner.write_files(	×
95	{
96	"src/BUILD": "cue_package()",
97	"src/example.cue": dedent(
98	"""\
99	package example
100
101	config: "value"
102	"""
103	),
104	}
105	)
106	assert_results(	×
107	rule_runner,
108	run_cue(rule_runner, ["src/example.cue"]),
109	ExpectedResult(),
110	)
111
112
113	@pytest.mark.platform_specific_behavior	2✔
114	def test_simple_cue_fmt_issue(rule_runner: RuleRunner) -> None:	2✔
115	rule_runner.write_files(	2✔
116	{
117	"src/BUILD": "cue_package()",
118	"src/example.cue": dedent(
119	"""\
120	package example
121
122	config:{
123	config: 42
124	}
125	"""
126	),
127	}
128	)
129	assert_results(	2✔
130	rule_runner,
131	run_cue(rule_runner, ["src/example.cue"]),
132	# `cue fmt` does not output anything. so we have only the formatted files to go on. :/
133	ExpectedResult(
134	files=[
135	(
136	"src/example.cue",
137	dedent(
138	"""\
139	package example
140
141	config: {
142	\tconfig: 42
143	}
144	"""
145	),
146	)
147	]
148	),
149	)

pantsbuild / pants / 25259185675

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