21531268767

Committed 30 Jan 2026 09:29PM UTC coverage: 80.331%. First build

Build # 21531268767

Build Type

Pull #23057

github

Committed by

web-flow

Commit Message

Merge 7f2eac11a into 78e2689de

Pull Request Pull Request #23057: Remove MultiGet from the codebase

Run Details

16 of 20 new or added lines in 3 files covered. (80.0%)

78558 of 97793 relevant lines covered (80.33%)

3.36 hits per line

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

75.97

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

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

from __future__ import annotations

import importlib
import sys
import textwrap
from collections.abc import Iterable
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path

import pytest

from pants.base.exceptions import RuleTypeError
from pants.engine.internals.rule_visitor import collect_awaitables
from pants.engine.internals.selectors import Get, GetParseError, concurrently
from pants.engine.rules import implicitly, rule
from pants.util.strutil import softwrap

# The visitor inspects the module for definitions.
STR = str
INT = int
BOOL = bool


@rule
async def str_from_int(i: int) -> str:
    return str(i)


@rule
async def int_from_str(s: str) -> int:
    return int(s)


async def _top_helper(arg1):
    a = await str_from_int(arg1)
    return await _helper_helper(a)


async def _helper_helper(arg1):
    return await int_from_str(arg1)


class HelperContainer:
    async def _method_helper(self, arg1: int):
        return await str_from_int(**implicitly({arg1: int}))

    @staticmethod
    async def _static_helper():
        a = await str_from_int(42)
        return await _helper_helper(a)


container_instance = HelperContainer()


class InnerScope:
    STR = str
    INT = int
    BOOL = bool

    HelperContainer = HelperContainer
    container_instance = container_instance


OutT = type
InT = type


def assert_awaitables(func, awaitable_types: Iterable[tuple[OutT, InT | list[InT]]]):
    gets = collect_awaitables(func)
    actual_types = tuple((get.output_type, list(get.input_types)) for get in gets)
    expected_types = tuple(
        (output, ([input_] if isinstance(input_, type) else input_))
        for output, input_ in awaitable_types
    )
    assert actual_types == expected_types


def assert_byname_awaitables(func, awaitable_types: Iterable[tuple[OutT, InT | list[InT], int]]):
    gets = collect_awaitables(func)
    actual_types = tuple(
        (get.output_type, list(get.input_types), get.explicit_args_arity) for get in gets
    )
    expected_types = tuple(
        (output, ([input_] if isinstance(input_, type) else input_), explicit_args_arity)
        for output, input_, explicit_args_arity in awaitable_types
    )
    assert actual_types == expected_types


@pytest.mark.call_by_type
def test_single_get() -> None:
    async def rule():
        await Get(STR, INT, 42)

    assert_awaitables(rule, [(str, int)])


@pytest.mark.call_by_type
def test_single_no_args_syntax() -> None:
    async def rule():
        await Get(STR)

    assert_awaitables(rule, [(str, [])])


@pytest.mark.call_by_type
def test_get_multi_param_syntax() -> None:
    async def rule():
        await Get(str, {42: int, "towel": str})

    assert_awaitables(rule, [(str, [int, str])])


@pytest.mark.call_by_type
def test_multiple_gets() -> None:
    async def rule():
        a = await Get(STR, INT, 42)
        if len(a) > 1:
            await Get(BOOL, STR("bob"))

    assert_awaitables(rule, [(str, int), (bool, str)])


@pytest.mark.call_by_type
def test_multiget_homogeneous() -> None:
    async def rule():
        await concurrently(Get(STR, INT(x)) for x in range(5))

    assert_awaitables(rule, [(str, int)])


@pytest.mark.call_by_type
def test_multiget_heterogeneous() -> None:
    async def rule():
        await concurrently(Get(STR, INT, 42), Get(INT, STR("bob")))

    assert_awaitables(rule, [(str, int), (int, str)])


@pytest.mark.call_by_type
def test_attribute_lookup() -> None:
    async def rule1():
        await Get(InnerScope.STR, InnerScope.INT, 42)
        await Get(InnerScope.STR, InnerScope.INT(42))

    assert_awaitables(rule1, [(str, int), (str, int)])


@pytest.mark.call_by_type
def test_get_no_index_call_no_subject_call_allowed() -> None:
    async def rule() -> None:
        get_type: type = Get  # noqa: F841

    assert_awaitables(rule, [])


