New design of system tests

[mnojek]

Hello everyone,

I am really excited to show you the collective work of the last few weeks!
This is the initial PR that prepares Airflow to be compliant with
AIP-47 which is related to the new design of system tests in Airflow. This is the first step to automated system tests in our CI. When it is merged, it will enable other providers to start migration of their system tests to the new design.

Basically, the main advantage of this design is that we will have self-contained tests encapsulated in DAGs that can be run with pytest as well as just imported into Airflow and be executed there. Requirements for the environment are kept at minimum and are mostly related to specific test needs. By using built-in features of Airflow, each system test is a DAG that is also an example of usage for tested Operator. All in one, in a simple file :)

It is a minimal PR that contains a bunch of tests for only 1 Google service (BigQuery) and everything else that is required by the tests to be working properly. It also consists of all necessary documentation, new pre-commit hook (check-watcher-in-examples), updated checks (in tests/always) and some other things. More tests will be migrated in a separate PRs, because we wanted this one to be as small as possible.

Please familiarize with the changes and also read the AIP-47 to fully understand why they are needed.
We really want to hear your feedback. In case of any questions, concerns or comments - we are here to answer them and explain things. The whole work was overseen by @potiuk and done together with @bhirsz and @kosteev and me (@mnojek).

[ferruzzi]

I know this was discussed in the email thread, but I'll raise the concern one last time because this example really drives it home, then I promise I'll drop it. We are adding the exact same line of code to every Example DAG (now System Test), shouldn't that really be handled behind the scenes when the test DAG is being parsed instead of adding this copy/pasted line to literally hundreds of files?

[mnojek]

I don't have a better idea how this can be achieved without explicitly defining this task dependency. This is actually quite clever because in the previous version of this design we were listing all the tasks again and then defining >> operator with the watcher. Now it's just this one line that has a reference to all tasks in a local dag and then appends the watcher to all of them. And it is of course only needed when there is a teardown task in the dag.
But of course I am open to any specific suggestions that may improve the design.

[potiuk]

We thought about a special "util" function for that.

I do not think we discussed about "magical" adding it. this is a bit of a hassle, I agreee but adding it would likely require something on the DAG level - for example a special parameter or decorator. And the problem with that is that it would have to be added somewhere at the beginning of the DAG - where you define the DAG most likely. And the problem with that is that you also use the DAG as examples in our documentation. And we extracts parts of the examples into the documentaiton and we should not pollute those examples with things that are not really good "examples" on how you should add your DAGs. - and those example usually show the DAG definition/default_args as part of the example - by having the special decorator, or parameter on the DAG to indicate that DAG shoudl have "watcher" added migth be too easily copied from those extracted examples.

Having explicit watcher makes it so much easier - it is at the and and it is explicit (which means there is no magic)

But actually what made me think now - we should actually make it even more separated an explicit.

@mnojek why don't we change the the watcher's local import and some comment there to make it even more separated and "explicit" for example:

         >> delete_bucket
    ) 
    
    from tests.system.utils.watcher import watcher
    # This test run as a system test needs watcher in order to mark success/failure
    # where "tearDown" task is part of the DAG
    list(dag.tasks) >> watcher()
    

[ferruzzi]

I am open to any specific suggestions

I get that. Unfortunately I only have half-baked suggestions. the only thing I can really think of would be if we had something like tests.system.DAG which inherits airflow.DAG and adds that watcher to the task list, but that's only a half-baked idea and I haven't really thought it through all the way. Writing the same line in hundreds of files just feels wrong, but without a more solid alternative, I'll stop beating that dead horse. Thanks for the effort you've put into this, and I hope I'm not coming off as argumentative or anything.

[bhirsz]

Writing the same line in hundreds of files just feels wrong

I fully agree with you and it was actually the reason for the redesign - where we had dozens of lines repeated for every test only to trigger other file with the DAG. And we always welcome any suggestions and discussion. We changed the design several times over the time thanks for the input from the others.
I agree with Jarek here - the goal is to have an example dag that can be used by other Airflow users as reference and it would be best to avoid any code specific only to system test. Proposed local import of the watcher looks like a good idea for me for this reason alone

[mnojek]

