Fix: added re-raising of exceptions to dag.test() execution

[styannik]

---

Fixing the exception handling issue from #34282

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[ketozhang]

This implementation will bring back the behavior that #30965 attempts to fix. That is, no callbacks will be ran.

[ketozhang]

Fixing the callback is out of scope in my opinion. However updating the documentation to address this missing behavior is a good add to this PR.

The users can always add their own calls to the callback after checking the state

dagrun = dag.test()
if dagrun.state == "failed":
    # run callback

[hussein-awala]

Why do you think we need to fail the command and stop the execution when a TI fails? Even in a real execution, we could have a success run with some failed TIs (if we change the trigger rule).

I just checked the main issue and the suggestion, but IMHO if we want to fix this, we need to somehow check the trigger rules instead of randomly fast failing the execution.

[ketozhang]

@hussein-awala Because that was the original behavior of dag.test() and the behavior before Airflow 2.6.2. For motivation behind
dag.test() behavior is discussed in its original PR: #26400.

Redefining the behavior of dag.test() is a bigger scope than what I intend in the issue. I agree to a degree that dag.test() is less useful if it does not cover a lot of the behavior we see in a normal scheduler.

[potiuk]

Yes. I think (and maybe @dimberman as original creator can confirm) the dags.test is not really a replacement for all the scheduling features. I believe the change as it is implemented might undermine a bit the story to be told at the upcoming Airflow Summit talk: https://airflowsummit.org/sessions/2023/to-debug-a-dag-the-airflow-local-dev-story/ so it would be great to make sure that the talk and the implementation are consistent.

@dimberman - what would you say ?

[mis98zb]

Will it be a better way that we check the dr.state at the end of test(), and raise exception in case it equals to DagRunState.FAILED?

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[potiuk]

Are you still working on it @styannik ? @dimberman - I think it waits for your input.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[dimberman]

@potiuk @hussein-awala @styannik apologies for being out of the loop here. Been a bit swamped on my end :).

I think that raising an exception here is the wrong way to go about this.

I think there should be a slightly larger conversation about what we do/return when we run dag.test.

One thought I've been having recently is that we should have dag.test return a dictionary of tasks -> status.

So let's say we have a task that fails with an exception, maybe we could have a dictionary read

{ 
    "task_a": { "status": "success"},
    "task_b": {"status": "failed", "exc_info": DivideByZeroException(...)},
    "dag_status": "failed",
}

This way people can run tests against their DAGs with different inputs and see what failure/acceptance scenarios have, and the tests will correctly fail as you can take the result and run assertions. WDYT?

[dimberman]

/

[ketozhang]

@dimberman That idea of generally allowing users to define their own failures sounds great to me.
Returning a dictionary for test status would be great if it isn't easy to get the status and exception from the dag.

[potiuk]

@potiuk @hussein-awala @styannik apologies for being out of the loop here. Been a bit swamped on my end :).

Welcome back :).

I think there should be a slightly larger conversation about what we do/return when we run dag.test.

Yep. I think we should document design decisions for it and provide some best practices on what is expected of this command and how to use it. The current description https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/executor/debug.html#testing-dags-with-dag-test is rather rudimentary and does not explain what happens and what to expect

I think there are two things that maybe need clarification/explanation that I saw people asked (and maybe making decisions):

• what happens with DB entries - dag run, task instances, xcoms? Are there left overs after running dags test? Should they be cleared (autoamtically with a flag ? manually?)
• what is the return value and exit codse (if used in CLI) f the test method (that's what you asked for. Yeah - dictionary is good for test() python method, but the airflow dags test CLI (that uses dag.test() method intrnally and it should derive exit code from the state of dag run, and possibly have a way to save the dictionary as output somewhere (or print at stdout). I think the last one is important for automation in CI.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.