Port `dag.test` to Task SDK #50300

kaxil · 2025-05-07T12:14:34Z

DAGs used to test

Filename: example_simplest_dag.py

from airflow.sdk import Variable, dag, task

@dag
def example_simplest_dag():
    @task
    def my_task():
        print("hellooo")
        assert Variable.get("my_variabe") == "my_value"

    my_task()


d = example_simplest_dag()

if __name__ == "__main__":
    d.test()

Result:

Async DAG:

Filename: example_time_delta_sensor_async.py

import datetime
import pendulum

from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.sensors.time_delta import TimeDeltaSensorAsync
from airflow.sdk import DAG

with DAG(
    dag_id="example_time_delta_sensor_async",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
) as dag:
    wait = TimeDeltaSensorAsync(task_id="wait", delta=datetime.timedelta(seconds=5))
    finish = EmptyOperator(task_id="finish")
    wait >> finish

if __name__ == "__main__":
    dag.test()

Result:

Testing

I will add some more tests, but the existing tests (that use dag.test) are sufficient and catch most of the issues. Longer-term we should have Task SDK Integration/system tests that allow accessing DB. The current Task SDK unit test does not allow any DB interaction so tests can't go there.

TODO

Support use_executor on dag.test. It is used in system tests (based on Enable task sdk for system tests #48699)
Handle DAG level callbacks

Handle fail_fast, which currently relies on the DAG being present in the serialized_dag table.
Since we skip task_runner.parse in dag.test(), the DAG is not auto-serialized.

airflow/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

Lines 374 to 378 in 952a004

    
           ser_dag = request.app.state.dag_bag.get_dag(dag_id) 
        
           if ser_dag and getattr(ser_dag, "fail_fast", False): 
        
               task_dict = getattr(ser_dag, "task_dict") 
        
               task_teardown_map = {k: v.is_teardown for k, v in task_dict.items()} 
        
               _stop_remaining_tasks(task_instance=ti, task_teardown_map=task_teardown_map, session=session)

Possible fix to add the following in dag.test():

 from airflow.models.serialized_dag import SerializedDagModel as SDM
 SDM.write_dag(dag=self, bundle_name="test")

Future Enhancements

Replace ti.run & TI._run_raw_task implementation to use this code so we can nuke a lot of code in models.TaskInstance to completely remove a separate execution pathlogs, API server logs etc
Better logging format to differentiate between Task
Not have to rely on DB access entirely

I will do (1) in the next few weeks as that is a big change too (affecting all providers) and (1) and (2) will be deferred for later.

airflow-core/src/airflow/api_fastapi/execution_api/app.py

amoghrajesh

Nice, no major concerns, a few qns here and there

airflow-core/src/airflow/models/dag.py

airflow-core/src/airflow/models/dagrun.py

task-sdk/src/airflow/sdk/definitions/dag.py

task-sdk/src/airflow/sdk/execution_time/supervisor.py

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

amoghrajesh · 2025-05-09T19:41:23Z

Oh i see you are using dag.test: https://github.com/apache/airflow/blob/main/devel-common/src/tests_common/test_utils/system_tests.py#L35-L75

kaxil · 2025-05-09T19:41:23Z

This blew up our system test dashboard, every test is now failing with

INFO httpx:_client.py:1026 HTTP Request: GET http://in-process.invalid./connections/aws_default "HTTP/1.1 500 Internal Server Error"

Are we supposed to be overriding client.base_url somewhere now?

Do you have more logs? Let's chat on slack and we can debug

kaxil · 2025-05-09T20:16:05Z

The "in-process" in the URL does mean that somehow the use_executor flag isn't passed True.

As an example, this is what you should see

dag.test logs:

root@e2e45b18eab2:/opt/airflow# python /files/dags/example_simplest_dag.py
[2025-05-09T20:11:07.373+0000] {taskinstance.py:291} WARNING - No serialized dag found for dag 'xample_simplest_dag'
[2025-05-09T20:11:08.451+0000] {dag.py:2306} INFO - created dagrun <DagRun xample_simplest_dag @ None: manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF, state:running, queued_at: None. run_type: manual>
[2025-05-09T20:11:08.453+0000] {executor_loader.py:269} INFO - Loaded executor: :LocalExecutor:
[2025-05-09T20:11:08.487+0000] {local_executor.py:61} INFO - Worker starting up pid=1345
[2025-05-09T20:11:08.519+0000] {dag.py:1159} WARNING - No tasks to run. unrunnable tasks: {<TaskInstance: xample_simplest_dag.my_task manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF [queued]>}
2025-05-09 20:11:08 [info     ] Secrets backends loaded for worker [supervisor] backend_classes=['EnvironmentVariablesBackend'] count=1
[2025-05-09T20:11:08.952+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://localhost:8080/execution/task-instances/0196b6ac-5123-7413-b995-96d80d1a2c58/run "HTTP/1.1 200 OK"
2025-05-09 20:11:08 [debug    ] Sending                        [supervisor] msg=StartupDetails(ti=TaskInstance(id=UUID('0196b6ac-5123-7413-b995-96d80d1a2c58'), task_id='my_task', dag_id='xample_simplest_dag', run_id='manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF', try_number=1, map_index=-1, pool_slots=1, queue='default', priority_weight=1, executor_config={}, parent_context_carrier={}, context_carrier=None, queued_dttm=None), dag_rel_path='/files/dags/example_simplest_dag.py', bundle_info=BundleInfo(name='dags-folder', version=None), requests_fd=23, start_date=datetime.datetime(2025, 5, 9, 20, 11, 8, 567906, tzinfo=datetime.timezone.utc), ti_context=TIRunContext(dag_run=DagRun(dag_id='xample_simplest_dag', run_id='manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF', logical_date=None, data_interval_start=None, data_interval_end=None, run_after=datetime.datetime(2025, 5, 9, 20, 11, 7, 394963, tzinfo=TzInfo(UTC)), start_date=datetime.datetime(2025, 5, 9, 20, 11, 7, 394963, tzinfo=TzInfo(UTC)), end_date=None, clear_number=0, run_type=<DagRunType.MANUAL: 'manual'>, conf={}, consumed_asset_events=[]), task_reschedule_count=0, max_tries=0, variables=[], connections=[], upstream_map_indexes=None, next_method=None, next_kwargs=None, xcom_keys_to_clear=[], should_retry=False), type='StartupDetails')
[2025-05-09T20:11:09.565+0000] {dag.py:1159} WARNING - No tasks to run. unrunnable tasks: {<TaskInstance: xample_simplest_dag.my_task manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF [running]>}
2025-05-09 20:11:09 [debug    ] Received message from task runner [supervisor] msg=SetRenderedFields(rendered_fields={'templates_dict': None, 'op_args': [], 'op_kwargs': {}}, type='SetRenderedFields')
[2025-05-09T20:11:10.605+0000] {dag.py:1159} WARNING - No tasks to run. unrunnable tasks: {<TaskInstance: xample_simplest_dag.my_task manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF [running]>}
[2025-05-09T20:11:10.714+0000] {_client.py:1026} INFO - HTTP Request: PUT http://localhost:8080/execution/task-instances/0196b6ac-5123-7413-b995-96d80d1a2c58/rtif "HTTP/1.1 201 Created"
2025-05-09 20:11:10 [debug    ] Received message from task runner [supervisor] msg=SucceedTask(state='success', end_date=datetime.datetime(2025, 5, 9, 20, 11, 8, 988171, tzinfo=TzInfo(UTC)), task_outlets=[], outlet_events=[], rendered_map_index=None, type='SucceedTask')
[2025-05-09T20:11:10.727+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://localhost:8080/execution/task-instances/0196b6ac-5123-7413-b995-96d80d1a2c58/state "HTTP/1.1 204 No Content"
2025-05-09 20:11:10 [info     ] Task finished                  [supervisor] duration=2.1683323599863797 exit_code=0 final_state=success
[2025-05-09T20:11:11.611+0000] {scheduler_job_runner.py:810} INFO - Received executor event with state success for task instance TaskInstanceKey(dag_id='xample_simplest_dag', task_id='my_task', run_id='manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF', try_number=1, map_index=-1)
[2025-05-09T20:11:11.644+0000] {scheduler_job_runner.py:852} INFO - TaskInstance Finished: dag_id=xample_simplest_dag, task_id=my_task, run_id=manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF, map_index=-1, run_start_date=2025-05-09 20:11:08.567906+00:00, run_end_date=None, run_duration=None, state=running, executor=LocalExecutor(parallelism=32), executor_state=success, try_number=1, max_tries=0, pool=default_pool, queue=default, priority_weight=1, operator=_PythonDecoratedOperator, queued_dttm=None, scheduled_dttm=2025-05-09 20:11:08.460021+00:00,queued_by_job_id=None, pid=1346
[2025-05-09T20:11:11.673+0000] {dagrun.py:1176} INFO - Marking run <DagRun xample_simplest_dag @ None: manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF, state:running, queued_at: None. run_type: manual> successful
Dag run in success state
Dag run start:2025-05-09 20:11:07.394963+00:00 end:2025-05-09 20:11:11.673977+00:00
[2025-05-09T20:11:11.679+0000] {dagrun.py:1226} INFO - DagRun Finished: dag_id=xample_simplest_dag, logical_date=None, run_id=manual__2025-05-09T20:11:07.394963+00:00_3s9xiRMF, run_start_date=2025-05-09 20:11:07.394963+00:00, run_end_date=2025-05-09 20:11:11.673977+00:00, run_duration=4.279014, state=success, run_type=manual, data_interval_start=None, data_interval_end=None,
[2025-05-09T20:11:11.699+0000] {local_executor.py:223} INFO - Shutting down LocalExecutor; waiting for running tasks to finish.  Signal again if you don't want to wait.

API Server logs:

Logs:

[2025-05-09T20:10:56.842+0000] {api_server_command.py:92} INFO - Running the uvicorn with:
Apps: all
Workers: 4
Host: 0.0.0.0:8080
Timeout: 120
Logfiles: -
=================================================================
INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
INFO:     Started parent process [1264]
INFO:     Started server process [1267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Started server process [1270]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Started server process [1268]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Started server process [1269]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
2025-05-09 20:11:08 [debug    ] Starting task instance run     hostname=e2e45b18eab2 pid=1346 ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58 unixname=root
2025-05-09 20:11:08 [debug    ] Retrieved task instance details dag_id=xample_simplest_dag state=queued task_id=my_task ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
2025-05-09 20:11:08 [info     ] Task started                   hostname=e2e45b18eab2 previous_state=queued ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
2025-05-09 20:11:08 [info     ] Task instance state updated    rows_affected=1 ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
INFO:     127.0.0.1:51544 - "PATCH /execution/task-instances/0196b6ac-5123-7413-b995-96d80d1a2c58/run HTTP/1.1" 200 OK
2025-05-09 20:11:09 [info     ] Updating RenderedTaskInstanceFields field_count=3 ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
2025-05-09 20:11:10 [debug    ] RenderedTaskInstanceFields updated successfully ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
INFO:     127.0.0.1:51544 - "PUT /execution/task-instances/0196b6ac-5123-7413-b995-96d80d1a2c58/rtif HTTP/1.1" 201 Created
2025-05-09 20:11:10 [debug    ] Updating task instance state   new_state=success ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
2025-05-09 20:11:10 [debug    ] Retrieved current task instance state max_tries=0 previous_state=running ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58 try_number=1
2025-05-09 20:11:10 [info     ] Task instance state updated    new_state=success rows_affected=1 ti_id=0196b6ac-5123-7413-b995-96d80d1a2c58
INFO:     127.0.0.1:51544 - "PATCH /execution/task-instances/0196b6ac-5123-7413-b995-96d80d1a2c58/state HTTP/1.1" 204 No Content

Dag used:

from airflow.sdk import Variable, dag, task
import os

@dag
def xample_simplest_dag():
    @task
    def my_task():
        print("hellooo")
        os.environ["AIRFLOW_VAR_MY_VARIABLE"] = "my_value"
        assert Variable.get("my_variable") == "my_value"

    my_task()


d = xample_simplest_dag()

if __name__ == "__main__":
    d.test(use_executor=True)

closes #45549 Key changes: - Moves `dag.test` implementation to Task SDK, leveraging the existing in-process execution infrastructure - Adds `JWTBearerTIPathDep` for proper task instance path validation - Updates `InProcessExecutionAPI` to support task instance validation - Removes legacy `dag.test` implementation from DAG class The changes ensure that `dag.test` uses the same execution path as regular task execution. (cherry picked from commit ef2da6f)

Follow-up of apache#50300 for running a single task.

Follow-up of #50300 for running a single task.

Follow-up of apache#50300 for running a single task.

Follow-up of #50300 for running a single task. (cherry picked from commit 3a9858e)

vincbeck · 2025-05-23T16:02:03Z

I think this PR introduces a bug for deferrable operators. If you look at AWS dashboard, all system tests in deferrable mode are failing. And they are failing since this PR has been merged.

More details

Here is the DAG I am using:

with DAG(
    dag_id=DAG_ID,
    schedule="@once",
    start_date=datetime(2021, 1, 1),
    tags=["example"],
    catchup=False,
) as dag:
    # This context contains the ENV_ID and any env variables requested when the
    # task was built above. Access the info as you would any other TaskFlow task.
    env_id = "722404908466step"
    role_arn = "arn:aws:iam::722404908466:role/RoleSysTest_example_step_functions"

    state_machine_arn = create_state_machine(env_id, role_arn)

    # [START howto_operator_step_function_start_execution]
    start_execution = StepFunctionStartExecutionOperator(
        task_id="start_execution", state_machine_arn=state_machine_arn, deferrable=True
    )
    # [END howto_operator_step_function_start_execution]

    execution_arn = start_execution.output

    # [START howto_sensor_step_function_execution]
    wait_for_execution = StepFunctionExecutionSensor(
        task_id="wait_for_execution", execution_arn=execution_arn
    )
    # [END howto_sensor_step_function_execution]
    wait_for_execution.poke_interval = 1

    chain(
        # TEST SETUP
        state_machine_arn,
        # TEST BODY
        start_execution,
        wait_for_execution,
    )

When I run this DAG with dag.test, execution_arn is None.

I can see this in logs:

[2025-05-23T15:57:34.992+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://in-process.invalid./task-instances/0196fddc-3987-7d7f-9e3b-87db92c63c52/run "HTTP/1.1 200 OK"
2025-05-23 15:57:34 [debug    ] Sending request                [task] msg=GetXCom(key='return_value', dag_id='example_step_functions', run_id='manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR', task_id='start_execution', map_index=None, include_prior_dates=False, type='GetXCom')
2025-05-23 15:57:34 [debug    ] Received message from task runner [task] msg=GetXCom(key='return_value', dag_id='example_step_functions', run_id='manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR', task_id='start_execution', map_index=None, include_prior_dates=False, type='GetXCom')
[2025-05-23T15:57:34.999+0000] {_client.py:1026} INFO - HTTP Request: GET http://in-process.invalid./xcoms/example_step_functions/manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR/start_execution/return_value "HTTP/1.1 404 Not Found"
2025-05-23 15:57:34 [warning  ] Server error                   [airflow.sdk.api.client] detail={'detail': {'reason': 'not_found', 'message': "XCom with key='return_value' map_index=-1 not found for task 'start_execution' in DAG run 'manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR' of 'example_step_functions'"}}
2025-05-23 15:57:34 [error    ] XCom not found                 [airflow.sdk.api.client] dag_id=example_step_functions detail={'detail': {'reason': 'not_found', 'message': "XCom with key='return_value' map_index=-1 not found for task 'start_execution' in DAG run 'manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR' of 'example_step_functions'"}} key=return_value map_index=None run_id=manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR status_code=404 task_id=start_execution

When executed in deferrable mode, the task start_execution does not return any value (None), which then makes the following task fail (because it needs the return value of start_execution to proceed). I looked at several system tests and they are all failing for the same reason.

Any chance someone knows why this happens? I'll look into the code but somebody has something in top of his mind, that will be faster :)

Follow-up of apache#50300 for running a single task.

kaxil · 2025-05-28T22:58:20Z

I think this PR introduces a bug for deferrable operators. If you look at AWS dashboard, all system tests in deferrable mode are failing. And they are failing since this PR has been merged.

More details

Here is the DAG I am using:

with DAG(
    dag_id=DAG_ID,
    schedule="@once",
    start_date=datetime(2021, 1, 1),
    tags=["example"],
    catchup=False,
) as dag:
    # This context contains the ENV_ID and any env variables requested when the
    # task was built above. Access the info as you would any other TaskFlow task.
    env_id = "722404908466step"
    role_arn = "arn:aws:iam::722404908466:role/RoleSysTest_example_step_functions"

    state_machine_arn = create_state_machine(env_id, role_arn)

    # [START howto_operator_step_function_start_execution]
    start_execution = StepFunctionStartExecutionOperator(
        task_id="start_execution", state_machine_arn=state_machine_arn, deferrable=True
    )
    # [END howto_operator_step_function_start_execution]

    execution_arn = start_execution.output

    # [START howto_sensor_step_function_execution]
    wait_for_execution = StepFunctionExecutionSensor(
        task_id="wait_for_execution", execution_arn=execution_arn
    )
    # [END howto_sensor_step_function_execution]
    wait_for_execution.poke_interval = 1

    chain(
        # TEST SETUP
        state_machine_arn,
        # TEST BODY
        start_execution,
        wait_for_execution,
    )

When I run this DAG with dag.test, execution_arn is None.

I can see this in logs:

[2025-05-23T15:57:34.992+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://in-process.invalid./task-instances/0196fddc-3987-7d7f-9e3b-87db92c63c52/run "HTTP/1.1 200 OK"
2025-05-23 15:57:34 [debug    ] Sending request                [task] msg=GetXCom(key='return_value', dag_id='example_step_functions', run_id='manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR', task_id='start_execution', map_index=None, include_prior_dates=False, type='GetXCom')
2025-05-23 15:57:34 [debug    ] Received message from task runner [task] msg=GetXCom(key='return_value', dag_id='example_step_functions', run_id='manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR', task_id='start_execution', map_index=None, include_prior_dates=False, type='GetXCom')
[2025-05-23T15:57:34.999+0000] {_client.py:1026} INFO - HTTP Request: GET http://in-process.invalid./xcoms/example_step_functions/manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR/start_execution/return_value "HTTP/1.1 404 Not Found"
2025-05-23 15:57:34 [warning  ] Server error                   [airflow.sdk.api.client] detail={'detail': {'reason': 'not_found', 'message': "XCom with key='return_value' map_index=-1 not found for task 'start_execution' in DAG run 'manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR' of 'example_step_functions'"}}
2025-05-23 15:57:34 [error    ] XCom not found                 [airflow.sdk.api.client] dag_id=example_step_functions detail={'detail': {'reason': 'not_found', 'message': "XCom with key='return_value' map_index=-1 not found for task 'start_execution' in DAG run 'manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR' of 'example_step_functions'"}} key=return_value map_index=None run_id=manual__2025-05-23T15:56:29.405558+00:00_NkzldSVR status_code=404 task_id=start_execution

When executed in deferrable mode, the task start_execution does not return any value (None), which then makes the following task fail (because it needs the return value of start_execution to proceed). I looked at several system tests and they are all failing for the same reason.

Any chance someone knows why this happens? I'll look into the code but somebody has something in top of his mind, that will be faster :)

Fixed in #51182

closes #45549 Key changes: - Moves `dag.test` implementation to Task SDK, leveraging the existing in-process execution infrastructure - Adds `JWTBearerTIPathDep` for proper task instance path validation - Updates `InProcessExecutionAPI` to support task instance validation - Removes legacy `dag.test` implementation from DAG class The changes ensure that `dag.test` uses the same execution path as regular task execution. (cherry picked from commit ef2da6f)

Follow-up of #50300 for running a single task. (cherry picked from commit 3a9858e)

Follow-up of apache#50300 for running a single task.

Follow-up of apache/airflow#50300 for running a single task. (cherry picked from commit 3a9858e7360129808274de1b8419f7abb7b6b9cb) GitOrigin-RevId: 6fc6c250c916aff8b8c06d4ce8d5232c9483f37e

Follow-up of apache/airflow#50300 for running a single task. GitOrigin-RevId: 3a9858e7360129808274de1b8419f7abb7b6b9cb

boring-cyborg bot added area:API Airflow's REST/HTTP API area:task-sdk labels May 7, 2025

kaxil force-pushed the port-dag-test branch 15 times, most recently from 5f00cac to 2257904 Compare May 8, 2025 18:31

kaxil mentioned this pull request May 8, 2025

Replace DagBag from global app state to with FastAPI dependency #50372

Merged

kaxil force-pushed the port-dag-test branch 3 times, most recently from 0fdd739 to 98c1675 Compare May 9, 2025 08:47

kaxil marked this pull request as ready for review May 9, 2025 08:49

kaxil requested review from XD-DENG, amoghrajesh, ashb, ephraimbuddy and jedcunningham as code owners May 9, 2025 08:49

kaxil added this to the Airflow 3.0.2 milestone May 9, 2025

ashb reviewed May 9, 2025

View reviewed changes

airflow-core/src/airflow/api_fastapi/execution_api/app.py Outdated Show resolved Hide resolved

amoghrajesh reviewed May 9, 2025

View reviewed changes

kaxil mentioned this pull request May 9, 2025

Fix bug with in-process request handling for dag.test #50419

Merged

kaxil added a commit to astronomer/airflow that referenced this pull request May 20, 2025

Port task.test to Task SDK

fa7df47

Follow-up of apache#50300 for running a single task.

kaxil mentioned this pull request May 20, 2025

Port task.test to Task SDK #50827

Merged

kaxil added a commit to astronomer/airflow that referenced this pull request May 20, 2025

Port task.test to Task SDK

46da49d

Follow-up of apache#50300 for running a single task.

kaxil added a commit to astronomer/airflow that referenced this pull request May 20, 2025

Port task.test to Task SDK

80c9c51

Follow-up of apache#50300 for running a single task.

kaxil added a commit to astronomer/airflow that referenced this pull request May 20, 2025

Port task.test to Task SDK

71c2401

Follow-up of apache#50300 for running a single task.

kaxil added a commit to astronomer/airflow that referenced this pull request May 20, 2025

Port task.test to Task SDK

bc2c9be

Follow-up of apache#50300 for running a single task.

kaxil added a commit that referenced this pull request May 20, 2025

Port task.test to Task SDK (#50827)

3a9858e

Follow-up of #50300 for running a single task.

nailo2c pushed a commit to nailo2c/airflow that referenced this pull request May 21, 2025

Port task.test to Task SDK (apache#50827)

747e106

Follow-up of apache#50300 for running a single task.

uranusjr mentioned this pull request May 22, 2025

Slack notifications stopped working after upgrade to 3.0.1 #50754

Closed

2 tasks

kaxil added a commit that referenced this pull request May 22, 2025

Port task.test to Task SDK (#50827)

5cba252

Follow-up of #50300 for running a single task. (cherry picked from commit 3a9858e)

tirkarthi mentioned this pull request May 23, 2025

AttributeError: 'DAG' object has no attribute 'test' #51002

Closed

2 tasks

collinmcnulty mentioned this pull request May 23, 2025

dag.test not working in Airflow 3 #51018

Closed

dadonnelly316 pushed a commit to dadonnelly316/airflow that referenced this pull request May 26, 2025

Port task.test to Task SDK (apache#50827)

c624ba3

Follow-up of apache#50300 for running a single task.

kaxil mentioned this pull request May 28, 2025

Fix deferred task resumption in dag.test() #51182

Merged

collinmcnulty mentioned this pull request Jun 3, 2025

dag.test() does not work if the DAG calls Variable.get or Connection.get #51316

Closed

2 tasks

kaxil mentioned this pull request Jun 3, 2025

Sync v3-0-stable with v3-0-test to release 3.0.2 #51365

Merged

kaxil added a commit that referenced this pull request Jun 3, 2025

Port task.test to Task SDK (#50827)

6fc6c25

Follow-up of #50300 for running a single task. (cherry picked from commit 3a9858e)

kaxil mentioned this pull request Jun 3, 2025

Status of testing of Apache Airflow 3.0.2rc2 #51381

Closed

sanederchik pushed a commit to sanederchik/airflow that referenced this pull request Jun 7, 2025

Port task.test to Task SDK (apache#50827)

0b9840c

Follow-up of apache#50300 for running a single task.

opeida mentioned this pull request Jun 16, 2025

ImportError: cannot import name 'SUPERVISOR_COMMS' with dag.test() #51816

Open

2 tasks

kosteev pushed a commit to GoogleCloudPlatform/composer-airflow that referenced this pull request Oct 23, 2025

Port task.test to Task SDK (#50827)

f5873b8

Follow-up of apache/airflow#50300 for running a single task. GitOrigin-RevId: 3a9858e7360129808274de1b8419f7abb7b6b9cb

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Port `dag.test` to Task SDK #50300

Port `dag.test` to Task SDK #50300

Uh oh!

kaxil commented May 7, 2025 •

edited

Loading

Uh oh!

Uh oh!

amoghrajesh left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amoghrajesh commented May 9, 2025

Uh oh!

kaxil commented May 9, 2025

Uh oh!

kaxil commented May 9, 2025

Uh oh!

vincbeck commented May 23, 2025

Uh oh!

kaxil commented May 28, 2025

More details

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

	ser_dag = request.app.state.dag_bag.get_dag(dag_id)
	if ser_dag and getattr(ser_dag, "fail_fast", False):
	task_dict = getattr(ser_dag, "task_dict")
	task_teardown_map = {k: v.is_teardown for k, v in task_dict.items()}
	_stop_remaining_tasks(task_instance=ti, task_teardown_map=task_teardown_map, session=session)

Port dag.test to Task SDK #50300

Port dag.test to Task SDK #50300

Uh oh!

Conversation

kaxil commented May 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

DAGs used to test

Testing

TODO

Future Enhancements

Uh oh!

Uh oh!

amoghrajesh left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amoghrajesh commented May 9, 2025

Uh oh!

kaxil commented May 9, 2025

Uh oh!

kaxil commented May 9, 2025

Uh oh!

vincbeck commented May 23, 2025

More details

Uh oh!

kaxil commented May 28, 2025

More details

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Port `dag.test` to Task SDK #50300

Port `dag.test` to Task SDK #50300

kaxil commented May 7, 2025 •

edited

Loading