Testing tasks from CLI when they need dependencies from other tasks

UPDATE: The title of the issue was updated (and type was changed to a feature) to reflect the discussion that we need a feature where we can supply dependencies from other tasks to those run via task test CLI command.

Apache Airflow version

2.5.0

What happened

Let's have following DAG with task echo:

@dag(start_date=pendulum.now())
def playground():

    @task
    def echo(message: str):
        logging.info(f'Received message: "{message}"')
        return message

    echo("hello world")

playground()

I would like to test the task from CLI. So according to the help and documentation I've created JSON with required parameter for the function and then I ran the following command:

$ airflow tasks test playground echo -t '{"message": "this is airflow"}'

In the output I'll get the following lines:

[2022-12-10 21:15:19,874] {playground.py:10} INFO - Received message: "hello world"
[2022-12-10 21:15:19,874] {python.py:177} INFO - Done. Returned value was: hello world

In my opinion, this is wrong behavior.

What you think should happen instead

If I provide the content of message parameter for the task, it should be used. So instead of message "hello world" the message "this is airflow" should be displayed and returned. Also it looks like, the task was called with the parameter from inside of the DAG.

If I change the task this way:

@task
def echo(message: str, *args, **kwargs):
   print(kwargs['params'])
   logging.info(f'Received message: "{message}"')
   return message

I can see in the output, that the parameter message from the test was really passed to the task with the proper value:

{'message': 'this is airflow'}
[2022-12-10 21:26:10,495] {playground.py:11} INFO - Received message: "hello world"
[2022-12-10 21:26:10,495] {python.py:177} INFO - Done. Returned value was: hello world

How to reproduce

1. Write simple task with TaskFlow API, such this:

   import logging
   from airflow.decorators import task, dag
   import pendulum
   
   
   @dag(start_date=pendulum.now())
   def playground():
       @task
       def echo(message: str, *args, **kwargs):
           print(kwargs['params'])
           logging.info(f'Received message: "{message}"')
           return message
      
       echo("hello world")
   
   
   playground()
   

2. Test the task from CLI:

   $ airflow tasks test playground echo -t '{"message": "this is airflow"}'
   

Operating System

Fedora Linux 37

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

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