Dags test doesn't clear database afert runs

[michalslowikowski00]

Apache Airflow version: dev
Environment: breeze and python 3.7.9

What happened:
When I run modified DAG, then task that are commented are trying to be executed.
This exception has beed raised airflow.exceptions.TaskNotFound: Task get_jobs_for_job_group not found

What you expected to happen:
Dags should execute only actual code

Database has not been cleared between runs

How to reproduce it:

1. Run example dag with more than one task in breeze, i.e. airflow dags test example_dataprep 2020-08-24
2. Leave only one task in previously run DAG, comment out rest of the task
3. Run example dag with only one task in breeze with the same data as you run previously, airflow dags test example_dataprep 2020-08-24
4. Observe if airflow.exceptions.TaskNotFound is raised and is printed in logs

Attachment:

root@7389db9e9f99:/opt/airflow# airflow dags test example_dataprep 2020-08-24
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 33, in <module>
    sys.exit(load_entry_point('apache-airflow', 'console_scripts', 'airflow')())
  File "/opt/airflow/airflow/__main__.py", line 40, in main
    args.func(args)
  File "/opt/airflow/airflow/cli/cli_parser.py", line 53, in command
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/utils/session.py", line 65, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/utils/cli.py", line 86, in wrapper
    return f(*args, **kwargs)
  File "/opt/airflow/airflow/cli/commands/dag_command.py", line 388, in dag_test
    dag.run(executor=DebugExecutor(), start_date=args.execution_date, end_date=args.execution_date)
  File "/opt/airflow/airflow/models/dag.py", line 1454, in run
    job.run()
  File "/opt/airflow/airflow/jobs/base_job.py", line 233, in run
    self._execute()
  File "/opt/airflow/airflow/utils/session.py", line 65, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/jobs/backfill_job.py", line 805, in _execute
    session=session)
  File "/opt/airflow/airflow/utils/session.py", line 61, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/jobs/backfill_job.py", line 733, in _execute_for_run_dates
    session=session)
  File "/opt/airflow/airflow/utils/session.py", line 61, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/jobs/backfill_job.py", line 629, in _process_backfill_task_instances
    run.update_state(session=session)
  File "/opt/airflow/airflow/utils/session.py", line 61, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/models/dagrun.py", line 308, in update_state
    ti.task = dag.get_task(ti.task_id)
  File "/opt/airflow/airflow/models/dag.py", line 1286, in get_task
    raise TaskNotFound("Task {task_id} not found".format(task_id=task_id))
airflow.exceptions.TaskNotFound: Task get_jobs_for_job_group not found

[michalslowikowski00]

@turbaszek Do you know smth about that? :)

[eladkal]

Is this still an issue? the command was significantly refactored in #26400
cc @dimberman

[pankajastro]

tested it look like this has been fixed

[potiuk]

Closing then. We can re-open if needed.