I support the idea about moving the import for the watcher to local scope. It will be more obvious why it is needed, and we can also add the comment that @potiuk suggested (or similar).
And yeah, copy-pasting the same piece of code in many places is not the best, but at least here this "piece" is small and it's better to have it explicitly than doing some magic stuff that will only shadow the real implementation.
I will do the changes next week.

[mnojek]

Updated the code and moved the watcher import closer its usage.
Also added the comment to demystify this piece of code.
Please check and confirm if it's fine now 😃

[mik-laj]

We can also configure a cluster policy that will perform the necessary steps only in tests.
Here are docs about cluster policy:
https://airflow.apache.org/docs/apache-airflow/stable/concepts/cluster-policies.html?highlight=cluster%20policy#dag-policies

[bhirsz]

Cluster policy looks nice but it's overkill in our case when after discussion in this PR we managed to store the watcher together in the test method. It's worth to remember though in case we will need any extra steps.

[o-nikolas]

I see that below this is being copy/pasted into each example dag file. What if this function needs to be modified or fixed in the future? Would that involve patching every single example dag file?

[bhirsz]

Unfortunately yes. Although in this case it isn't complicated to replace few lines - one file or 100 it doesn't matter for your awk/sed script ;). It's bit of the trade off - we could also leave example dags without any pytest method and generate tests dynamically but by adding this method it's easier to run just one system tests.

[o-nikolas]

one file or 100 it doesn't matter for your awk/sed script

Hmm, this is not the kind of precedent I want to be setting 😄 It's still quite finicky and error prone.

You can clean this up quite a bit using python itself. The test_run method can be defined in one location in a separate module and then just imported into each example dag file.

I've mocked up an example, see the attached image below:

[potiuk]

That's a nice idea. If pytest still discovers it this way, I lke it a lot better. also you can even connect it with local import to make it even more localized (same reason as in the other comment , where we care about "example_dag" being still an example dag despite also being a self-contained test. Also it nicely solves the problem which I thought about - what happens if we have two dags in an example. They would have to have slightiy different method ( dag1.clear() dag1.run() vs dag2.clear(() dag2.run()). Passing dag as parameter solves it nicely.

I'd be for something like this:

# Needed to run the example dag as system test <link to the docs>
from tests.airflow.proivders.util import get_test_run
test_run = get_test_run(dag)

[bhirsz]

I like the idea - it's also making it easier for temporary updates of the test method (ie adding verbose mode, overwriting executor etc). I've tested it and it appears to be working. Pytest recognizes it if I run it using pytest command however it's not detected by IDE as test method (but maybe I can workaround it somehow with pytest ini). I will discuss it with @mnojek and if we both agree we can update it

[o-nikolas]

Also it nicely solves the problem which I thought about - what happens if we have two dags in an example.

Yupp, good catch! I meant to add this to my original post but totally forgot. It makes it much more configurable this way.

[bhirsz]

Updated the code with the test method proposed by @o-nikolas (just added # noqa: E402 since we're importing not on the top of the file)

[mnojek]

Updated the check with a regex pattern, so that it accepts whether there is a comment in between lines in "pytest function" or not.

[kosteev]

Thank you very much for this effort!

[potiuk]

Fixed Python 3.6 (needed for 2.2.5 release) in main. I hope it will succeed this time :)

[potiuk]

Echch

[potiuk]

Rebuilding (aftermath of preparing 2.2.5 release)

[ferruzzi]

Nice. Congrats on the merge.

[Bowrna]

If every task will pass, then watcher will be skipped. Does it mean the teardown task(e.g. to clean up the resources) will be skipped? @mnojek

[mnojek]

No, the teardown task will be executed because by design it has a trigger rule set to "all done" which means it is executed always when all parent tasks are executed. The watcher is a child for every task so it doesn't count into the tasks that need to be executed to trigger the teardown. It is skipped only if everything passed.
And the watcher is triggered only if any of its parent tasks failed, and because all tasks are parent for the watcher, if any of them fails, the watcher is triggered.
Hope it answers your question. If you still have concerns, please comment and I will try to help.

[potiuk]

Summary of ALL performed actions: for 41 issues
Re-added file number: 0
Completed file number: 4
Done 43/272 = 15.81%