From a98f24ede1aa290c1705b997ad95b49e6fa74940 Mon Sep 17 00:00:00 2001 From: Chenglong Yan Date: Sun, 17 Jul 2022 03:07:42 +0800 Subject: [PATCH] Migrate Google example gcs_to_sftp to new design AIP-47 related: #22447, #22430 --- .../operators/transfer/gcs_to_sftp.rst | 8 +-- .../transfers/test_gcs_to_sftp_system.py | 63 ------------------- .../google/cloud/transfers/__init__.py | 16 +++++ .../cloud/transfers}/example_gcs_to_sftp.py | 36 ++++++++--- 4 files changed, 48 insertions(+), 75 deletions(-) delete mode 100644 tests/providers/google/cloud/transfers/test_gcs_to_sftp_system.py create mode 100644 tests/system/providers/google/cloud/transfers/__init__.py rename {airflow/providers/google/cloud/example_dags => tests/system/providers/google/cloud/transfers}/example_gcs_to_sftp.py (81%) diff --git a/docs/apache-airflow-providers-google/operators/transfer/gcs_to_sftp.rst b/docs/apache-airflow-providers-google/operators/transfer/gcs_to_sftp.rst index 06966d4efc01..37c2c8f4fb04 100644 --- a/docs/apache-airflow-providers-google/operators/transfer/gcs_to_sftp.rst +++ b/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] diff --git a/tests/providers/google/cloud/transfers/test_gcs_to_sftp_system.py b/tests/providers/google/cloud/transfers/test_gcs_to_sftp_system.py deleted file mode 100644 index 27cd3848ec73..000000000000 --- a/tests/providers/google/cloud/transfers/test_gcs_to_sftp_system.py +++ /dev/null @@ -1,63 +0,0 @@ -# -# 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. -"""System tests for Google Cloud Build operators""" -import os -from itertools import product - -import pytest - -from airflow.providers.google.cloud.example_dags.example_gcs_to_sftp import ( - BUCKET_SRC, - OBJECT_SRC_1, - OBJECT_SRC_2, -) -from tests.providers.google.cloud.utils.gcp_authenticator import GCP_GCS_KEY -from tests.test_utils.gcp_system_helpers import CLOUD_DAG_FOLDER, GoogleSystemTest, provide_gcp_context - - -@pytest.mark.credential_file(GCP_GCS_KEY) -class GcsToSftpExampleDagsSystemTest(GoogleSystemTest): - @provide_gcp_context(GCP_GCS_KEY) - def setUp(self): - super().setUp() - - # 1. Create buckets - self.create_gcs_bucket(BUCKET_SRC) - - # 2. Prepare files - for bucket_src, object_source in product( - ( - BUCKET_SRC, - f"{BUCKET_SRC}/dir-1", - f"{BUCKET_SRC}/dir-2", - ), - (OBJECT_SRC_1, OBJECT_SRC_2), - ): - source_path = f"gs://{bucket_src}/{object_source}" - self.upload_content_to_gcs( - lines=f"{os.urandom(1 * 1024 * 1024)}", bucket=source_path, filename=object_source - ) - - @provide_gcp_context(GCP_GCS_KEY) - def test_run_example_dag(self): - self.run_dag("example_gcs_to_sftp", CLOUD_DAG_FOLDER) - - @provide_gcp_context(GCP_GCS_KEY) - def tearDown(self): - self.delete_gcs_bucket(BUCKET_SRC) - super().tearDown() diff --git a/tests/system/providers/google/cloud/transfers/__init__.py b/tests/system/providers/google/cloud/transfers/__init__.py new file mode 100644 index 000000000000..13a83393a912 --- /dev/null +++ b/tests/system/providers/google/cloud/transfers/__init__.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. diff --git a/airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py b/tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.py similarity index 81% rename from airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py rename to tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.py index ff0431f0c369..c8b2d2dfafe7 100644 --- a/airflow/providers/google/cloud/example_dags/example_gcs_to_sftp.py +++ b/tests/system/providers/google/cloud/transfers/example_gcs_to_sftp.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)