Run system tests using dag.test()

[vincbeck]

Resolves #31826
Related and continuation of #36236

Deferrable operators cannot be run by DebugExecutor. DebugExecutor is the default executor when running system tests. The purpose of this PR is to no longer use DebugExecutor for system tests but dag.test(). See https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/debug.html#testing-dags-with-dag-test

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[syedahsn]

I'm not sure what the solution is, but I believe the reason the test does not pass is because pytest, which is running the test, exits out once it receives an Exception, which is what happens when a task is deferred. We know that when a Task is Deferred, it raises the TaskDeferred Exception. We would need to catch that Exception, and process it differently.

Here's an example that might be useful: #37542

[potiuk]

It's because System Tests use debug executor that has no triggerer/deferred support https://github.com/apache/airflow/blob/main/tests/system/conftest.py#L29

Proper way would likely be to change System Tests to use airflow dag test under the hood which was replacement for Debug Executor: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/debug.html#testing-dags-with-dag-test

That would even enable to remove DebugExecutor because this is the last thing it is used for.

[potiuk]

I think dag test had recently (2.8?) got the suppor to run deferred tasks

[potiuk]

This is how system tests are run currently:

https://github.com/apache/airflow/blob/main/tests/system/utils/__init__.py#L34

[vincbeck]

I see! Stupid question: why dont we use the SequentialExecutor for system tests instead of DebugExecutor. The change would be even easier no?

EDIT: nevermind, your solution is even easier. Looking good.

[vincbeck]

The problem I have now (I updated the code) is the DAG always succeed, even though when it should fail. I guess the test.debug() swallow any exception which makes the DAG always succed?

[potiuk]

The problem I have now (I updated the code) is the DAG always succeed, even though when it should fail. I guess the test.debug() swallow any exception which makes the DAG always succed?

The dag.test() returns DagRun and that dag run state should reflect whether the dag run failed or not - but I think it should be tested in the context of System Tests - so I guess switching to use deferrables might be a good opportunity.

[vincbeck]

Thanks @potiuk for the information! That helps. I have an implementation working, could you take a look?

[vincbeck]

I ran all AWS system tests using this and they all succeeded. I also tested the negative case by "breaking" some system tests. All seems fine

[potiuk]

Looks good then.

[vincbeck]

Question on this one though. From the doc it says, dag.test() does not use an executor at all. So it is not possible, somehow, to configure to use a specific executor. I had in mind to possibly run some system tests with different executor (e.g. ECSExecutor). How hard would that be?

[potiuk]

It's a completely different mechanism - it just runs _run_raw_task for all tasks according to dependencies - look at the code.

If you run it with a different executor, what you really need you need to run Airflow Scheduler with the executor and you need to trigger the DAG (via APi or CLI). You should not really use Pytest for that. Previously backfill was used to run the dags (this is how Debug Executor worked) - but that was a feature of Debug Executor - if you have ECS executor, then having scheduler to run the DAG is probably much better.

[vincbeck]

Just for testing, I tried to run a system test with the backfill approach with AwsEcsExecutor and it seems to work (besides the issue I mentioned here). At least an ECS task is triggered and tries to run the system test. Do you still do not recommend this approach @potiuk?

        dag.clear(dag_run_state=State.QUEUED)
        dag.run(
            local=False,
            executor=AwsEcsExecutor(),
        )

Edit: nevermind, it does not work. I'll go with the scheduler option

[potiuk]

Edit: nevermind, it does not work. I'll go with the scheduler option

Thought so :)