Migrate Alibaba example DAGs to new design #22437

docs/apache-airflow-providers-alibaba/index.rst

@@ -39,7 +39,7 @@ Content
     :maxdepth: 1
     :caption: Resources
 
-    Example DAGs <https://github.com/apache/airflow/tree/main/airflow/providers/alibaba/cloud/example_dags>
+    Example DAGs <https://github.com/apache/airflow/tree/main/tests/system/providers/alibaba>
     PyPI Repository <https://pypi.org/project/apache-airflow-providers-alibaba/>
     Installing from sources <installing-providers-from-sources>
 

docs/apache-airflow-providers-alibaba/operators/oss.rst

@@ -45,7 +45,7 @@ Defining tasks
 
 In the following code we create a new bucket and then delete the bucket.
 
-.. exampleinclude:: /../../airflow/providers/alibaba/cloud/example_dags/example_oss_bucket.py
+.. exampleinclude:: /../../tests/system/providers/alibaba/example_oss_bucket.py
     :language: python
     :start-after: [START howto_operator_oss_bucket]
     :end-before: [END howto_operator_oss_bucket]

airflow/providers/alibaba/cloud/example_dags/example_oss_bucket.py → tests/system/providers/alibaba/example_oss_bucket.py

@@ -17,15 +17,17 @@
 
 # Ignore missing args provided by default_args
 # type: ignore[call-arg]
-
+import os
 from datetime import datetime
 
 from airflow.models.dag import DAG
 from airflow.providers.alibaba.cloud.operators.oss import OSSCreateBucketOperator, OSSDeleteBucketOperator
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "oss_bucket_dag"
 # [START howto_operator_oss_bucket]
 with DAG(
-    dag_id='oss_bucket_dag',
+    dag_id=DAG_ID,
     start_date=datetime(2021, 1, 1),
     default_args={'bucket_name': 'your bucket', 'region': 'your region'},
     max_active_runs=1,
@@ -38,4 +40,15 @@
     delete_bucket = OSSDeleteBucketOperator(task_id='task2')
 
     create_bucket >> delete_bucket
+
+    from tests.system.utils.watcher import watcher
+
+    # This test needs watcher in order to properly mark success/failure
+    # when "tearDown" task with trigger rule is part of the DAG
+    list(dag.tasks) >> watcher()
 # [END howto_operator_oss_bucket]
+
+from tests.system.utils import get_test_run  # noqa: E402
+
+# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
+test_run = get_test_run(dag)

airflow/providers/alibaba/cloud/example_dags/example_oss_object.py → tests/system/providers/alibaba/example_oss_object.py

@@ -16,7 +16,7 @@
 # under the License.
 # Ignore missing args provided by default_args
 # type: ignore[call-arg]
-
+import os
 from datetime import datetime
 
 from airflow.models.dag import DAG
@@ -27,8 +27,10 @@
     OSSUploadObjectOperator,
 )
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "oss_object_dag"
 with DAG(
-    dag_id='oss_object_dag',
+    dag_id=DAG_ID,
     start_date=datetime(2021, 1, 1),
     default_args={'bucket_name': 'your bucket', 'region': 'your region'},
     max_active_runs=1,
@@ -59,3 +61,15 @@
     )
 
     create_object >> download_object >> delete_object >> delete_batch_object
+
+    from tests.system.utils.watcher import watcher
+
+    # This test needs watcher in order to properly mark success/failure
+    # when "tearDown" task with trigger rule is part of the DAG
+    list(dag.tasks) >> watcher()
+
+
+from tests.system.utils import get_test_run  # noqa: E402
+
+# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
+test_run = get_test_run(dag)