def test_byname() -> None:
    @rule
    async def rule0() -> int:
        return 11

    @rule
    async def rule1(arg: int) -> int:
        return arg

    @rule
    async def rule2(arg1: float, arg2: str) -> int:
        return int(arg1) + int(arg2)

    async def rule3() -> int:
        r0 = await rule0()
        r1_explicit = await rule1(22)
        r1_implicit = await rule1(**implicitly(int(23)))
        r2_explicit = await rule2(33.3, "44")
        r2_implicit = await rule2(**implicitly({33.4: float, "45": str}))
        r2_mixed = await rule2(33.5, **implicitly({"45": str}))
        return r0 + r1_explicit + r1_implicit + r2_explicit + r2_implicit + r2_mixed

    assert_byname_awaitables(
        rule3,
        [
            (int, [], 0),
            (int, [], 1),
            (int, int, 0),
            (int, [], 2),
            (int, [float, str], 0),
            (int, [str], 1),
        ],
    )


@contextmanager
def temporary_module(tmp_path: Path, rule_code: str):
    module_name = "_temp_module"
    src_file = tmp_path / f"{module_name}.py"
    with open(src_file, "w") as fp:
        fp.write(rule_code)
    spec = importlib.util.spec_from_file_location(module_name, src_file)
    assert spec
    assert spec.loader
    module = importlib.util.module_from_spec(spec)
    assert module
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    yield module
    del sys.modules[module_name]


def test_byname_recursion(tmp_path: Path) -> None:
    # Note that it's important that the rule is defined inside this function, so that
    # the @rule decorator is evaluated at test runtime, and not test file parse time.
    # However recursion is only supported for rules at module scope, so we have to
    # jump through some hoops to create a module at runtime.

    rule_code = textwrap.dedent("""
        from pants.engine.rules import rule

        @rule
        async def recursive_rule(arg: int) -> int:
            if arg == 0:
                return 0
            recursive = await recursive_rule(arg - 1)
            return recursive
    """)
    with temporary_module(tmp_path, rule_code) as module:
        assert_byname_awaitables(module.recursive_rule, [(int, [], 1)])


def test_byname_mutual_recursion(tmp_path: Path) -> None:
    # Note that it's important that the rules are defined inside this function, so that
    # the @rule decorators are evaluated at test runtime, and not test file parse time.
    # However recursion is only supported for rules at module scope, so we have to
    # jump through some hoops to create a module at runtime.

    rule_code = textwrap.dedent("""
        from pants.engine.rules import rule

        @rule
        async def mutually_recursive_rule_1(arg: str) -> int:
            if arg == "0":
                return 0
            recursive = await mutually_recursive_rule_2(int(arg) - 1)
            return int(recursive)

        @rule
        async def mutually_recursive_rule_2(arg: int) -> str:
            recursive = await mutually_recursive_rule_1(str(arg - 1))
            return str(recursive)
    """)

    with temporary_module(tmp_path, rule_code) as module:
        assert_byname_awaitables(module.mutually_recursive_rule_1, [(str, [], 1)])
        assert_byname_awaitables(module.mutually_recursive_rule_2, [(int, [], 1)])


def test_rule_helpers_free_functions() -> None:
    async def rule():
        _top_helper(1)

    assert_byname_awaitables(rule, [(str, [], 1), (int, [], 1)])


def test_rule_helpers_class_methods() -> None:
    async def rule1():
        HelperContainer()._static_helper(1)

    async def rule1_inner():
        InnerScope.HelperContainer()._static_helper(1)

    async def rule2():
        HelperContainer._static_helper(1)

    async def rule2_inner():
        InnerScope.HelperContainer._static_helper(1)

    async def rule3():
        container_instance._static_helper(1)

    async def rule3_inner():
        InnerScope.container_instance._static_helper(1)

    async def rule4():
        container_instance._method_helper(1)

    async def rule4_inner():
        InnerScope.container_instance._method_helper(1)

    # Rule helpers must be called via module-scoped attribute lookup
    assert_byname_awaitables(rule1, [])
    assert_byname_awaitables(rule1_inner, [])
    assert_byname_awaitables(rule2, [(str, [], 1), (int, [], 1)])
    assert_byname_awaitables(rule2_inner, [(str, [], 1), (int, [], 1)])
    assert_byname_awaitables(rule3, [(str, [], 1), (int, [], 1)])
    assert_byname_awaitables(rule3_inner, [(str, [], 1), (int, [], 1)])
    assert_byname_awaitables(rule4, [(str, [int], 0)])
    assert_byname_awaitables(rule4_inner, [(str, int, 0)])


