[airflow] branch master updated: Use check_output to capture in celery task (#10310)

[po...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 439f7dc  Use check_output to capture in celery task (#10310)
439f7dc is described below

commit 439f7dc1d1d531405ecd6e3772bd69a6fcb9f508
Author: Ping Zhang <pi...@umich.edu>
AuthorDate: Tue Aug 18 03:55:46 2020 -0700

    Use check_output to capture in celery task (#10310)
    
    See: https://docs.python.org/3/library/subprocess.html#subprocess.CalledProcessError
    
    The check_call does not set output to the subprocess.CalledProcessError so the log.error(e.output) code is always None.
    
    By using check_ouput, when there is CalledProcessError, it will correctly log the error output
---
 airflow/executors/celery_executor.py    | 6 ++++--
 tests/executors/test_celery_executor.py | 8 ++++----
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/airflow/executors/celery_executor.py b/airflow/executors/celery_executor.py
index 81e6072..f78664a 100644
--- a/airflow/executors/celery_executor.py
+++ b/airflow/executors/celery_executor.py
@@ -75,8 +75,10 @@ def execute_command(command_to_exec: CommandType) -> None:
     log.info("Executing command in Celery: %s", command_to_exec)
     env = os.environ.copy()
     try:
-        subprocess.check_call(command_to_exec, stderr=subprocess.STDOUT,
-                              close_fds=True, env=env)
+        # pylint: disable=unexpected-keyword-arg
+        subprocess.check_output(command_to_exec, stderr=subprocess.STDOUT,
+                                close_fds=True, env=env)
+        # pylint: disable=unexpected-keyword-arg
     except subprocess.CalledProcessError as e:
         log.exception('execute_command encountered a CalledProcessError')
         log.error(e.output)
diff --git a/tests/executors/test_celery_executor.py b/tests/executors/test_celery_executor.py
index 28a3285..abda851 100644
--- a/tests/executors/test_celery_executor.py
+++ b/tests/executors/test_celery_executor.py
@@ -197,16 +197,16 @@ class TestCeleryExecutor(unittest.TestCase):
         [['airflow', 'version'], ValueError],
         [['airflow', 'tasks', 'run'], None]
     ))
-    @mock.patch('subprocess.check_call')
-    def test_command_validation(self, command, expected_exception, mock_check_call):
+    @mock.patch('subprocess.check_output')
+    def test_command_validation(self, command, expected_exception, mock_check_output):
         # Check that we validate _on the receiving_ side, not just sending side
         if expected_exception:
             with pytest.raises(expected_exception):
                 celery_executor.execute_command(command)
-            mock_check_call.assert_not_called()
+            mock_check_output.assert_not_called()
         else:
             celery_executor.execute_command(command)
-            mock_check_call.assert_called_once_with(
+            mock_check_output.assert_called_once_with(
                 command, stderr=mock.ANY, close_fds=mock.ANY, env=mock.ANY,
             )