20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 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 / 20632486505

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