Airflow tasks test is broken for xcom_pull based tasks

[mmuru]

Apache Airflow version

2.3.2 (latest released)

What happened

Testing example_xcom_args via CLI throws following exception.

airflow.exceptions.AirflowException: XComArg result from generate_value at example_xcom_args with key="return_value" is not found!

full exception stack:
[2022-06-05 08:46:48,081] {taskinstance.py:1889} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/venv-af/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1451, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1555, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 2212, in render_templates
    rendered_task = self.task.render_template_fields(context)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/baseoperator.py", line 1185, in render_template_fields
    self._do_render_template_fields(self, self.template_fields, context, jinja_env, set())
  File "/venv-af/lib/python3.9/site-packages/airflow/utils/session.py", line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/abstractoperator.py", line 344, in _do_render_template_fields
    rendered_content = self.render_template(
  File "/venv-af/lib/python3.9/site-packages/airflow/models/abstractoperator.py", line 398, in render_template
    return tuple(self.render_template(element, context, jinja_env) for element in value)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/abstractoperator.py", line 398, in <genexpr>
    return tuple(self.render_template(element, context, jinja_env) for element in value)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/abstractoperator.py", line 394, in render_template
    return value.resolve(context)
  File "/venv-af/lib/python3.9/site-packages/airflow/utils/session.py", line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File "/venv-af/lib/python3.9/site-packages/airflow/models/xcom_arg.py", line 152, in resolve
    raise AirflowException(
airflow.exceptions.AirflowException: XComArg result from generate_value at example_xcom_args with key="return_value" is not found!

What you think should happen instead

It should have return the results without any exception.

However, test dags works without exceptions.

airflow dags test example_xcom_args 2022-06-05

It seems that can't test individual tasks via CLI if the task has xcom. I think, since airflow 2.3.x, xcom stores the task results in-memory instead of DB, it couldn't retrieve previous task result from xcom.

How to reproduce

Run tasks from airflow example_dags/example_xcomargs.py or example_xcom.py

airflow tasks test example_xcom_args generate_value 2022-06-05
airflow tasks test example_xcom_args print_value 2022-06-05

Operating System

Mac OS and Python 3.9.12

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else

I believe it is a regression since airflow 2.3.x release xcom changes.

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

[ephraimbuddy]

So, this airflow tasks test example_xcom_args print_value 2022-06-05 is pulling xcom pushed by generate_value task. Maybe we should have a test_mode for XComArg.resolve and return something(string) when test_mode.
WDYT @uranusjr

[TsubasaBE]

Hi everybody,

Yeah... I've noticed the same since our upgrade to 2.3.2.
It used to be that if we run tasks using the 'airflow tasks test' command, Airflow stored the xcom keys in the metadatabase.
eg.

This isn't the case anymore and makes testing and debugging projects a lot more cumbersome to do.

Hopefully this can be resolved in a future upgrade as we rely a lot on this.

Thank you.

[uranusjr]

Yeah just returning an arbitrary value is probably is the best we can do. Since anything is equally arbitrary, None is also a possible value. I think @ashb mentioned a similar case yesterday in another issue where we can use None.

[github-actions]

This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.

[mmuru]

Team: I tested it again with latest Airflow version 2.6.2 and this issue is still exists. Can't use Airflow task test to test any xcom_pull based jobs anymore. It's been open for so long, no one really bother to fix it.

[potiuk]

Suggestion @mmuru : Maybe a good time to attempt to fix it simply? Airflow has 2500 contributors who gave back to free software they use by fixing issues, documentation. And anyone can attempt to fix such an issue - anyone can submit a PR that can be reviewed and merged. This is really how open source works - if somethign bothers you, you roll your sleeves up, figure out how to fix it, submit a PR, and get it merged.

Simply the fact of life is that no-one fixed it yet, because no-one found it important enough to be fixed. Likely few people have similar issue. We have now dags test command as well in new versions, maybe this is a better way of doing what you are trying to do ? Or if not maybe you could take a look at fixing it, or somoene at your company would like to take a look and maybe fix it? What I can do as a maintainer, I might mark it as "good first issue" and hope that somoene picks it up, but really the best way to speed up implementing things is to ... implement it.

[uranusjr]

#29608 was on the right track but the author ran out of steam unfortunately. It’s probably easiest to build on that existing implementation.

[jccalvojackson]

how one would go about running a single task that depends on an xcom, even if not using airflow tasks test?
to use airflow tasks run it needs to be associated to a dagrun, no?