Migrate Microsoft example DAGs to new design apache#22452 - azure

docs/apache-airflow-providers-microsoft-azure/index.rst

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

docs/apache-airflow-providers-microsoft-azure/operators/adf_run_pipeline.rst

@@ -31,15 +31,15 @@ This functionality can be disabled for an asynchronous wait -- typically with th
 
 Below is an example of using this operator to execute an Azure Data Factory pipeline.
 
-  .. exampleinclude:: /../../airflow/providers/microsoft/azure/example_dags/example_adf_run_pipeline.py
+  .. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_adf_run_pipeline.py
       :language: python
       :dedent: 0
       :start-after: [START howto_operator_adf_run_pipeline]
       :end-before: [END howto_operator_adf_run_pipeline]
 
 Here is a different example of using this operator to execute a pipeline but coupled with the :class:`~airflow.providers.microsoft.azure.sensors.data_factory.AzureDataFactoryPipelineRunSensor` to perform an asynchronous wait.
 
-    .. exampleinclude:: /../../airflow/providers/microsoft/azure/example_dags/example_adf_run_pipeline.py
+    .. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_adf_run_pipeline.py
         :language: python
         :dedent: 0
         :start-after: [START howto_operator_adf_run_pipeline_async]

docs/apache-airflow-providers-microsoft-azure/operators/adls.rst

@@ -35,7 +35,7 @@ file(s) from Azure DataLake Storage
 
 Below is an example of using this operator to delete a file from ADL.
 
-.. exampleinclude:: /../../airflow/providers/microsoft/azure/example_dags/example_adls_delete.py
+.. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_adls_delete.py
     :language: python
     :dedent: 0
     :start-after: [START howto_operator_adls_delete]

docs/apache-airflow-providers-microsoft-azure/operators/azure_blob_to_gcs.rst

@@ -54,7 +54,7 @@ To get information about jobs within a Azure Blob Storage use:
 
 Example usage:
 
-.. exampleinclude:: /../../airflow/providers/microsoft/azure/example_dags/example_azure_blob_to_gcs.py
+.. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_azure_blob_to_gcs.py
     :language: python
     :start-after: [START how_to_azure_blob_to_gcs]
     :end-before: [END how_to_azure_blob_to_gcs]

docs/apache-airflow-providers-microsoft-azure/operators/local_to_adls.rst

@@ -38,7 +38,7 @@ upload data from local filesystem to ADL.
 
 Below is an example of using this operator to upload a file to ADL.
 
-.. exampleinclude:: /../../airflow/providers/microsoft/azure/example_dags/example_local_to_adls.py
+.. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_local_to_adls.py
     :language: python
     :dedent: 0
     :start-after: [START howto_operator_local_to_adls]

docs/apache-airflow-providers-microsoft-azure/operators/sftp_to_wasb.rst

@@ -54,7 +54,7 @@ To get information about jobs within a Azure Blob Storage use:
 :class:`~airflow.providers.microsoft.azure.transfers.sftp_to_wasb.SFTPToWasbOperator`
 Example usage:
 
-.. exampleinclude:: /../../airflow/providers/microsoft/azure/example_dags/example_sftp_to_wasb.py
+.. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_sftp_to_wasb.py
     :language: python
     :dedent: 4
     :start-after: [START how_to_sftp_to_wasb]

airflow/providers/microsoft/azure/example_dags/example_adf_run_pipeline.py → tests/system/providers/microsoft/azure/example_adf_run_pipeline.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
+import os
 from datetime import datetime, timedelta
 
 from airflow.models import DAG, BaseOperator
