Migrate Google ads example to new design AIP-47 (#24941)

related: #22447, #22430

Author: chenglongyan

docs/apache-airflow-providers-google/example-dags.rst

@@ -20,7 +20,7 @@ Example DAGs
 
 You can learn how to use Google integrations by analyzing the source code of the example DAGs:
 
-* `Google Ads <https://github.com/apache/airflow/tree/providers-google/8.0.0/airflow/providers/google/ads/example_dags>`__
+* `Google Ads <https://github.com/apache/airflow/tree/providers-google/8.1.0/tests/system/providers/google/ads>`__
 * `Google Cloud (legacy) <https://github.com/apache/airflow/tree/providers-google/8.0.0/airflow/providers/google/cloud/example_dags>`__
 * `Google Cloud <https://github.com/apache/airflow/tree/providers-google/8.0.0/tests/system/providers/google/cloud>`__
 * `Google Firebase <https://github.com/apache/airflow/tree/providers-google/8.1.0/tests/system/providers/google/firebase>`__

docs/apache-airflow-providers-google/operators/ads.rst

@@ -33,7 +33,7 @@ Google Ads to GCS
 To query the Google Ads API and generate a CSV report of the results use
 :class:`~airflow.providers.google.ads.transfers.ads_to_gcs.GoogleAdsToGcsOperator`.
 
-.. exampleinclude:: /../../airflow/providers/google/ads/example_dags/example_ads.py
+.. exampleinclude:: /../../tests/system/providers/google/ads/example_ads.py
     :language: python
     :dedent: 4
     :start-after: [START howto_google_ads_to_gcs_operator]
@@ -52,7 +52,7 @@ Upload Google Ads Accounts to GCS
 To upload Google Ads accounts to Google Cloud Storage bucket use the
 :class:`~airflow.providers.google.ads.transfers.ads_to_gcs.GoogleAdsListAccountsOperator`.
 
-.. exampleinclude:: /../../airflow/providers/google/ads/example_dags/example_ads.py
+.. exampleinclude:: /../../tests/system/providers/google/ads/example_ads.py
     :language: python
     :dedent: 4
     :start-after: [START howto_ads_list_accounts_operator]

airflow/providers/google/ads/example_dags/__init__.py renamed to tests/system/providers/google/ads/__init__.py

@@ -24,10 +24,17 @@
 from airflow import models
 from airflow.providers.google.ads.operators.ads import GoogleAdsListAccountsOperator
 from airflow.providers.google.ads.transfers.ads_to_gcs import GoogleAdsToGcsOperator
+from airflow.providers.google.cloud.operators.gcs import GCSCreateBucketOperator, GCSDeleteBucketOperator
+from airflow.utils.trigger_rule import TriggerRule
 
 # [START howto_google_ads_env_variables]
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT")
+
+DAG_ID = "example_google_ads"
+
+BUCKET_NAME = f"bucket_{DAG_ID}_{ENV_ID}"
 CLIENT_IDS = ["1111111111", "2222222222"]
-BUCKET = os.environ.get("GOOGLE_ADS_BUCKET", "gs://INVALID BUCKET NAME")
 GCS_OBJ_PATH = "folder_name/google-ads-api-results.csv"
 GCS_ACCOUNTS_CSV = "folder_name/accounts.csv"
 QUERY = """
@@ -61,28 +68,58 @@
     "metrics.all_conversions.value",
     "metrics.cost_micros.value",
 ]
-
 # [END howto_google_ads_env_variables]
 
 with models.DAG(
-    "example_google_ads",
-    schedule_interval=None,  # Override to match your needs
+    DAG_ID,
+    schedule_interval='@once',
     start_date=datetime(2021, 1, 1),
     catchup=False,
+    tags=["example", "ads"],
 ) as dag:
+    create_bucket = GCSCreateBucketOperator(
+        task_id="create_bucket", bucket_name=BUCKET_NAME, project_id=PROJECT_ID
+    )
+
     # [START howto_google_ads_to_gcs_operator]
     run_operator = GoogleAdsToGcsOperator(
         client_ids=CLIENT_IDS,
         query=QUERY,
         attributes=FIELDS_TO_EXTRACT,
         obj=GCS_OBJ_PATH,
-        bucket=BUCKET,
+        bucket=BUCKET_NAME,
         task_id="run_operator",
     )
     # [END howto_google_ads_to_gcs_operator]
 
     # [START howto_ads_list_accounts_operator]
     list_accounts = GoogleAdsListAccountsOperator(
-        task_id="list_accounts", bucket=BUCKET, object_name=GCS_ACCOUNTS_CSV
+        task_id="list_accounts", bucket=BUCKET_NAME, object_name=GCS_ACCOUNTS_CSV
     )
     # [END howto_ads_list_accounts_operator]
+
+    delete_bucket = GCSDeleteBucketOperator(
+        task_id="delete_bucket", bucket_name=BUCKET_NAME, trigger_rule=TriggerRule.ALL_DONE
+    )
+
+    (
+        # TEST SETUP
+        create_bucket
+        # TEST BODY
+        >> run_operator
+        >> list_accounts
+        # TEST TEARDOWN
+        >> 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()
+
+
+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)