[airflow] branch main updated: Fail ``LocalFilesystemToGCSOperator`` if src does not exist (#22772)

[ka...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kaxilnaik pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 838cf401b9 Fail ``LocalFilesystemToGCSOperator`` if src does not exist (#22772)
838cf401b9 is described below

commit 838cf401b9a424ad0fbccd5fb8d3040a8f4a7f44
Author: pankajastro <98...@users.noreply.github.com>
AuthorDate: Thu Apr 7 00:52:38 2022 +0530

    Fail ``LocalFilesystemToGCSOperator`` if src does not exist (#22772)
    
    Fix #22705.
    
    Fail LocalFilesystemToGCSOperator if the src file does not exist
    
    `src` argument of LocalFilesystemToGCSOperator accept either list of source file path or a single source file path as a string. In the case of a single source file path we are using [glob](https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/transfers/local_to_gcs.py#L111) to parse the file path and glob return empty list if file path does not exist. In the next step, we iterate on this list and call [hook api](https://github.com/apache/airflow/blob/main/airflow/ [...]
    
    Change
    - Raise an exception if [filepath](https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/transfers/local_to_gcs.py#L111) list is empty
    
    After this change below task will fail if example-text.txt does not exist
    
    ```
    upload_file = LocalFilesystemToGCSOperator(
            task_id="upload_file",
            src="example-text.txt",
            dst=DESTINATION_FILE_LOCATION,
            bucket=BUCKET_NAME,
        )
    
    ```
---
 airflow/providers/google/cloud/transfers/local_to_gcs.py    |  2 ++
 tests/providers/google/cloud/transfers/test_local_to_gcs.py | 11 +++++++++++
 2 files changed, 13 insertions(+)

diff --git a/airflow/providers/google/cloud/transfers/local_to_gcs.py b/airflow/providers/google/cloud/transfers/local_to_gcs.py
index 4f0993763a..5ed84109e7 100644
--- a/airflow/providers/google/cloud/transfers/local_to_gcs.py
+++ b/airflow/providers/google/cloud/transfers/local_to_gcs.py
@@ -109,6 +109,8 @@ class LocalFilesystemToGCSOperator(BaseOperator):
         )
 
         filepaths = self.src if isinstance(self.src, list) else glob(self.src)
+        if not filepaths:
+            raise FileNotFoundError(self.src)
         if os.path.basename(self.dst):  # path to a file
             if len(filepaths) > 1:  # multiple file upload
                 raise ValueError(
diff --git a/tests/providers/google/cloud/transfers/test_local_to_gcs.py b/tests/providers/google/cloud/transfers/test_local_to_gcs.py
index e7efb905f0..8772afdc92 100644
--- a/tests/providers/google/cloud/transfers/test_local_to_gcs.py
+++ b/tests/providers/google/cloud/transfers/test_local_to_gcs.py
@@ -81,6 +81,17 @@ class TestFileToGcsOperator(unittest.TestCase):
             object_name='test/test1.csv',
         )
 
+    def test_execute_with_empty_src(self):
+        operator = LocalFilesystemToGCSOperator(
+            task_id='local_to_sensor',
+            dag=self.dag,
+            src="no_file.txt",
+            dst='test/no_file.txt',
+            **self._config,
+        )
+        with pytest.raises(FileNotFoundError):
+            operator.execute(None)
+
     @mock.patch('airflow.providers.google.cloud.transfers.local_to_gcs.GCSHook', autospec=True)
     def test_execute_multiple(self, mock_hook):
         mock_instance = mock_hook.return_value