tests/test_async_driver.py - hamilton - Git at Google

 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership.  The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.

 import asyncio

 import pandas as pd
 import pytest

 from hamilton import ad_hoc_utils, async_driver, base
 from hamilton.lifecycle.base import (
     BasePostGraphConstruct,
     BasePostGraphConstructAsync,
     BasePostGraphExecute,
     BasePostGraphExecuteAsync,
     BasePostNodeExecute,
     BasePostNodeExecuteAsync,
     BasePreGraphExecute,
     BasePreGraphExecuteAsync,
     BasePreNodeExecute,
     BasePreNodeExecuteAsync,
 )

 from .resources import async_dq_module, simple_async_module


 async def async_identity(n: int) -> int:
     await asyncio.sleep(0.01)
     return n


 @pytest.mark.asyncio
 async def test_await_dict_of_coroutines():
     tasks = {n: async_identity(n) for n in range(0, 10)}
     results = await async_driver.await_dict_of_tasks(tasks)
     assert results == {n: await async_identity(n) for n in range(0, 10)}


 @pytest.mark.asyncio
 async def test_await_dict_of_tasks():
     tasks = {n: asyncio.create_task(async_identity(n)) for n in range(0, 10)}
     results = await async_driver.await_dict_of_tasks(tasks)
     assert results == {n: await async_identity(n) for n in range(0, 10)}


 # The following are not parameterized as we need to use the event loop -- fixtures will complicate this
 @pytest.mark.asyncio
 async def test_process_value_raw():
     assert await async_driver.process_value(1) == 1


 @pytest.mark.asyncio
 async def test_process_value_coroutine():
     assert await async_driver.process_value(async_identity(1)) == 1


 @pytest.mark.asyncio
 async def test_process_value_task():
     assert await async_driver.process_value(asyncio.create_task(async_identity(1))) == 1


 @pytest.mark.asyncio
 async def test_driver_end_to_end():
     dr = async_driver.AsyncDriver({}, simple_async_module)
     all_vars = [var.name for var in dr.list_available_variables() if var.name != "return_df"]
     result = await dr.raw_execute(final_vars=all_vars, inputs={"external_input": 1})
     result["a"] = result["a"].to_dict()  # convert to dict for comparison
     result["b"] = result["b"].to_dict()  # convert to dict for comparison
     assert result == {
         "a": pd.Series([1, 2, 3]).to_dict(),
         "another_async_func": 8,
         "async_func_with_param": 4,
         "b": pd.Series([4, 5, 6]).to_dict(),
         "external_input": 1,
         "non_async_func_with_decorator": {"result_1": 9, "result_2": 5},
         "result_1": 9,
         "result_2": 5,
         "result_3": 1,
         "result_4": 2,
         "return_dict": {"result_3": 1, "result_4": 2},
         "simple_async_func": 2,
         "simple_non_async_func": 7,
     }


 @pytest.mark.asyncio
 async def test_async_driver_end_to_end_async_lifecycle_methods():
     tracked_calls = []

     class AsyncTrackingAdapter(
         BasePostGraphConstructAsync,
         BasePreGraphExecuteAsync,
         BasePostGraphExecuteAsync,
         BasePreNodeExecuteAsync,
         BasePostNodeExecuteAsync,
     ):
         def __init__(self, calls: list, pause_time: float = 0.01):
             self.pause_time = pause_time
             self.calls = calls

         async def _pause(self):
             return await asyncio.sleep(self.pause_time)

         async def pre_graph_execute(self, **kwargs):
             await self._pause()
             self.calls.append(("pre_graph_execute", kwargs))

         async def post_graph_execute(self, **kwargs):
             await self._pause()
             self.calls.append(("post_graph_execute", kwargs))

         async def pre_node_execute(self, **kwargs):
             await self._pause()
             self.calls.append(("pre_node_execute", kwargs))

         async def post_node_execute(self, **kwargs):
             await self._pause()
             self.calls.append(("post_node_execute", kwargs))

         async def post_graph_construct(self, **kwargs):
             await self._pause()
             self.calls.append(("post_graph_construct", kwargs))

     adapter = AsyncTrackingAdapter(tracked_calls)

     dr = await async_driver.AsyncDriver(
         {}, simple_async_module, result_builder=base.DictResult(), adapters=[adapter]
     ).ainit()
     all_vars = [var.name for var in dr.list_available_variables() if var.name != "return_df"]
     result = await dr.execute(final_vars=all_vars, inputs={"external_input": 1})
     hooks_called = [call[0] for call in tracked_calls]
     assert set(hooks_called) == {
         "pre_graph_execute",
         "post_graph_execute",
         "pre_node_execute",
         "post_node_execute",
         "post_graph_construct",
     }
     result["a"] = result["a"].to_dict()
     result["b"] = result["b"].to_dict()
     assert result == {
         "a": pd.Series([1, 2, 3]).to_dict(),
         "another_async_func": 8,
         "async_func_with_param": 4,
         "b": pd.Series([4, 5, 6]).to_dict(),
         "external_input": 1,
         "non_async_func_with_decorator": {"result_1": 9, "result_2": 5},
         "result_1": 9,
         "result_2": 5,
         "result_3": 1,
         "result_4": 2,
         "return_dict": {"result_3": 1, "result_4": 2},
         "simple_async_func": 2,
         "simple_non_async_func": 7,
     }


 @pytest.mark.asyncio
 async def test_async_driver_end_to_end_sync_lifecycle_methods():
     tracked_calls = []

     class AsyncTrackingAdapter(
         BasePostGraphConstruct,
         BasePreGraphExecute,
         BasePostGraphExecute,
         BasePreNodeExecute,
         BasePostNodeExecute,
     ):
         def __init__(self, calls: list, pause_time: float = 0.01):
             self.pause_time = pause_time
             self.calls = calls

         def pre_graph_execute(self, **kwargs):
             self.calls.append(("pre_graph_execute", kwargs))

         def post_graph_execute(self, **kwargs):
             self.calls.append(("post_graph_execute", kwargs))

         def pre_node_execute(self, **kwargs):
             self.calls.append(("pre_node_execute", kwargs))

         def post_node_execute(self, **kwargs):
             self.calls.append(("post_node_execute", kwargs))

         def post_graph_construct(self, **kwargs):
             self.calls.append(("post_graph_construct", kwargs))

     adapter = AsyncTrackingAdapter(tracked_calls)

     dr = await async_driver.AsyncDriver(
         {}, simple_async_module, result_builder=base.DictResult(), adapters=[adapter]
     ).ainit()
     all_vars = [var.name for var in dr.list_available_variables() if var.name != "return_df"]
     result = await dr.execute(final_vars=all_vars, inputs={"external_input": 1})
     hooks_called = [call[0] for call in tracked_calls]
     assert set(hooks_called) == {
         "pre_graph_execute",
         "post_graph_execute",
         "pre_node_execute",
         "post_node_execute",
         "post_graph_construct",
     }
     result["a"] = result["a"].to_dict()
     result["b"] = result["b"].to_dict()
     assert result == {
         "a": pd.Series([1, 2, 3]).to_dict(),
         "another_async_func": 8,
         "async_func_with_param": 4,
         "b": pd.Series([4, 5, 6]).to_dict(),
         "external_input": 1,
         "non_async_func_with_decorator": {"result_1": 9, "result_2": 5},
         "result_1": 9,
         "result_2": 5,
         "result_3": 1,
         "result_4": 2,
         "return_dict": {"result_3": 1, "result_4": 2},
         "simple_async_func": 2,
         "simple_non_async_func": 7,
     }


 @pytest.mark.asyncio
 async def test_async_driver_overrides():
     # This specifically tests that raw execute does not return
     # and does not compute the non-returned items
     this_list_should_be_empty = []

     async def never_run() -> int:
         this_list_should_be_empty.append("never_run")
         await asyncio.sleep(0.01)
         return 1

     async def overridden(never_run: int) -> int:
         this_list_should_be_empty.append("overridden")
         await asyncio.sleep(0.01)
         return never_run + 1

     async def will_be_run(overridden: int) -> int:
         await asyncio.sleep(0.01)
         return 1 + overridden

     mod = ad_hoc_utils.create_temporary_module(never_run, will_be_run, overridden)
     dr = await async_driver.Builder().with_modules(mod).build()
     # Testing execute as it's a part of the contract and will remain so
     # raw_execute is closer to the error, but don't want to test it instead
     res = await dr.execute(final_vars=["will_be_run"], overrides={"overridden": 3})
     assert len(this_list_should_be_empty) == 0
     assert res == {"will_be_run": 4}


 @pytest.mark.asyncio
 async def test_async_builder_result_builder_custom():
     def foo() -> int:
         return 1

     mod = ad_hoc_utils.create_temporary_module(foo)
     result_builder = base.PandasDataFrameResult()
     dr = await async_driver.Builder().with_adapters(result_builder).with_modules(mod).build()
     assert isinstance(await dr.execute(final_vars=["foo"]), pd.DataFrame)


 @pytest.mark.asyncio
 async def test_async_builder_result_builder_default():
     def foo() -> int:
         return 1

     mod = ad_hoc_utils.create_temporary_module(foo)
     dr = await async_driver.Builder().with_modules(mod).build()
     assert (await dr.execute(final_vars=["foo"])) == {"foo": 1}
     # builder.result_builder = result_builder


 @pytest.mark.asyncio
 async def test_async_builder_without_init():
     def foo() -> int:
         return 1

     class TestLifecycleMethod(BasePostGraphConstructAsync):
         def __init__(self):
             self.ran = False

         async def post_graph_construct(self, **kwargs):
             await asyncio.sleep(0.01)
             self.ran = True

     mod = ad_hoc_utils.create_temporary_module(foo)
     hook = TestLifecycleMethod()
     # non-async
     async_driver.Builder().with_modules(mod).with_adapters(hook).build_without_init()
     assert not hook.ran
     await async_driver.Builder().with_modules(mod).with_adapters(hook).build()
     assert hook.ran

     # builder.result_builder = result_builder


 @pytest.mark.asyncio
 async def test_async_builder_allow_module_overrides():
     def foo() -> int:
         return 1

     mod1 = ad_hoc_utils.create_temporary_module(foo)

     def foo() -> int:
         return 2

     mod2 = ad_hoc_utils.create_temporary_module(foo)

     # Should raise without .allow_module_overrides()
     with pytest.raises(ValueError) as e:
         await async_driver.Builder().with_modules(mod1, mod2).build()

     assert "Cannot define function foo more than once." in str(e.value)

     # build_without_init should also raise
     with pytest.raises(ValueError) as e:
         async_driver.Builder().with_modules(mod1, mod2).build_without_init()

     assert "Cannot define function foo more than once." in str(e.value)

     # Should not raise with .allow_module_overrides()
     dr = await async_driver.Builder().with_modules(mod1, mod2).allow_module_overrides().build()
     assert (await dr.execute(final_vars=["foo"])) == {"foo": 2}

     # Same with build_without_init
     dr = (
         async_driver.Builder()
         .with_modules(mod1, mod2)
         .allow_module_overrides()
         .build_without_init()
     )
     assert (await dr.execute(final_vars=["foo"])) == {"foo": 2}


 @pytest.mark.asyncio
 async def test_async_driver_with_async_validator():
     """End-to-end: async validator with AsyncDriver should execute correctly."""
     dr = async_driver.AsyncDriver({}, async_dq_module, result_builder=base.DictResult())
     result = await dr.raw_execute(final_vars=["async_validated"], inputs={})
     assert result["async_validated"] == 20  # input_value=10, *2=20


 @pytest.mark.asyncio
 async def test_async_driver_with_sync_validator():
     """End-to-end: sync validator on async function with AsyncDriver should still work."""
     dr = async_driver.AsyncDriver({}, async_dq_module, result_builder=base.DictResult())
     result = await dr.raw_execute(final_vars=["sync_validated"], inputs={})
     assert result["sync_validated"] == 12  # input_value=10, +2=12


 @pytest.mark.asyncio
 async def test_async_driver_with_mixed_validators():
     """End-to-end: mixed sync+async validators on the same node with AsyncDriver."""
     dr = async_driver.AsyncDriver({}, async_dq_module, result_builder=base.DictResult())
     result = await dr.raw_execute(final_vars=["mixed_validated"], inputs={})
     assert result["mixed_validated"] == 20  # input_value=10, +10=20
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.

	import asyncio

	import pandas as pd
	import pytest

	from hamilton import ad_hoc_utils, async_driver, base
	from hamilton.lifecycle.base import (
	BasePostGraphConstruct,
	BasePostGraphConstructAsync,
	BasePostGraphExecute,
	BasePostGraphExecuteAsync,
	BasePostNodeExecute,
	BasePostNodeExecuteAsync,
	BasePreGraphExecute,
	BasePreGraphExecuteAsync,
	BasePreNodeExecute,
	BasePreNodeExecuteAsync,
	)

	from .resources import async_dq_module, simple_async_module


	async def async_identity(n: int) -> int:
	await asyncio.sleep(0.01)
	return n


	@pytest.mark.asyncio
	async def test_await_dict_of_coroutines():
	tasks = {n: async_identity(n) for n in range(0, 10)}
	results = await async_driver.await_dict_of_tasks(tasks)
	assert results == {n: await async_identity(n) for n in range(0, 10)}


	@pytest.mark.asyncio
	async def test_await_dict_of_tasks():
	tasks = {n: asyncio.create_task(async_identity(n)) for n in range(0, 10)}
	results = await async_driver.await_dict_of_tasks(tasks)
	assert results == {n: await async_identity(n) for n in range(0, 10)}


	# The following are not parameterized as we need to use the event loop -- fixtures will complicate this
	@pytest.mark.asyncio
	async def test_process_value_raw():
	assert await async_driver.process_value(1) == 1


	@pytest.mark.asyncio
	async def test_process_value_coroutine():
	assert await async_driver.process_value(async_identity(1)) == 1


	@pytest.mark.asyncio
	async def test_process_value_task():
	assert await async_driver.process_value(asyncio.create_task(async_identity(1))) == 1


	@pytest.mark.asyncio
	async def test_driver_end_to_end():
	dr = async_driver.AsyncDriver({}, simple_async_module)
	all_vars = [var.name for var in dr.list_available_variables() if var.name != "return_df"]
	result = await dr.raw_execute(final_vars=all_vars, inputs={"external_input": 1})
	result["a"] = result["a"].to_dict() # convert to dict for comparison
	result["b"] = result["b"].to_dict() # convert to dict for comparison
	assert result == {
	"a": pd.Series([1, 2, 3]).to_dict(),
	"another_async_func": 8,
	"async_func_with_param": 4,
	"b": pd.Series([4, 5, 6]).to_dict(),
	"external_input": 1,
	"non_async_func_with_decorator": {"result_1": 9, "result_2": 5},
	"result_1": 9,
	"result_2": 5,
	"result_3": 1,
	"result_4": 2,
	"return_dict": {"result_3": 1, "result_4": 2},
	"simple_async_func": 2,
	"simple_non_async_func": 7,
	}


	@pytest.mark.asyncio
	async def test_async_driver_end_to_end_async_lifecycle_methods():
	tracked_calls = []

	class AsyncTrackingAdapter(
	BasePostGraphConstructAsync,
	BasePreGraphExecuteAsync,
	BasePostGraphExecuteAsync,
	BasePreNodeExecuteAsync,
	BasePostNodeExecuteAsync,
	):
	def __init__(self, calls: list, pause_time: float = 0.01):
	self.pause_time = pause_time
	self.calls = calls

	async def _pause(self):
	return await asyncio.sleep(self.pause_time)

	async def pre_graph_execute(self, **kwargs):
	await self._pause()
	self.calls.append(("pre_graph_execute", kwargs))

	async def post_graph_execute(self, **kwargs):
	await self._pause()
	self.calls.append(("post_graph_execute", kwargs))

	async def pre_node_execute(self, **kwargs):
	await self._pause()
	self.calls.append(("pre_node_execute", kwargs))

	async def post_node_execute(self, **kwargs):
	await self._pause()
	self.calls.append(("post_node_execute", kwargs))

	async def post_graph_construct(self, **kwargs):
	await self._pause()
	self.calls.append(("post_graph_construct", kwargs))

	adapter = AsyncTrackingAdapter(tracked_calls)

	dr = await async_driver.AsyncDriver(
	{}, simple_async_module, result_builder=base.DictResult(), adapters=[adapter]
	).ainit()
	all_vars = [var.name for var in dr.list_available_variables() if var.name != "return_df"]
	result = await dr.execute(final_vars=all_vars, inputs={"external_input": 1})
	hooks_called = [call[0] for call in tracked_calls]
	assert set(hooks_called) == {
	"pre_graph_execute",
	"post_graph_execute",
	"pre_node_execute",
	"post_node_execute",
	"post_graph_construct",
	}
	result["a"] = result["a"].to_dict()
	result["b"] = result["b"].to_dict()
	assert result == {
	"a": pd.Series([1, 2, 3]).to_dict(),
	"another_async_func": 8,
	"async_func_with_param": 4,
	"b": pd.Series([4, 5, 6]).to_dict(),
	"external_input": 1,
	"non_async_func_with_decorator": {"result_1": 9, "result_2": 5},
	"result_1": 9,
	"result_2": 5,
	"result_3": 1,
	"result_4": 2,
	"return_dict": {"result_3": 1, "result_4": 2},
	"simple_async_func": 2,
	"simple_non_async_func": 7,
	}


	@pytest.mark.asyncio
	async def test_async_driver_end_to_end_sync_lifecycle_methods():
	tracked_calls = []

	class AsyncTrackingAdapter(
	BasePostGraphConstruct,
	BasePreGraphExecute,
	BasePostGraphExecute,
	BasePreNodeExecute,
	BasePostNodeExecute,
	):
	def __init__(self, calls: list, pause_time: float = 0.01):
	self.pause_time = pause_time
	self.calls = calls

	def pre_graph_execute(self, **kwargs):
	self.calls.append(("pre_graph_execute", kwargs))

	def post_graph_execute(self, **kwargs):
	self.calls.append(("post_graph_execute", kwargs))

	def pre_node_execute(self, **kwargs):
	self.calls.append(("pre_node_execute", kwargs))

	def post_node_execute(self, **kwargs):
	self.calls.append(("post_node_execute", kwargs))

	def post_graph_construct(self, **kwargs):
	self.calls.append(("post_graph_construct", kwargs))

	adapter = AsyncTrackingAdapter(tracked_calls)

	dr = await async_driver.AsyncDriver(
	{}, simple_async_module, result_builder=base.DictResult(), adapters=[adapter]
	).ainit()
	all_vars = [var.name for var in dr.list_available_variables() if var.name != "return_df"]
	result = await dr.execute(final_vars=all_vars, inputs={"external_input": 1})
	hooks_called = [call[0] for call in tracked_calls]
	assert set(hooks_called) == {
	"pre_graph_execute",
	"post_graph_execute",
	"pre_node_execute",
	"post_node_execute",
	"post_graph_construct",
	}
	result["a"] = result["a"].to_dict()
	result["b"] = result["b"].to_dict()
	assert result == {
	"a": pd.Series([1, 2, 3]).to_dict(),
	"another_async_func": 8,
	"async_func_with_param": 4,
	"b": pd.Series([4, 5, 6]).to_dict(),
	"external_input": 1,
	"non_async_func_with_decorator": {"result_1": 9, "result_2": 5},
	"result_1": 9,
	"result_2": 5,
	"result_3": 1,
	"result_4": 2,
	"return_dict": {"result_3": 1, "result_4": 2},
	"simple_async_func": 2,
	"simple_non_async_func": 7,
	}


	@pytest.mark.asyncio
	async def test_async_driver_overrides():
	# This specifically tests that raw execute does not return
	# and does not compute the non-returned items
	this_list_should_be_empty = []

	async def never_run() -> int:
	this_list_should_be_empty.append("never_run")
	await asyncio.sleep(0.01)
	return 1

	async def overridden(never_run: int) -> int:
	this_list_should_be_empty.append("overridden")
	await asyncio.sleep(0.01)
	return never_run + 1

	async def will_be_run(overridden: int) -> int:
	await asyncio.sleep(0.01)
	return 1 + overridden

	mod = ad_hoc_utils.create_temporary_module(never_run, will_be_run, overridden)
	dr = await async_driver.Builder().with_modules(mod).build()
	# Testing execute as it's a part of the contract and will remain so
	# raw_execute is closer to the error, but don't want to test it instead
	res = await dr.execute(final_vars=["will_be_run"], overrides={"overridden": 3})
	assert len(this_list_should_be_empty) == 0
	assert res == {"will_be_run": 4}


	@pytest.mark.asyncio
	async def test_async_builder_result_builder_custom():
	def foo() -> int:
	return 1

	mod = ad_hoc_utils.create_temporary_module(foo)
	result_builder = base.PandasDataFrameResult()
	dr = await async_driver.Builder().with_adapters(result_builder).with_modules(mod).build()
	assert isinstance(await dr.execute(final_vars=["foo"]), pd.DataFrame)


	@pytest.mark.asyncio
	async def test_async_builder_result_builder_default():
	def foo() -> int:
	return 1

	mod = ad_hoc_utils.create_temporary_module(foo)
	dr = await async_driver.Builder().with_modules(mod).build()
	assert (await dr.execute(final_vars=["foo"])) == {"foo": 1}
	# builder.result_builder = result_builder


	@pytest.mark.asyncio
	async def test_async_builder_without_init():
	def foo() -> int:
	return 1

	class TestLifecycleMethod(BasePostGraphConstructAsync):
	def __init__(self):
	self.ran = False

	async def post_graph_construct(self, **kwargs):
	await asyncio.sleep(0.01)
	self.ran = True

	mod = ad_hoc_utils.create_temporary_module(foo)
	hook = TestLifecycleMethod()
	# non-async
	async_driver.Builder().with_modules(mod).with_adapters(hook).build_without_init()
	assert not hook.ran
	await async_driver.Builder().with_modules(mod).with_adapters(hook).build()
	assert hook.ran

	# builder.result_builder = result_builder


	@pytest.mark.asyncio
	async def test_async_builder_allow_module_overrides():
	def foo() -> int:
	return 1

	mod1 = ad_hoc_utils.create_temporary_module(foo)

	def foo() -> int:
	return 2

	mod2 = ad_hoc_utils.create_temporary_module(foo)

	# Should raise without .allow_module_overrides()
	with pytest.raises(ValueError) as e:
	await async_driver.Builder().with_modules(mod1, mod2).build()

	assert "Cannot define function foo more than once." in str(e.value)

	# build_without_init should also raise
	with pytest.raises(ValueError) as e:
	async_driver.Builder().with_modules(mod1, mod2).build_without_init()

	assert "Cannot define function foo more than once." in str(e.value)

	# Should not raise with .allow_module_overrides()
	dr = await async_driver.Builder().with_modules(mod1, mod2).allow_module_overrides().build()
	assert (await dr.execute(final_vars=["foo"])) == {"foo": 2}

	# Same with build_without_init
	dr = (
	async_driver.Builder()
	.with_modules(mod1, mod2)
	.allow_module_overrides()
	.build_without_init()
	)
	assert (await dr.execute(final_vars=["foo"])) == {"foo": 2}


	@pytest.mark.asyncio
	async def test_async_driver_with_async_validator():
	"""End-to-end: async validator with AsyncDriver should execute correctly."""
	dr = async_driver.AsyncDriver({}, async_dq_module, result_builder=base.DictResult())
	result = await dr.raw_execute(final_vars=["async_validated"], inputs={})
	assert result["async_validated"] == 20 # input_value=10, *2=20


	@pytest.mark.asyncio
	async def test_async_driver_with_sync_validator():
	"""End-to-end: sync validator on async function with AsyncDriver should still work."""
	dr = async_driver.AsyncDriver({}, async_dq_module, result_builder=base.DictResult())
	result = await dr.raw_execute(final_vars=["sync_validated"], inputs={})
	assert result["sync_validated"] == 12 # input_value=10, +2=12


	@pytest.mark.asyncio
	async def test_async_driver_with_mixed_validators():
	"""End-to-end: mixed sync+async validators on the same node with AsyncDriver."""
	dr = async_driver.AsyncDriver({}, async_dq_module, result_builder=base.DictResult())
	result = await dr.raw_execute(final_vars=["mixed_validated"], inputs={})
	assert result["mixed_validated"] == 20 # input_value=10, +10=20