[GitHub] XD-DENG closed pull request #3746: [AIRFLOW-2896] Improve HdfsSensor()

[GitBox <gi...@apache.org>]

XD-DENG closed pull request #3746: [AIRFLOW-2896] Improve HdfsSensor()
URL: https://github.com/apache/incubator-airflow/pull/3746
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/airflow/sensors/hdfs_sensor.py b/airflow/sensors/hdfs_sensor.py
index 4d95556f47..4175f4cf07 100644
--- a/airflow/sensors/hdfs_sensor.py
+++ b/airflow/sensors/hdfs_sensor.py
@@ -81,19 +81,20 @@ def filter_for_ignored_ext(result, ignored_ext, ignore_copying):
         Will filter if instructed to do so the result to remove matching criteria
 
         :param result: (list) of dicts returned by Snakebite ls
-        :param ignored_ext: (list) of ignored extensions
+        :param ignored_ext: (list) of ignored extensions, like ``['exe', 'py']``
         :param ignore_copying: (bool) shall we ignore ?
         :return: (list) of dicts which were not removed
         """
         if ignore_copying:
             log = LoggingMixin().log
-            regex_builder = "^.*\.(%s$)$" % '$|'.join(ignored_ext)
+            regex_builder = "^.*\.(%s$)$" % '$|'.join([e.lower() for e in ignored_ext])
             ignored_extensions_regex = re.compile(regex_builder)
             log.debug(
                 'Filtering result for ignored extensions: %s in files %s',
                 ignored_extensions_regex.pattern, map(lambda x: x['path'], result)
             )
-            result = [x for x in result if not ignored_extensions_regex.match(x['path'])]
+            result = [x for x in result
+                      if not ignored_extensions_regex.match(x['path'].lower())]
             log.debug('HdfsSensor.poke: after ext filter result is %s', result)
         return result
 
diff --git a/tests/sensors/test_hdfs_sensor.py b/tests/sensors/test_hdfs_sensor.py
index b94065d842..13b1f3449c 100644
--- a/tests/sensors/test_hdfs_sensor.py
+++ b/tests/sensors/test_hdfs_sensor.py
@@ -89,3 +89,38 @@ def test_legacy_file_does_not_exists(self):
         # Then
         with self.assertRaises(AirflowSensorTimeout):
             task.execute(None)
+
+    def test_filter_for_ignored_ext(self):
+        """
+        Test the method HdfsSensor.filter_for_ignored_ext
+        :return:
+        """
+        sample_files = [{'path': 'x.py'}, {'path': 'x.txt'}, {'path': 'x.exe'}]
+
+        check_1 = HdfsSensor.filter_for_ignored_ext(result=sample_files,
+                                                    ignored_ext=['exe', 'py'],
+                                                    ignore_copying=True)
+        self.assertTrue(len(check_1) == 1)
+        self.assertEqual(check_1[0]['path'].rsplit(".")[-1], "txt")
+
+        check_2 = HdfsSensor.filter_for_ignored_ext(result=sample_files,
+                                                    ignored_ext=['EXE', 'PY'],
+                                                    ignore_copying=True)
+        self.assertTrue(len(check_2) == 1)
+        self.assertEqual(check_2[0]['path'].rsplit(".")[-1], "txt")
+
+    def test_filter_for_filesize(self):
+        """
+        Test the method HdfsSensor.filter_for_filesize
+        :return:
+        """
+        # unit of 'length' here is "byte"
+        sample_files = [{'path': 'small_file_1.txt', 'length': 1024},
+                        {'path': 'small_file_2.txt', 'length': 2048},
+                        {'path': 'big_file.txt', 'length': 1024 ** 2 + 1}]
+
+        # unit of argument 'size' inside HdfsSensor.filter_for_filesize is "MB"
+        check = HdfsSensor.filter_for_filesize(result=sample_files,
+                                               size=1)
+        self.assertTrue(len(check) == 1)
+        self.assertEqual(check[0]['path'], 'big_file.txt')


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services