fixed FunctionInputOutputTypeChecker rejects correct values for fixed…

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

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"])

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])