Skip to content

Commit

Permalink
Migrate Google example gcs_to_sftp to new design AIP-47 (#25107)
Browse files Browse the repository at this point in the history
related: #22447, #22430
  • Loading branch information
chenglongyan committed Jul 18, 2022
1 parent 8aa0b24 commit c0ce8a8
Show file tree
Hide file tree
Showing 4 changed files with 48 additions and 75 deletions.
Expand Up @@ -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]
Expand All @@ -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]
Expand All @@ -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]
Expand All @@ -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]
Expand Down
63 changes: 0 additions & 63 deletions tests/providers/google/cloud/transfers/test_gcs_to_sftp_system.py

This file was deleted.

16 changes: 16 additions & 0 deletions 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.
Expand Up @@ -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"
Expand All @@ -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(
Expand Down Expand Up @@ -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)

0 comments on commit c0ce8a8

Please sign in to comment.