@@ -224,3 +224,44 @@ async def parent_flow():
224224 # Both parent and nested should see the deployment parameters
225225 assert seen_params ["parent" ] == {"foo" : "from_deployment" }
226226 assert seen_params ["nested" ] == {"foo" : "from_deployment" }
227+
228+ async def test_parameters_accessible_in_deeply_nested_flows (
229+ self , deployment_id , prefect_client
230+ ):
231+ """Test that deployment.parameters propagates through multiple nesting levels."""
232+ from prefect .flow_engine import run_flow_async
233+ from prefect .runtime import deployment as deployment_runtime
234+
235+ # Track what we saw at each level
236+ seen_params = {}
237+
238+ @flow
239+ async def doubly_nested_flow ():
240+ seen_params ["doubly_nested" ] = deployment_runtime .parameters
241+ return deployment_runtime .parameters
242+
243+ @flow
244+ async def nested_flow ():
245+ seen_params ["nested" ] = deployment_runtime .parameters
246+ result = await doubly_nested_flow ()
247+ return result
248+
249+ @flow
250+ async def parent_flow ():
251+ seen_params ["parent" ] = deployment_runtime .parameters
252+ result = await nested_flow ()
253+ return result
254+
255+ # Create flow run from deployment with custom parameters
256+ flow_run = await prefect_client .create_flow_run_from_deployment (
257+ deployment_id , parameters = {"foo" : "from_deployment" , "level" : "deep" }
258+ )
259+
260+ # Execute the flow using the flow engine helper
261+ await run_flow_async (parent_flow , flow_run = flow_run )
262+
263+ # All levels should see the deployment parameters
264+ expected = {"foo" : "from_deployment" , "level" : "deep" }
265+ assert seen_params ["parent" ] == expected
266+ assert seen_params ["nested" ] == expected
267+ assert seen_params ["doubly_nested" ] == expected
0 commit comments