GitHub user skiedude closed the discussion with a comment: What is the correct way to unit test the PythonBranchOperator
Guess this was a case of the rubber duck. I had read over https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#unit-tests previously, and tried the suggestion of testing a custom operator. However I realized I tried `task.run()` and not `ti.run()` The following changes got the test working again all the way up to 2.10.4 ``` @@ -27,10 +28,10 @@ def create_dag_task_context(): state=DagRunState.RUNNING, run_type=DagRunType.MANUAL ) - task = dag.get_task(task_id) - taskinst = dagrun.get_task_instance(task_id) - taskinst.task = task - context = taskinst.get_template_context() - return task, context - return _create - + ti = dagrun.get_task_instance(task_id) + ti.task = dag.get_task(task_id) + ti.task.trigger_rule = TriggerRule.ALWAYS + ti.task.on_success_callback = None + context = ti.get_template_context() + return ti, context + return _create ``` And in the test ``` - task, context = create_dag_task_context('dag_name', task_id, dagbag) - task.execute(context) - xcom_return = task.xcom_pull(task_ids=task_id, key='skipmixin_key', context=context) + ti, context = create_dag_task_context('dag_name', task_id, dagbag) + ti.run(ignore_ti_state=True) + xcom_return = ti.task.xcom_pull(task_ids=task_id, key='skipmixin_key', context=context) ``` GitHub link: https://github.com/apache/airflow/discussions/45565#discussioncomment-11803009 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected]