@pytest.mark.call_by_type
def test_valid_get_unresolvable_product_type() -> None:
    async def rule():
        Get(DNE, STR(42))  # noqa: F821

    with pytest.raises(RuleTypeError, match="Could not resolve type for `DNE` in module"):
        collect_awaitables(rule)


@pytest.mark.call_by_type
def test_valid_get_unresolvable_subject_declared_type() -> None:
    async def rule():
        Get(int, DNE, "bob")  # noqa: F821

    with pytest.raises(RuleTypeError, match="Could not resolve type for `DNE` in module"):
        collect_awaitables(rule)


@pytest.mark.call_by_type
def test_invalid_get_no_args() -> None:
    async def rule():
        Get()

    with pytest.raises(GetParseError):
        collect_awaitables(rule)


@pytest.mark.call_by_type
def test_invalid_get_too_many_subject_args() -> None:
    async def rule():
        Get(STR, INT, "bob", 3)

    with pytest.raises(GetParseError):
        collect_awaitables(rule)


@pytest.mark.call_by_type
def test_invalid_get_invalid_subject_arg_no_constructor_call() -> None:
    async def rule():
        Get(STR, "bob")

    with pytest.raises(RuleTypeError, match="Expected a type, but got: (Str|Constant) 'bob'"):
        collect_awaitables(rule)


@pytest.mark.call_by_type
def test_invalid_get_invalid_product_type_not_a_type_name() -> None:
    async def rule():
        Get(call(), STR("bob"))  # noqa: F821

    with pytest.raises(RuleTypeError, match="Expected a type, but got: Call 'call'"):
        collect_awaitables(rule)


@pytest.mark.call_by_type
def test_invalid_get_dict_value_not_type() -> None:
    async def rule():
        Get(int, {"str": "not a type"})

    with pytest.raises(
        RuleTypeError, match="Expected a type, but got: (Str|Constant) 'not a type'"
    ):
        collect_awaitables(rule)


@dataclass(frozen=True)
class Request:
    arg1: str
    arg2: float

    async def _helped(self) -> Request:
        return self

    @staticmethod
    def create_get() -> Get:
        return Get(Request, int)

    def bad_meth(self):
        return Request("uh", 4.2)


@pytest.mark.call_by_type
def test_deep_infer_types() -> None:
    async def rule(request: Request):
        # 1
        r = await request._helped()
        Get(int, r.arg1)
        # 2
        s = request.arg2
        Get(bool, s)
        # 3, 4
        a, b = await concurrently(
            Get(list, str),
            Get(tuple, str),
        )
        # 5
        Get(dict, a)
        # 6
        Get(dict, b)
        # 7 -- this is huge!
        c = Request.create_get()
        # 8 -- the `c` is already accounted for, make sure it's not duplicated.
        await concurrently([c, Get(str, dict)])
        # 9
        Get(float, request._helped())

    assert_awaitables(
        rule,
        [
            (int, str),  # 1
            (bool, float),  # 2
            (list, str),  # 3
            (tuple, str),  # 4
            (dict, list),  # 5
            (dict, tuple),  # 6
            (Request, int),  # 7
            (str, dict),  # 8
            (float, Request),  # 9
        ],
    )


@pytest.mark.call_by_type
def test_missing_type_annotation() -> None:
    async def myrule(request: Request):
        Get(str, request.bad_meth())

    err = softwrap(
        r"""
        /.*/rule_visitor_test\.py:\d+: Could not resolve type for `request\.bad_meth`
        in module pants\.engine\.internals\.rule_visitor_test\.

        Failed to look up return type hint for `bad_meth` in /.*/rule_visitor_test\.py:\d+
        """
    )
    with pytest.raises(RuleTypeError, match=err):
        collect_awaitables(myrule)


@pytest.mark.call_by_type
def test_closure() -> None:
    def closure_func() -> int:
        return 44

    async def myrule(request: Request):
        Get(str, closure_func())

    assert_awaitables(myrule, [(str, int)])


