1010import subprocess
1111import sys
1212from collections .abc import AsyncGenerator , Generator
13- from datetime import datetime , timedelta
13+ from datetime import UTC , datetime , timedelta , timezone
1414from functools import partial , update_wrapper , wraps
15- from typing import Any
15+ from types import SimpleNamespace
16+ from typing import Any , cast
1617
1718import pytest
1819
1920from durabletask import entities , task , worker
2021from durabletask .internal import helpers , shared
2122from durabletask .internal import orchestrator_service_pb2 as pb
23+ from durabletask .internal .asyncio_loop import DurableEventLoop
2224from durabletask .serialization import JsonDataConverter
2325
26+ pytestmark = pytest .mark .filterwarnings ("error::RuntimeWarning" )
27+
2428
2529INSTANCE_ID = "coroutine-test"
2630NOW = datetime (2026 , 1 , 1 )
@@ -211,14 +215,55 @@ def wrapper(*args: Any, **kwargs: Any) -> Generator[int, None, None]:
211215 assert inspect .getgeneratorstate (created [0 ]) == inspect .GEN_CLOSED
212216
213217
214- def test_cyclic_async_wrapper_fails_instead_of_hanging () -> None :
218+ @pytest .mark .parametrize ("shape" , ["function" , "partial" , "callable" ])
219+ def test_generator_wrapper_of_async_function_is_synchronous (shape : str ) -> None :
220+ async def async_orchestrator (ctx : task .OrchestrationContext , value : int ) -> int :
221+ return value
222+
223+ @wraps (async_orchestrator )
224+ def generator (ctx : task .OrchestrationContext , value : int ) -> Generator [task .Task [Any ], Any , int ]:
225+ return (yield ctx .call_activity ("work" , input = value , return_type = int ))
226+
227+ class GeneratorCallable :
228+ def __call__ (self , ctx : task .OrchestrationContext , value : int ) -> Generator [task .Task [Any ], Any , int ]:
229+ return (yield ctx .call_activity ("work" , input = value , return_type = int ))
230+
231+ update_wrapper (GeneratorCallable .__call__ , async_orchestrator )
232+ variants : dict [str , Any ] = {
233+ "function" : generator ,
234+ "partial" : partial (generator ),
235+ "callable" : GeneratorCallable (),
236+ }
237+ fn = variants [shape ]
238+ assert not shared .is_async_callable (fn )
239+ actions = _execute (fn , [], _start ()).actions
240+ assert [a .scheduleTask .name for a in actions ] == ["work" ]
241+
242+
243+ def test_orchestrator_is_classified_once_at_registration (monkeypatch : pytest .MonkeyPatch ) -> None :
244+ async def orchestrator (ctx : task .OrchestrationContext , value : int ) -> int :
245+ return await ctx .call_activity ("work" , input = value , return_type = int )
246+
247+ registry = worker ._Registry () # pyright: ignore[reportPrivateUsage]
248+ registry .add_named_orchestrator ("test" , orchestrator )
249+ calls : list [Any ] = []
250+ original = shared .is_async_callable
251+ monkeypatch .setattr (shared , "is_async_callable" , lambda fn : calls .append (fn ) or original (fn ))
252+ executor = worker ._OrchestrationExecutor ( # pyright: ignore[reportPrivateUsage]
253+ registry , logging .getLogger (__name__ ), JsonDataConverter ())
254+ old = _start () + [helpers .new_task_scheduled_event (1 , "work" )]
255+ executor .execute (INSTANCE_ID , [], _start ())
256+ executor .execute (INSTANCE_ID , old , [helpers .new_task_completed_event (1 , "4" )])
257+ assert calls == []
258+
259+
260+ def test_cyclic_async_wrapper_is_rejected_at_registration () -> None :
215261 def wrapper (ctx : task .OrchestrationContext , value : int ) -> int :
216262 return value
217263
218264 wrapper .__wrapped__ = wrapper # type: ignore[attr-defined]
219- complete = _completion (_execute (wrapper , [], _start ()))
220- assert complete .orchestrationStatus == pb .ORCHESTRATION_STATUS_FAILED
221- assert "wrapper loop" in complete .failureDetails .errorMessage
265+ with pytest .raises (ValueError , match = "wrapper loop" ):
266+ worker ._Registry ().add_named_orchestrator ("test" , wrapper ) # pyright: ignore[reportPrivateUsage]
222267
223268
224269def _assert_mixed_wrapper_cycle_fails (shape : str ) -> None :
@@ -235,21 +280,16 @@ def __call__(self, ctx: task.OrchestrationContext, value: int) -> int:
235280 wrapper .__wrapped__ = CallableWrapper () # type: ignore[attr-defined]
236281 CallableWrapper .__call__ .__wrapped__ = partial (wrapper ) # type: ignore[attr-defined]
237282
238- # No input: dispatch must detect the cycle without relying on type discovery.
239- history = [
240- helpers .new_orchestrator_started_event (NOW ),
241- helpers .new_execution_started_event ("test" , INSTANCE_ID ),
242- ]
243- complete = _completion (_execute (wrapper , [], history ))
244- assert complete .orchestrationStatus == pb .ORCHESTRATION_STATUS_FAILED
245- assert complete .failureDetails .errorType == "builtins.ValueError"
246- assert "wrapper loop" in complete .failureDetails .errorMessage
283+ registry = worker ._Registry () # pyright: ignore[reportPrivateUsage]
284+ with pytest .raises (ValueError , match = "wrapper loop" ):
285+ registry .add_named_orchestrator ("test" , wrapper )
286+ assert registry .get_orchestrator ("test" ) is None
247287 with pytest .raises (ValueError , match = "wrapper loop" ):
248288 shared .is_async_callable (wrapper )
249289
250290
251291@pytest .mark .parametrize ("shape" , ["partial" , "callable-partial" ])
252- def test_mixed_wrapper_cycle_fails_within_bounded_worker_dispatch (shape : str ) -> None :
292+ def test_mixed_wrapper_cycle_is_rejected_within_bounded_registration (shape : str ) -> None :
253293 result = subprocess .run (
254294 [
255295 sys .executable , "-c" ,
@@ -555,3 +595,12 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> AsyncGenerator
555595 complete = _completion (_execute (orchestrator , [], _start ()))
556596 assert complete .orchestrationStatus == pb .ORCHESTRATION_STATUS_FAILED
557597 assert "Async generator orchestrators are not supported" in complete .failureDetails .errorMessage
598+
599+
600+ @pytest .mark .parametrize (("now" , "offset" ), [
601+ (NOW , 0 ), (NOW .replace (tzinfo = UTC ), 0 ), (NOW .replace (tzinfo = timezone (timedelta (hours = 2 ))), 7200 ),
602+ ])
603+ def test_loop_time_accepts_naive_and_aware_orchestration_clocks (now : datetime , offset : int ) -> None :
604+ ctx = cast (task .OrchestrationContext , SimpleNamespace (current_utc_datetime = now ))
605+ loop = DurableEventLoop (ctx , is_complete = lambda : False )
606+ assert loop .time () == (NOW - datetime (1970 , 1 , 1 )).total_seconds () - offset
0 commit comments