airflow tasks test cli not seeing xcom result

[vinny-santaiti]

Apache Airflow version

2.3.3 (latest released)

What happened

The airflow tasks test CLI doesn't seem to work for xcoms, it used to work in version 2.2.5. It works fine in the UI.
Example: airflow tasks test tutorial print_date 2015-06-01
kwargs['ti'].xcom_pull(task_ids='push_data')
https://airflow.apache.org/docs/apache-airflow/stable/tutorial.html#id2
When you push data in a task, then pull xcom data by task_ids in a downstream task, it doesnt work.

What you think should happen instead

The same cli cmd did work fine in airflow 2.2.5

How to reproduce

import pendulum

from airflow import DAG
from airflow.operators.python import PythonOperator


with DAG(
    'example_xcom',
    schedule_interval=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
) as dag:

    def push_data(**kwargs):
        return 'data'

    def pull_data(**kwargs):
        return kwargs['ti'].xcom_pull(task_ids='push_data')

    push_task = PythonOperator(
        task_id='push_data',
        python_callable=push_data
    )
    pull_task = PythonOperator(
        task_id='pull_data',
        python_callable=pull_data
    )
    push_task >> pull_task

airflow tasks test example_xcom push_data 2022-06-01
[2022-07-12 20:22:53,288] {python.py:173} INFO - Done. Returned value was: data
airflow tasks test example_xcom pull_data 2022-06-01
[2022-07-12 20:23:05,402] {python.py:173} INFO - Done. Returned value was: None

Operating System

docker

Versions of Apache Airflow Providers

apache-airflow==2.3.3
apache-airflow-providers-amazon==4.0.0
apache-airflow-providers-celery==3.0.0
apache-airflow-providers-cncf-kubernetes==4.1.0
apache-airflow-providers-docker==3.0.0
apache-airflow-providers-elasticsearch==4.0.0
apache-airflow-providers-ftp==3.0.0
apache-airflow-providers-google==8.1.0
apache-airflow-providers-grpc==3.0.0
apache-airflow-providers-hashicorp==3.0.0
apache-airflow-providers-http==3.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-microsoft-azure==4.0.0
apache-airflow-providers-mysql==3.0.0
apache-airflow-providers-odbc==3.0.0
apache-airflow-providers-postgres==5.0.0
apache-airflow-providers-redis==3.0.0
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-sftp==3.0.0
apache-airflow-providers-slack==5.0.0
apache-airflow-providers-sqlite==3.0.0
apache-airflow-providers-ssh==3.0.0

Deployment

Docker-Compose

Deployment details

apache/airflow:2.3.3-python3.8
https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

Ths is not a bug. Simply previous behaviour was an unintended side effect.

via https://airflow.apache.org/docs/apache-airflow/stable/tutorial.html#id2

Note that the airflow tasks test command runs task instances locally, outputs their log to stdout (on screen), does not bother with dependencies, and does not communicate state (running, success, failed, …) to the database. It simply allows testing a single task instance.

While "Xcom" is not specified here, leaving an XCom behing from test command is something that falls into that .

In 2.3. Foreign Key was added between XCom and Task Instance - #22334 (it was missing before). The result of missing foreign key was that the XCom created during the 'test" command was never cleaned.

This piece of code 71c980a#diff-ad618185a072910e49c11770954af009d1cc070b120a4fde5f2fc095a588271bR550 removes the dag run (and thus transitively task instance and xcom) created during the tests)

I believe (@uranusjr ? Can you confirm please) this was fully intentional change.

Maybe this is a good case however to add the version of airflow tasks test (aiflow tasks tests --no-cleanup ?) command that actually leaves some traces behind. WDYT @uranusjr ?

[uranusjr]

It’s less unintended, more that we considered this as a bug and not actually relied on by users :(

It’s viable to add a flag for this, although it should communicated quite clearly that this will leave junk in the database and you will need to be fine with that.

[potiuk]

So as I thought. Closing as invalid. @vinny-santaiti - if you find it useful to have similar feature with leaving data after tests - feel free to make PR to add such feature. Generally speaking "tasks test" is only intended to run each task in isolation and not leave traces.

[andilabs]

That is pretty sad. We used to relay in our test suit on combo of airflow tasks test and XCom.set and XCom.get_one. After upgrade, we have to refactor a lot. BTW any chance to combine now XCom.set and relay on it in task executes self.xcom_pull meanwhile using airflow tasks test for testing?

[potiuk]

That is pretty sad. We used to relay in our test suit on combo of airflow tasks test and XCom.set and XCom.get_one. After upgrade, we have to refactor a lot. BTW any chance to combine now XCom.set and relay on it in task executes self.xcom_pull meanwhile using airflow tasks test for testing?

Sure. PRs are always welcome :)