diff --git a/hamilton/htypes.py b/hamilton/htypes.py index 38fc1a419..78525832d 100644 --- a/hamilton/htypes.py +++ b/hamilton/htypes.py @@ -443,10 +443,15 @@ def check_instance(obj: Any, type_: Any) -> bool: return False # If the type has arguments (i.e., it's a parameterized generic type like List[int]) - if hasattr(type_, "__args__"): + if hasattr(type_, "__args__") and type_.__args__ is not None: # Get the element type(s) of the generic type element_type = type_.__args__ + if origin is tuple and not (len(element_type) == 2 and element_type[1] is Ellipsis): + if not isinstance(obj, tuple) or len(obj) != len(element_type): + return False + return all(check_instance(v, t) for v, t in zip(obj, element_type, strict=True)) + # If the object is a dictionary if isinstance(obj, dict): all_items_meet_condition = True diff --git a/tests/lifecycle/test_default.py b/tests/lifecycle/test_default.py index 30e4ef9c8..1e6d74bf2 100644 --- a/tests/lifecycle/test_default.py +++ b/tests/lifecycle/test_default.py @@ -15,6 +15,8 @@ # specific language governing permissions and limitations # under the License. +from typing import Any + import pytest from hamilton import ad_hoc_utils, driver @@ -68,3 +70,94 @@ def evens(n: int) -> list[int] | None: ) with pytest.raises(TypeError, match="Node evens returned a result"): dr.execute(["evens"], inputs={"n": 3}) + + +def test_function_input_output_type_checker_handles_fixed_length_tuple(): + def get_pair() -> tuple[int, str]: + return (42, "hello") + + def format_pair(get_pair: tuple[int, str]) -> str: + num, text = get_pair + return f"{text}:{num}" + + dr = ( + driver.Builder() + .with_modules(ad_hoc_utils.create_temporary_module(get_pair, format_pair)) + .with_adapters(default.FunctionInputOutputTypeChecker()) + .build() + ) + res = dr.execute(["format_pair"]) + assert res == {"format_pair": "hello:42"} + + +def test_function_input_output_type_checker_rejects_wrong_length_tuple_result(): + def get_pair() -> tuple[int, str]: + return (42,) + + dr = ( + driver.Builder() + .with_modules(ad_hoc_utils.create_temporary_module(get_pair)) + .with_adapters(default.FunctionInputOutputTypeChecker()) + .build() + ) + with pytest.raises(TypeError, match="Node get_pair returned a result"): + dr.execute(["get_pair"]) + + +def test_function_input_output_type_checker_rejects_wrong_type_tuple_result(): + def get_pair() -> tuple[int, str]: + return ("hello", 42) + + dr = ( + driver.Builder() + .with_modules(ad_hoc_utils.create_temporary_module(get_pair)) + .with_adapters(default.FunctionInputOutputTypeChecker()) + .build() + ) + with pytest.raises(TypeError, match="Node get_pair returned a result"): + dr.execute(["get_pair"]) + + +def test_function_input_output_type_checker_rejects_wrong_tuple_node_input(): + def produce_pair() -> Any: + return (42,) + + def consume_pair(produce_pair: tuple[int, str]) -> str: + return f"{produce_pair[0]}" + + dr = ( + driver.Builder() + .with_modules(ad_hoc_utils.create_temporary_module(produce_pair, consume_pair)) + .with_adapters(default.FunctionInputOutputTypeChecker(check_output=False, check_input=True)) + .build() + ) + with pytest.raises(TypeError, match="Node consume_pair received an input"): + dr.execute(["consume_pair"]) + + +def test_function_input_output_type_checker_handles_empty_tuple(): + def empty_node() -> tuple[()]: + return () + + dr = ( + driver.Builder() + .with_modules(ad_hoc_utils.create_temporary_module(empty_node)) + .with_adapters(default.FunctionInputOutputTypeChecker()) + .build() + ) + res = dr.execute(["empty_node"]) + assert res == {"empty_node": ()} + + +def test_function_input_output_type_checker_rejects_non_empty_tuple_for_empty_tuple_type(): + def empty_node() -> tuple[()]: + return (1,) + + dr = ( + driver.Builder() + .with_modules(ad_hoc_utils.create_temporary_module(empty_node)) + .with_adapters(default.FunctionInputOutputTypeChecker()) + .build() + ) + with pytest.raises(TypeError, match="Node empty_node returned a result"): + dr.execute(["empty_node"]) diff --git a/tests/test_type_utils.py b/tests/test_type_utils.py index 15bbe5a26..7aeb39a0d 100644 --- a/tests/test_type_utils.py +++ b/tests/test_type_utils.py @@ -470,3 +470,59 @@ class RandomObject: assert not check_instance( RandomObject(), Union[int, str, list, dict, None, float, bool, bytes, complex, type(None)] ) + + +def test_check_instance_with_fixed_length_tuple(): + # 1. Correct fixed-length tuple + assert check_instance((1, "a"), tuple[int, str]) + assert check_instance((1, "a", True), tuple[int, str, bool]) + assert check_instance((42,), tuple[int]) + assert check_instance((1.0, 2.0, 3.0), typing.Tuple[float, float, float]) + + # 2. Wrong-length tuple + assert not check_instance((1,), tuple[int, str]) + assert not check_instance((1, "a", True), tuple[int, str]) + assert not check_instance((), tuple[int, str]) + assert not check_instance((1, 2), tuple[int]) + assert not check_instance((1.0, 2.0), typing.Tuple[float, float, float]) + + # 3. Tuple with incorrect types in one or more positions + assert not check_instance(("a", 1), tuple[int, str]) + assert not check_instance((1, 2), tuple[int, str]) + assert not check_instance(("a", "b"), tuple[int, str]) + assert not check_instance((1, "a", 3), tuple[int, str, bool]) + + # Value not a tuple + assert not check_instance([1, "a"], tuple[int, str]) + + +def test_check_instance_with_empty_tuple_type(): + # 4. tuple[()] and typing.Tuple[()] + assert check_instance((), tuple[()]) + assert check_instance((), typing.Tuple[()]) + assert not check_instance((1,), tuple[()]) + assert not check_instance((1,), typing.Tuple[()]) + assert not check_instance((1, 2), tuple[()]) + assert not check_instance([], tuple[()]) + assert not check_instance(None, tuple[()]) + + +def test_check_instance_with_variadic_tuple(): + # 5. Existing variadic tuple behavior + assert check_instance((1, 2, 3), tuple[int, ...]) + assert check_instance((), tuple[int, ...]) + assert check_instance((1,), tuple[int, ...]) + assert not check_instance((1, "a"), tuple[int, ...]) + assert not check_instance([1, 2, 3], tuple[int, ...]) + + assert check_instance(("x", "y"), typing.Tuple[str, ...]) + assert not check_instance(("x", 1), typing.Tuple[str, ...]) + + +def test_check_instance_with_nested_and_unsubscripted_tuple(): + assert check_instance((1, 2), typing.Tuple) + assert not check_instance([1, 2], typing.Tuple) + assert check_instance([(1, "a"), (2, "b")], list[tuple[int, str]]) + assert not check_instance([(1, "a"), (2, 2)], list[tuple[int, str]]) + assert check_instance(([1, 2], "ok"), tuple[list[int], str]) + assert not check_instance((["not", "int"], "ok"), tuple[list[int], str])