amoghrajesh commented on PR #46176:
URL: https://github.com/apache/airflow/pull/46176#issuecomment-2626244060
Very interesting pattern really:
```
def test_dag_param_resolves_from_task(self, create_runtime_ti,
mock_supervisor_comms, time_machine):
"""Test dagparam resolves on operator execution"""
instant = timezone.datetime(2024, 12, 3, 10, 0)
time_machine.move_to(instant, tick=False)
dag = DAG(dag_id="dag_with_dag_params",
start_date=timezone.datetime(2024, 12, 3))
dag.param("value", default="NOTSET")
class CustomOperator(BaseOperator):
def execute(self, context):
# important to use self.dag here
assert self.dag.params["value"] == "NOTSET"
task = CustomOperator(task_id="task_with_dag_params")
runtime_ti = create_runtime_ti(task=task,
dag_id="dag_with_dag_params")
run(runtime_ti, log=mock.MagicMock())
mock_supervisor_comms.send_request.assert_called_once_with(
msg=SucceedTask(
state=TerminalTIState.SUCCESS, end_date=instant,
task_outlets=[], outlet_events=[]
),
log=mock.ANY,
)
```
At runtime, we parse the `ti` which will trigger reparsing our dag as per
the fixtures and will add `params` to it if needed. At runtime,
`self.dag.params` should have the required info for a task :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