class McUnion:
    b: bool
    v: int | float


@pytest.mark.call_by_type
def test_union_types() -> None:
    async def somerule(mc: McUnion):
        Get(str, mc.b)

    assert_awaitables(somerule, [(str, bool)])

1	# Copyright 2022 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 importlib	1✔
7	import sys	1✔
8	import textwrap	1✔
9	from collections.abc import Iterable	1✔
10	from contextlib import contextmanager	1✔
11	from dataclasses import dataclass	1✔
12	from pathlib import Path	1✔
13
14	import pytest	1✔
15
16	from pants.base.exceptions import RuleTypeError	1✔
17	from pants.engine.internals.rule_visitor import collect_awaitables	1✔
18	from pants.engine.internals.selectors import Get, GetParseError, concurrently	1✔
19	from pants.engine.rules import implicitly, rule	1✔
20	from pants.util.strutil import softwrap	1✔
21
22	# The visitor inspects the module for definitions.
23	STR = str	1✔
24	INT = int	1✔
25	BOOL = bool	1✔
26
27
28	@rule	1✔
29	async def str_from_int(i: int) -> str:	1✔
30	return str(i)	×
31
32
33	@rule	1✔
34	async def int_from_str(s: str) -> int:	1✔
35	return int(s)	×
36
37
38	async def _top_helper(arg1):	1✔
39	a = await str_from_int(arg1)	×
40	return await _helper_helper(a)	×
41
42
43	async def _helper_helper(arg1):	1✔
44	return await int_from_str(arg1)	×
45
46
47	class HelperContainer:	1✔
48	async def _method_helper(self, arg1: int):	1✔
49	return await str_from_int(**implicitly({arg1: int}))	×
50
51	@staticmethod	1✔
52	async def _static_helper():	1✔
53	a = await str_from_int(42)	×
54	return await _helper_helper(a)	×
55
56
57	container_instance = HelperContainer()	1✔
58
59
60	class InnerScope:	1✔
61	STR = str	1✔
62	INT = int	1✔
63	BOOL = bool	1✔
64
65	HelperContainer = HelperContainer	1✔
66	container_instance = container_instance	1✔
67
68
69	OutT = type	1✔
70	InT = type	1✔
71
72
73	def assert_awaitables(func, awaitable_types: Iterable[tuple[OutT, InT \| list[InT]]]):	1✔
74	gets = collect_awaitables(func)	1✔
75	actual_types = tuple((get.output_type, list(get.input_types)) for get in gets)	1✔
76	expected_types = tuple(	1✔
77	(output, ([input_] if isinstance(input_, type) else input_))
78	for output, input_ in awaitable_types
79	)
80	assert actual_types == expected_types	1✔
81
82
83	def assert_byname_awaitables(func, awaitable_types: Iterable[tuple[OutT, InT \| list[InT], int]]):	1✔
84	gets = collect_awaitables(func)	1✔
85	actual_types = tuple(	1✔
86	(get.output_type, list(get.input_types), get.explicit_args_arity) for get in gets
87	)
88	expected_types = tuple(	1✔
89	(output, ([input_] if isinstance(input_, type) else input_), explicit_args_arity)
90	for output, input_, explicit_args_arity in awaitable_types
91	)
92	assert actual_types == expected_types	1✔
93
94
95	@pytest.mark.call_by_type	1✔
96	def test_single_get() -> None:	1✔
97	async def rule():	1✔
98	await Get(STR, INT, 42)	×
99
100	assert_awaitables(rule, [(str, int)])	1✔
101
102
103	@pytest.mark.call_by_type	1✔
104	def test_single_no_args_syntax() -> None:	1✔
105	async def rule():	1✔
106	await Get(STR)	×
107
108	assert_awaitables(rule, [(str, [])])	1✔
109
110
111	@pytest.mark.call_by_type	1✔
112	def test_get_multi_param_syntax() -> None:	1✔
113	async def rule():	1✔
114	await Get(str, {42: int, "towel": str})	×
115
116	assert_awaitables(rule, [(str, [int, str])])	1✔
117
118
119	@pytest.mark.call_by_type	1✔
120	def test_multiple_gets() -> None:	1✔
121	async def rule():	1✔
122	a = await Get(STR, INT, 42)	×
123	if len(a) > 1:	×
124	await Get(BOOL, STR("bob"))	×
125
126	assert_awaitables(rule, [(str, int), (bool, str)])	1✔
127
128
129	@pytest.mark.call_by_type	1✔
130	def test_multiget_homogeneous() -> None:	1✔
131	async def rule():	1✔
NEW 132	await concurrently(Get(STR, INT(x)) for x in range(5))	×
133
134	assert_awaitables(rule, [(str, int)])	1✔
135
136
137	@pytest.mark.call_by_type	1✔
138	def test_multiget_heterogeneous() -> None:	1✔
139	async def rule():	1✔
NEW 140	await concurrently(Get(STR, INT, 42), Get(INT, STR("bob")))	×
141
142	assert_awaitables(rule, [(str, int), (int, str)])	1✔
143
144
145	@pytest.mark.call_by_type	1✔
146	def test_attribute_lookup() -> None:	1✔
147	async def rule1():	1✔
148	await Get(InnerScope.STR, InnerScope.INT, 42)	×
149	await Get(InnerScope.STR, InnerScope.INT(42))	×
150
151	assert_awaitables(rule1, [(str, int), (str, int)])	1✔
152
153
154	@pytest.mark.call_by_type	1✔
155	def test_get_no_index_call_no_subject_call_allowed() -> None:	1✔
156	async def rule() -> None:	1✔
157	get_type: type = Get # noqa: F841	×
158
159	assert_awaitables(rule, [])	1✔
160
161
162	def test_byname() -> None:	1✔
163	@rule	1✔
164	async def rule0() -> int:	1✔
165	return 11	×
166
167	@rule	1✔
168	async def rule1(arg: int) -> int:	1✔
169	return arg	×
170
171	@rule	1✔
172	async def rule2(arg1: float, arg2: str) -> int:	1✔
173	return int(arg1) + int(arg2)	×
174
175	async def rule3() -> int:	1✔
176	r0 = await rule0()	×
177	r1_explicit = await rule1(22)	×
178	r1_implicit = await rule1(**implicitly(int(23)))	×
179	r2_explicit = await rule2(33.3, "44")	×
180	r2_implicit = await rule2(**implicitly({33.4: float, "45": str}))	×
181	r2_mixed = await rule2(33.5, **implicitly({"45": str}))	×
182	return r0 + r1_explicit + r1_implicit + r2_explicit + r2_implicit + r2_mixed	×
183
184	assert_byname_awaitables(	1✔
185	rule3,
186	[
187	(int, [], 0),
188	(int, [], 1),
189	(int, int, 0),
190	(int, [], 2),
191	(int, [float, str], 0),
192	(int, [str], 1),
193	],
194	)
195
196
197	@contextmanager	1✔
198	def temporary_module(tmp_path: Path, rule_code: str):	1✔
199	module_name = "_temp_module"	1✔
200	src_file = tmp_path / f"{module_name}.py"	1✔
201	with open(src_file, "w") as fp:	1✔
202	fp.write(rule_code)	1✔
203	spec = importlib.util.spec_from_file_location(module_name, src_file)	1✔
204	assert spec	1✔
205	assert spec.loader	1✔
206	module = importlib.util.module_from_spec(spec)	1✔
207	assert module	1✔
208	sys.modules[module_name] = module	1✔
209	spec.loader.exec_module(module)	1✔
210	yield module	1✔
211	del sys.modules[module_name]	1✔
212
213
214	def test_byname_recursion(tmp_path: Path) -> None:	1✔
215	# Note that it's important that the rule is defined inside this function, so that
216	# the @rule decorator is evaluated at test runtime, and not test file parse time.
217	# However recursion is only supported for rules at module scope, so we have to
218	# jump through some hoops to create a module at runtime.
219
220	rule_code = textwrap.dedent("""	1✔
221	from pants.engine.rules import rule
222
223	@rule
224	async def recursive_rule(arg: int) -> int:
225	if arg == 0:
226	return 0
227	recursive = await recursive_rule(arg - 1)
228	return recursive
229	""")
230	with temporary_module(tmp_path, rule_code) as module:	1✔
231	assert_byname_awaitables(module.recursive_rule, [(int, [], 1)])	1✔
232
233
234	def test_byname_mutual_recursion(tmp_path: Path) -> None:	1✔
235	# Note that it's important that the rules are defined inside this function, so that
236	# the @rule decorators are evaluated at test runtime, and not test file parse time.
237	# However recursion is only supported for rules at module scope, so we have to
238	# jump through some hoops to create a module at runtime.
239
240	rule_code = textwrap.dedent("""	1✔
241	from pants.engine.rules import rule
242
243	@rule
244	async def mutually_recursive_rule_1(arg: str) -> int:
245	if arg == "0":
246	return 0
247	recursive = await mutually_recursive_rule_2(int(arg) - 1)
248	return int(recursive)
249
250	@rule
251	async def mutually_recursive_rule_2(arg: int) -> str:
252	recursive = await mutually_recursive_rule_1(str(arg - 1))
253	return str(recursive)
254	""")
255
256	with temporary_module(tmp_path, rule_code) as module:	1✔
257	assert_byname_awaitables(module.mutually_recursive_rule_1, [(str, [], 1)])	1✔
258	assert_byname_awaitables(module.mutually_recursive_rule_2, [(int, [], 1)])	1✔
259
260
261	def test_rule_helpers_free_functions() -> None:	1✔
262	async def rule():	1✔
263	_top_helper(1)	×
264
265	assert_byname_awaitables(rule, [(str, [], 1), (int, [], 1)])	1✔
266
267
268	def test_rule_helpers_class_methods() -> None:	1✔
269	async def rule1():	1✔
270	HelperContainer()._static_helper(1)	×
271
272	async def rule1_inner():	1✔
273	InnerScope.HelperContainer()._static_helper(1)	×
274
275	async def rule2():	1✔
276	HelperContainer._static_helper(1)	×
277
278	async def rule2_inner():	1✔
279	InnerScope.HelperContainer._static_helper(1)	×
280
281	async def rule3():	1✔
282	container_instance._static_helper(1)	×
283
284	async def rule3_inner():	1✔
285	InnerScope.container_instance._static_helper(1)	×
286
287	async def rule4():	1✔
288	container_instance._method_helper(1)	×
289
290	async def rule4_inner():	1✔
291	InnerScope.container_instance._method_helper(1)	×
292
293	# Rule helpers must be called via module-scoped attribute lookup
294	assert_byname_awaitables(rule1, [])	1✔
295	assert_byname_awaitables(rule1_inner, [])	1✔
296	assert_byname_awaitables(rule2, [(str, [], 1), (int, [], 1)])	1✔
297	assert_byname_awaitables(rule2_inner, [(str, [], 1), (int, [], 1)])	1✔
298	assert_byname_awaitables(rule3, [(str, [], 1), (int, [], 1)])	1✔
299	assert_byname_awaitables(rule3_inner, [(str, [], 1), (int, [], 1)])	1✔
300	assert_byname_awaitables(rule4, [(str, [int], 0)])	1✔
301	assert_byname_awaitables(rule4_inner, [(str, int, 0)])	1✔
302
303
304	@pytest.mark.call_by_type	1✔
305	def test_valid_get_unresolvable_product_type() -> None:	1✔
306	async def rule():	1✔
307	Get(DNE, STR(42)) # noqa: F821	×
308
309	with pytest.raises(RuleTypeError, match="Could not resolve type for `DNE` in module"):	1✔
310	collect_awaitables(rule)	1✔
311
312
313	@pytest.mark.call_by_type	1✔
314	def test_valid_get_unresolvable_subject_declared_type() -> None:	1✔
315	async def rule():	1✔
316	Get(int, DNE, "bob") # noqa: F821	×
317
318	with pytest.raises(RuleTypeError, match="Could not resolve type for `DNE` in module"):	1✔
319	collect_awaitables(rule)	1✔
320
321
322	@pytest.mark.call_by_type	1✔
323	def test_invalid_get_no_args() -> None:	1✔
324	async def rule():	1✔
325	Get()	×
326
327	with pytest.raises(GetParseError):	1✔
328	collect_awaitables(rule)	1✔
329
330
331	@pytest.mark.call_by_type	1✔
332	def test_invalid_get_too_many_subject_args() -> None:	1✔
333	async def rule():	1✔
334	Get(STR, INT, "bob", 3)	×
335
336	with pytest.raises(GetParseError):	1✔
337	collect_awaitables(rule)	1✔
338
339
340	@pytest.mark.call_by_type	1✔
341	def test_invalid_get_invalid_subject_arg_no_constructor_call() -> None:	1✔
342	async def rule():	1✔
343	Get(STR, "bob")	×
344
345	with pytest.raises(RuleTypeError, match="Expected a type, but got: (Str\|Constant) 'bob'"):	1✔
346	collect_awaitables(rule)	1✔
347
348
349	@pytest.mark.call_by_type	1✔
350	def test_invalid_get_invalid_product_type_not_a_type_name() -> None:	1✔
351	async def rule():	1✔
352	Get(call(), STR("bob")) # noqa: F821	×
353
354	with pytest.raises(RuleTypeError, match="Expected a type, but got: Call 'call'"):	1✔
355	collect_awaitables(rule)	1✔
356
357
358	@pytest.mark.call_by_type	1✔
359	def test_invalid_get_dict_value_not_type() -> None:	1✔
360	async def rule():	1✔
361	Get(int, {"str": "not a type"})	×
362
363	with pytest.raises(	1✔
364	RuleTypeError, match="Expected a type, but got: (Str\|Constant) 'not a type'"
365	):
366	collect_awaitables(rule)	1✔
367
368
369	@dataclass(frozen=True)	1✔
370	class Request:	1✔
371	arg1: str	1✔
372	arg2: float	1✔
373
374	async def _helped(self) -> Request:	1✔
375	return self	×
376
377	@staticmethod	1✔
378	def create_get() -> Get:	1✔
379	return Get(Request, int)	×
380
381	def bad_meth(self):	1✔
382	return Request("uh", 4.2)	×
383
384
385	@pytest.mark.call_by_type	1✔
386	def test_deep_infer_types() -> None:	1✔
387	async def rule(request: Request):	1✔
388	# 1
389	r = await request._helped()	×
390	Get(int, r.arg1)	×
391	# 2
392	s = request.arg2	×
393	Get(bool, s)	×
394	# 3, 4
NEW 395	a, b = await concurrently(	×
396	Get(list, str),
397	Get(tuple, str),
398	)
399	# 5
400	Get(dict, a)	×
401	# 6
402	Get(dict, b)	×
403	# 7 -- this is huge!
404	c = Request.create_get()	×
405	# 8 -- the `c` is already accounted for, make sure it's not duplicated.
NEW 406	await concurrently([c, Get(str, dict)])	×
407	# 9
408	Get(float, request._helped())	×
409
410	assert_awaitables(	1✔
411	rule,
412	[
413	(int, str), # 1
414	(bool, float), # 2
415	(list, str), # 3
416	(tuple, str), # 4
417	(dict, list), # 5
418	(dict, tuple), # 6
419	(Request, int), # 7
420	(str, dict), # 8
421	(float, Request), # 9
422	],
423	)
424
425
426	@pytest.mark.call_by_type	1✔
427	def test_missing_type_annotation() -> None:	1✔
428	async def myrule(request: Request):	1✔
429	Get(str, request.bad_meth())	×
430
431	err = softwrap(	1✔
432	r"""
433	/.*/rule_visitor_test\.py:\d+: Could not resolve type for `request\.bad_meth`
434	in module pants\.engine\.internals\.rule_visitor_test\.
435
436	Failed to look up return type hint for `bad_meth` in /.*/rule_visitor_test\.py:\d+
437	"""
438	)
439	with pytest.raises(RuleTypeError, match=err):	1✔
440	collect_awaitables(myrule)	1✔
441
442
443	@pytest.mark.call_by_type	1✔
444	def test_closure() -> None:	1✔
445	def closure_func() -> int:	1✔
446	return 44	×
447
448	async def myrule(request: Request):	1✔
449	Get(str, closure_func())	×
450
451	assert_awaitables(myrule, [(str, int)])	1✔
452
453
454	class McUnion:	1✔
455	b: bool	1✔
456	v: int \| float	1✔
457
458
459	@pytest.mark.call_by_type	1✔
460	def test_union_types() -> None:	1✔
461	async def somerule(mc: McUnion):	1✔
462	Get(str, mc.b)	×
463
464	assert_awaitables(somerule, [(str, bool)])	1✔

pantsbuild / pants / 21531268767

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