SFTP to Google Cloud Storage Transfer system tests migration (AIP-47) (#26799)

* SFTP to Google Cloud Storage Transfer system tests migration (AIP-47)

Author: bkossakowska

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

@@ -46,7 +46,7 @@ Copying single files
 
 The following Operator copies a single file.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_sftp_to_gcs.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/gcs/example_sftp_to_gcs.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_sftp_to_gcs_copy_single_file]
@@ -59,7 +59,7 @@ To move the file use the ``move_object`` parameter. Once the file is copied to G
 the original file from the SFTP is deleted.
 The ``destination_path`` parameter defines the full path of the file in the bucket.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_sftp_to_gcs.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/gcs/example_sftp_to_gcs.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_sftp_to_gcs_move_single_file_destination]
@@ -71,7 +71,7 @@ Copying directory
 
 Use the ``wildcard`` in ``source_path`` parameter to copy the directory.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_sftp_to_gcs.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/gcs/example_sftp_to_gcs.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_sftp_to_gcs_copy_directory]
@@ -87,7 +87,7 @@ e.g. ``tests_sftp_hook_dir/subdir/parent-1.bin`` is copied to ``specific_files/p
 and ``tests_sftp_hook_dir/subdir/parent-2.bin`` is copied to ``specific_files/parent-2.bin`` .
 ``tests_sftp_hook_dir/subdir/parent-3.txt`` is skipped.
 
-.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_sftp_to_gcs.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/gcs/example_sftp_to_gcs.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_sftp_to_gcs_move_specific_files]

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

@@ -22,39 +22,61 @@
 
 import os
 from datetime import datetime
+from pathlib import Path
 
 from airflow import models
+from airflow.models.baseoperator import chain
+from airflow.operators.bash import BashOperator
+from airflow.providers.google.cloud.operators.gcs import GCSCreateBucketOperator, GCSDeleteBucketOperator
 from airflow.providers.google.cloud.transfers.sftp_to_gcs import SFTPToGCSOperator
+from airflow.utils.trigger_rule import TriggerRule
 
-BUCKET_SRC = os.environ.get("GCP_GCS_BUCKET_1_SRC", "test-sftp-gcs")
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT")
 
-TMP_PATH = "/tmp"
+DAG_ID = "example_sftp_to_gcs"
+BUCKET_NAME = f"bucket-{DAG_ID}-{ENV_ID}"
+
+TMP_PATH = "tmp"
 DIR = "tests_sftp_hook_dir"
 SUBDIR = "subdir"
 
 OBJECT_SRC_1 = "parent-1.bin"
 OBJECT_SRC_2 = "parent-2.bin"
-OBJECT_SRC_3 = "parent-3.txt"
+
+CURRENT_FOLDER = Path(__file__).parent
+LOCAL_PATH = str(Path(CURRENT_FOLDER) / "resources")
+
+FILE_LOCAL_PATH = str(Path(LOCAL_PATH) / TMP_PATH / DIR)
+FILE_NAME = "tmp.tar.gz"
 
 
 with models.DAG(
-    "example_sftp_to_gcs",
+    DAG_ID,
+    schedule="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
 ) as dag:
+
+    create_bucket = GCSCreateBucketOperator(task_id="create_bucket", bucket_name=BUCKET_NAME)
+
+    unzip_file = BashOperator(
+        task_id="unzip_data_file", bash_command=f"tar xvf {LOCAL_PATH}/{FILE_NAME} -C {LOCAL_PATH}"
+    )
+
     # [START howto_operator_sftp_to_gcs_copy_single_file]
     copy_file_from_sftp_to_gcs = SFTPToGCSOperator(
         task_id="file-copy-sftp-to-gcs",
-        source_path=os.path.join(TMP_PATH, DIR, OBJECT_SRC_1),
-        destination_bucket=BUCKET_SRC,
+        source_path=f"{FILE_LOCAL_PATH}/{OBJECT_SRC_1}",
+        destination_bucket=BUCKET_NAME,
     )
     # [END howto_operator_sftp_to_gcs_copy_single_file]
 
     # [START howto_operator_sftp_to_gcs_move_single_file_destination]
     move_file_from_sftp_to_gcs_destination = SFTPToGCSOperator(
         task_id="file-move-sftp-to-gcs-destination",
-        source_path=os.path.join(TMP_PATH, DIR, OBJECT_SRC_2),
-        destination_bucket=BUCKET_SRC,
+        source_path=f"{FILE_LOCAL_PATH}/{OBJECT_SRC_2}",
+        destination_bucket=BUCKET_NAME,
         destination_path="destination_dir/destination_filename.bin",
         move_object=True,
     )
@@ -63,17 +85,46 @@
     # [START howto_operator_sftp_to_gcs_copy_directory]
     copy_directory_from_sftp_to_gcs = SFTPToGCSOperator(
         task_id="dir-copy-sftp-to-gcs",
-        source_path=os.path.join(TMP_PATH, DIR, SUBDIR, "*"),
-        destination_bucket=BUCKET_SRC,
+        source_path=f"{FILE_LOCAL_PATH}/{SUBDIR}/*",
+        destination_bucket=BUCKET_NAME,
     )
     # [END howto_operator_sftp_to_gcs_copy_directory]
 
     # [START howto_operator_sftp_to_gcs_move_specific_files]
     move_specific_files_from_gcs_to_sftp = SFTPToGCSOperator(
         task_id="dir-move-specific-files-sftp-to-gcs",
-        source_path=os.path.join(TMP_PATH, DIR, SUBDIR, "*.bin"),
-        destination_bucket=BUCKET_SRC,
+        source_path=f"{FILE_LOCAL_PATH}/{SUBDIR}/*.bin",
+        destination_bucket=BUCKET_NAME,
         destination_path="specific_files/",
         move_object=True,
     )
     # [END howto_operator_sftp_to_gcs_move_specific_files]
+
+    delete_bucket = GCSDeleteBucketOperator(
+        task_id="delete_bucket", bucket_name=BUCKET_NAME, trigger_rule=TriggerRule.ALL_DONE
+    )
+
+    chain(
+        # TEST SETUP
+        create_bucket,
+        unzip_file,
+        # TEST BODY
+        copy_file_from_sftp_to_gcs,
+        move_file_from_sftp_to_gcs_destination,
+        copy_directory_from_sftp_to_gcs,
+        move_specific_files_from_gcs_to_sftp,
+        # 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)