@@ -27,8 +27,11 @@
 from airflow.providers.microsoft.azure.sensors.data_factory import AzureDataFactoryPipelineRunStatusSensor
 from airflow.utils.edgemodifier import Label
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_adf_run_pipeline"
+
 with DAG(
-    dag_id="example_adf_run_pipeline",
+    dag_id=DAG_ID,
     start_date=datetime(2021, 8, 13),
     schedule_interval="@daily",
     catchup=False,
@@ -71,3 +74,14 @@
 
     # Task dependency created via `XComArgs`:
     #   run_pipeline2 >> pipeline_run_sensor
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_adls_delete.py → tests/system/providers/microsoft/azure/example_adls_delete.py

@@ -25,9 +25,11 @@
 LOCAL_FILE_PATH = os.environ.get("LOCAL_FILE_PATH", 'localfile.txt')
 REMOTE_FILE_PATH = os.environ.get("REMOTE_LOCAL_PATH", 'remote.txt')
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_adls_delete"
 
 with models.DAG(
-    "example_adls_delete",
+    DAG_ID,
     start_date=datetime(2021, 1, 1),
     schedule_interval=None,
     tags=['example'],
@@ -43,3 +45,14 @@
     # [END howto_operator_adls_delete]
 
     upload_file >> remove_file
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_azure_blob_to_gcs.py → tests/system/providers/microsoft/azure/example_azure_blob_to_gcs.py

@@ -31,10 +31,12 @@
 GCP_BUCKET_FILE_PATH = os.environ.get("GCP_BUCKET_FILE_PATH", "file.txt")
 GCP_BUCKET_NAME = os.environ.get("GCP_BUCKET_NAME", "INVALID BUCKET NAME")
 GCP_OBJECT_NAME = os.environ.get("GCP_OBJECT_NAME", "file.txt")
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_azure_blob_to_gcs"
 
 # [START how_to_azure_blob_to_gcs]
 with DAG(
-    "example_azure_blob_to_gcs",
+    DAG_ID,
     schedule_interval=None,
     start_date=datetime(2021, 1, 1),  # Override to match your needs
     default_args={"container_name": AZURE_CONTAINER_NAME, "blob_name": BLOB_NAME},
@@ -57,3 +59,14 @@
     # [END how_to_azure_blob_to_gcs]
 
     wait_for_blob >> transfer_files_to_gcs
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_azure_container_instances.py → tests/system/providers/microsoft/azure/example_azure_container_instances.py

@@ -18,13 +18,17 @@
 """
 This is an example dag for using the AzureContainerInstancesOperator.
 """
+import os
 from datetime import datetime, timedelta
 
 from airflow import DAG
 from airflow.providers.microsoft.azure.operators.container_instances import AzureContainerInstancesOperator
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "aci_example"
+
 with DAG(
-    dag_id='aci_example',
+    dag_id=DAG_ID,
     default_args={'retries': 1},
     schedule_interval=timedelta(days=1),
     start_date=datetime(2018, 11, 1),
@@ -45,3 +49,9 @@
         cpu=1.0,
         task_id='start_container',
     )
+
+
+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/microsoft/azure/example_dags/example_azure_cosmosdb.py → tests/system/providers/microsoft/azure/example_azure_cosmosdb.py

@@ -28,15 +28,18 @@
 *Note: Make sure that connection `azure_cosmos_default` is properly set before running
 this example.*
 """
-
+import os
 from datetime import datetime
 
 from airflow import DAG
 from airflow.providers.microsoft.azure.operators.cosmos import AzureCosmosInsertDocumentOperator
 from airflow.providers.microsoft.azure.sensors.cosmos import AzureCosmosDocumentSensor
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_azure_cosmosdb_sensor"
+
 with DAG(
-    dag_id='example_azure_cosmosdb_sensor',
+    dag_id=DAG_ID,
     default_args={'database_name': 'airflow_example_db'},
     start_date=datetime(2021, 1, 1),
     catchup=False,
@@ -57,3 +60,14 @@
     )
 
     t1 >> t2
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_fileshare.py → tests/system/providers/microsoft/azure/example_fileshare.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
+import os
 from datetime import datetime
 
 from airflow.decorators import task
@@ -23,6 +23,8 @@
 
 NAME = 'myfileshare'
 DIRECTORY = "mydirectory"
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_fileshare"
 
 
 @task
@@ -44,9 +46,20 @@ def delete_fileshare():
 
 
 with DAG(
-    "example_fileshare",
+    DAG_ID,
     schedule_interval="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
 ) as dag:
     create_fileshare() >> delete_fileshare()
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_local_to_adls.py → tests/system/providers/microsoft/azure/example_local_to_adls.py

@@ -24,9 +24,11 @@
 
 LOCAL_FILE_PATH = os.environ.get("LOCAL_FILE_PATH", 'localfile.txt')
 REMOTE_FILE_PATH = os.environ.get("REMOTE_LOCAL_PATH", 'remote.txt')
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_local_to_adls"
 
 with models.DAG(
-    "example_local_to_adls",
+    DAG_ID,
     start_date=datetime(2021, 1, 1),
     catchup=False,
     schedule_interval=None,
@@ -43,3 +45,14 @@
     delete_file = ADLSDeleteOperator(task_id="remove_task", path=REMOTE_FILE_PATH, recursive=True)
 
     upload_file >> delete_file
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_local_to_wasb.py → tests/system/providers/microsoft/azure/example_local_to_wasb.py

@@ -26,9 +26,11 @@
 from airflow.providers.microsoft.azure.transfers.local_to_wasb import LocalFilesystemToWasbOperator
 
 PATH_TO_UPLOAD_FILE = os.environ.get('AZURE_PATH_TO_UPLOAD_FILE', 'example-text.txt')
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_local_to_wasb"
 
 with DAG(
-    "example_local_to_wasb",
+    DAG_ID,
     schedule_interval="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
@@ -38,3 +40,14 @@
     delete = WasbDeleteBlobOperator(task_id="delete_file")
 
     upload >> delete
+
+    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)

airflow/providers/microsoft/azure/example_dags/example_sftp_to_wasb.py → tests/system/providers/microsoft/azure/example_sftp_to_wasb.py

@@ -32,6 +32,8 @@
 SAMPLE_FILENAME = os.environ.get("SFTP_SAMPLE_FILENAME", "sftp_to_wasb_test.txt")
 FILE_COMPLETE_PATH = os.path.join(LOCAL_FILE_PATH, SAMPLE_FILENAME)
 SFTP_FILE_COMPLETE_PATH = os.path.join(SFTP_SRC_PATH, SAMPLE_FILENAME)
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_sftp_to_wasb"
 
 
 @task
@@ -41,7 +43,7 @@ def delete_sftp_file():
 
 
 with DAG(
-    "example_sftp_to_wasb",
+    DAG_ID,
     schedule_interval=None,
     catchup=False,
     start_date=datetime(2021, 1, 1),  # Override to match your needs
@@ -70,3 +72,14 @@ def delete_sftp_file():
     )
 
     transfer_files_to_sftp_step >> transfer_files_to_azure >> delete_blob_file_step >> delete_sftp_file()
+
+    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)