Migrate Google example gcs_to_sftp to new design AIP-47 (#25107)

related: #22447, #22430

Author: chenglongyan

docs/apache-airflow-providers-google/operators/transfer/gcs_to_sftp.rst

@@ -48,7 +48,7 @@ Copying a single file
 
 The following Operator copies a single file.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_gcs_to_sftp_copy_single_file]
@@ -61,7 +61,7 @@ To move the file use the ``move_object`` parameter. Once the file is copied to S
 the original file from the Google Storage is deleted. The ``destination_path`` parameter defines the
 full path of the file on the SFTP server.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_gcs_to_sftp_move_single_file_destination]
@@ -73,7 +73,7 @@ Copying a directory
 
 Use the ``wildcard`` in ``source_path`` parameter to copy a directory.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_gcs_to_sftp_copy_directory]
@@ -85,7 +85,7 @@ Moving specific files
 Use the ``wildcard`` in ``source_path`` parameter to move the specific files.
 The ``destination_path`` defines the path that is prefixed to all copied files.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_gcs_to_sftp_move_specific_files]

tests/providers/google/cloud/transfers/test_gcs_to_sftp_system.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

tests/system/providers/google/cloud/transfers/__init__.py

@@ -26,6 +26,10 @@
 from airflow.providers.google.cloud.transfers.gcs_to_sftp import GCSToSFTPOperator
 from airflow.providers.sftp.sensors.sftp import SFTPSensor
 
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+
+DAG_ID = "example_gcs_to_sftp"
+
 SFTP_CONN_ID = "ssh_default"
 BUCKET_SRC = os.environ.get("GCP_GCS_BUCKET_1_SRC", "test-gcs-sftp")
 OBJECT_SRC_1 = "parent-1.bin"
@@ -35,13 +39,12 @@
 DESTINATION_PATH_2 = "/tmp/dest-dir-1/"
 DESTINATION_PATH_3 = "/tmp/dest-dir-2/"
 
-
 with models.DAG(
-    "example_gcs_to_sftp",
+    DAG_ID,
     schedule_interval='@once',
     start_date=datetime(2021, 1, 1),
     catchup=False,
-    tags=['example'],
+    tags=['example', 'gcs'],
 ) as dag:
     # [START howto_operator_gcs_to_sftp_copy_single_file]
     copy_file_from_gcs_to_sftp = GCSToSFTPOperator(
@@ -113,9 +116,26 @@
         path=os.path.join(DESTINATION_PATH_3, OBJECT_SRC_1),
     )
 
-    move_file_from_gcs_to_sftp >> check_move_file_from_gcs_to_sftp
-    copy_dir_from_gcs_to_sftp >> check_copy_file_from_gcs_to_sftp
+    (
+        # TEST BODY
+        copy_file_from_gcs_to_sftp
+        >> check_copy_file_from_gcs_to_sftp
+        >> move_file_from_gcs_to_sftp
+        >> check_move_file_from_gcs_to_sftp
+        >> copy_dir_from_gcs_to_sftp
+        >> check_copy_dir_from_gcs_to_sftp
+        >> move_dir_from_gcs_to_sftp
+        >> check_move_dir_from_gcs_to_sftp
+    )
+
+    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
 
-    copy_dir_from_gcs_to_sftp >> move_dir_from_gcs_to_sftp
-    copy_dir_from_gcs_to_sftp >> check_copy_dir_from_gcs_to_sftp
-    move_dir_from_gcs_to_sftp >> check_move_dir_from_gcs_to_sftp
+# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
+test_run = get_test_run(dag)